paulpas Security Engineering — Secure Development Lifecycle
Senior security engineer implementing the full secure development lifecycle (SDLC) across threat modeling, secure coding standards, zero-trust architecture, supply chain integrity, and automated security pipeline integration. This skill provides concrete patterns for preventing vulnerabilities at every stage of the software lifecycle — from design-time threat analysis through production security controls. Follow OWASP ASVS v4.0.3, NIST SSDF (SP 800-218), and MITRE ATT&CK as the authoritative security frameworks for all implementation decisions.
TL;DR Checklist
- Run STRIDE threat modeling on every new service or major feature before writing code
- Validate all external inputs: parameterize queries, encode outputs, enforce content types
- Implement least-privilege access controls at both infrastructure and application levels
- Generate SBOM (CycloneDX) on every build and scan with Trivy or Grype
- Enforce mTLS between all service-to-service communications in the mesh
- Integrate SAST (Semgrep, CodeQL), DAST (OWASP ZAP), and secret scanning (gitleaks) into CI
- Store all secrets in HashiCorp Vault with dynamic credentials — never in environment variables or source code
- Apply OPA/Gatekeeper policies to reject non-compliant Kubernetes manifests before deployment
When to Use
Use this skill when:
- Designing a new service or system that requires threat modeling and security controls
- Implementing secure coding patterns to prevent OWASP Top 10 vulnerabilities in application code
- Building microservices that need zero-trust networking with mTLS and identity verification
- Setting up CI/CD security pipelines with SAST, DAST, dependency scanning, and secret detection
- Creating supply chain security controls: SBOM generation, artifact signing, and provenance attestation
- Configuring policy-as-code enforcement (OPA, Kyverno) to prevent insecure infrastructure deployments
- Migrating a legacy application to least-privilege access patterns with service identities
- Conducting a security design review for a system handling sensitive data (PII, payment, health records)
When NOT to Use
Avoid this skill for:
- Routine code quality or style issues — use
coding-code-reviewinstead - Analyzing a specific CVE in a dependency — use
coding-dependency-supply-chain-securityinstead - Generating runbooks or incident response playbooks — use an operations-focused skill
- High-level security strategy without implementation specifics — narrow the scope to concrete patterns
Core Workflow
Threat Model the System — Apply STRIDE to enumerate threats per component, then prioritize with DREAD scoring. Checkpoint: Every external data flow must have at least one identified threat with a mitigation assigned.
Define Security Requirements — Translate threats into concrete security requirements using OWASP ASVS as the baseline. Separate L1 (basic), L2 (intermediate), and L3 (elevated) controls based on data sensitivity. Checkpoint: Every STRIDE threat maps to at least one ASVS control category.
Implement Secure Coding Patterns — Apply input validation, output encoding, parameterized queries, and CSRF protections in application code. Use the patterns in Section 2 below.
Enforce Zero-Trust Network Controls — Configure mTLS between services, enforce identity-aware access policies, and implement least-privilege RBAC/ABAC at every layer. Checkpoint: No service-to-service communication should ever be unauthenticated or unencrypted.
Integrate Security Automation — Add SAST, DAST, dependency scanning, secret detection, and policy-as-code gates to the CI/CD pipeline before any artifact reaches production. Checkpoint: Every merge to main triggers the full security pipeline; no manual bypass allowed.
Verify and Report — Run automated scans, review findings, track remediation metrics, and maintain an attack surface inventory. Checkpoint: Zero Critical or High findings should exist in production without documented risk acceptance.
Implementation Patterns / Reference Guide
Pattern 1: STRIDE Threat Modeling
STRIDE categorizes threats into six classes. Apply it to every component and data flow in the system design.
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
class StrideCategory(Enum):
"""STRIDE threat categories per Microsoft's security framework."""
SPOOFING = "spoofing" # Identity forgery, credential theft
TAMPERING = "tampering" # Data modification in transit or at rest
REPUDIATION = "repudiation" # Actions not logged, no non-repudiation proof
INFORMATION_DISCLOSURE = "info_disclosure" # Sensitive data exposure
DENIAL_OF_SERVICE = "dos" # Resource exhaustion, availability attacks
ELEVATION_OF_PRIVILEGE = "privilege_escalation" # Unauthorized access level increase
@dataclass
class Threat:
"""A STRIDE threat identified during threat modeling."""
component: str
data_flow: str
category: StrideCategory
description: str
severity: int = 0 # 1-10 DREAD score calculated later
mitigation: str = ""
@dataclass
class ThreatModel:
"""Container for a STRIDE threat model of a system component."""
system_name: str
version: str
threats: list[Threat] = field(default_factory=list)
attack_surface: list[str] = field(default_factory=list)
def add_threat(
self,
component: str,
data_flow: str,
category: StrideCategory,
description: str,
mitigation: str = ""
) -> Threat:
"""Record a threat with its STRIDE category and proposed mitigation."""
threat = Threat(
component=component,
data_flow=data_flow,
category=category,
description=description,
mitigation=mitigation
)
self.threats.append(threat)
return threat
def threats_by_component(self, component: str) -> list[Threat]:
"""Return all threats affecting a specific component."""
return [t for t in self.threats if t.component == component]
DREAD Risk Scoring (Post-Threat-Model Prioritization)
After identifying threats with STRIDE, score each using the DREAD model to prioritize remediation:
from enum import Enum
class DataSensitivity(Enum):
PUBLIC = 1
INTERNAL = 2
CONFIDENTIAL = 3
RESTRICTED = 4
def calculate_dread_score(
difficulty: int, # How easy is exploitation? (1=easy, 10=hard)
exploitability: int, # Likelihood of attack succeeding (1-10)
discovery: int, # How obvious is the vulnerability? (1=hidden, 10=obvious)
affected_users: int, # Proportion of users impacted (1-10)
damage_potential: int, # Severity of impact if exploited (1-10)
reproducibility: int, # How reliably can the attack be repeated? (1-10)
data_sensitivity: DataSensitivity = DataSensitivity.INTERNAL
) -> dict:
"""Calculate DREAD risk score with weighted factors.
Returns a dictionary with total score, component scores, and severity rating.
Scores are normalized to 0-10 range.
"""
weights = {
"difficulty": 1.0,
"exploitability": 2.0, # Highest weight — focus on what's easily attackable
"discovery": 1.0,
"affected_users": 1.5,
"damage_potential": 2.0, # Highest weight — prioritize high-impact issues
"reproducibility": 1.0,
}
damage_multiplier = data_sensitivity.value # Scale up for sensitive data
component_scores = {}
weighted_total = 0.0
for factor, weight in weights.items():
score = min(10, max(1, locals()[factor]))
weighted_score = (score / 10) * weight
component_scores[factor] = round(weighted_score, 2)
weighted_total += weighted_score
max_possible_weight = sum(weights.values())
normalized_score = (weighted_total / max_possible_weight) * damage_multiplier
normalized_score = min(10, normalized_score)
if normalized_score >= 8:
severity = "CRITICAL"
elif normalized_score >= 6:
severity = "HIGH"
elif normalized_score >= 4:
severity = "MEDIUM"
else:
severity = "LOW"
return {
"total_score": round(normalized_score, 2),
"component_scores": component_scores,
"severity": severity,
"data_sensitivity_multiplier": damage_multiplier,
}
Pattern 2: OWASP Top 10 Prevention Code
These patterns prevent the most common web application vulnerabilities. Each example shows the BAD approach and a secure alternative with typed signatures.
A01:2021 — Broken Access Control (IDOR Prevention)
from functools import wraps
from typing import Callable, Any
import logging
logger = logging.getLogger(__name__)
def require_ownership(
resource_id_field: str = "resource_id",
user_id_field: str = "user.id"
) -> Callable:
"""Decorator that enforces object-level authorization (prevents IDOR attacks).
Verifies the authenticated user has ownership or explicit permission
to access the requested resource before allowing the handler to execute.
Args:
resource_id_field: Query parameter or path variable name containing the resource ID.
user_id_field: Dot-notation path to extract user ID from request context.
Usage:
@require_ownership("order_id")
def get_order(request): ...
"""
def decorator(fn: Callable) -> Callable:
@wraps(fn)
def wrapper(*args: Any, **kwargs: Any) -> Any:
request = kwargs.get("request") or args[0]
# Extract resource ID from the request
resource_id = kwargs.get(resource_id_field)
if not resource_id:
raise ValueError(f"Missing required field: {resource_id_field}")
# Extract authenticated user context
user_context = request.get("user", {})
user_id = _resolve_dot_notation(user_context, user_id_field)
if not user_id:
logger.warning("Access attempt with unauthenticated user for resource %s", resource_id)
raise PermissionError("Authentication required")
# Check ownership — this is the critical IDOR prevention step
resource = _fetch_resource(resource_id) # Your data access layer
if resource and str(resource.get("owner_id")) != str(user_id):
logger.warning(
"IDOR attempt: user %s tried to access resource owned by %s",
user_id,
resource.get("owner_id")
)
raise PermissionError("Access denied: insufficient permissions")
return fn(*args, **kwargs)
return wrapper
return decorator
def _resolve_dot_notation(obj: dict, path: str) -> Any:
"""Safely resolve dot-notation path in a nested dictionary."""
keys = path.split(".")
current = obj
for key in keys:
if isinstance(current, dict):
current = current.get(key)
else:
return None
if current is None:
return None
return current
def _fetch_resource(resource_id: str) -> dict | None:
"""Placeholder — replace with your actual data access layer."""
pass # Implementation depends on your ORM / database layer
A03:2021 — Injection (SQL + Command Injection Prevention)
❌ BAD: SQL Injection via String Formatting
# DANGEROUS: User input directly interpolated into SQL query string
def query_users_bad(db_path, role):
conn = sqlite3.connect(db_path)
# Never do this — user controls the query string entirely
query = f"SELECT * FROM users WHERE role = '{role}'"
return conn.execute(query).fetchall() # Attacker passes: ' OR '1'='1
✅ GOOD: Parameterized Query (Secure)
import sqlite3
from typing import List
def query_users_by_role(db_path: str, role: str) -> List[tuple]:
"""Secure parameterized query preventing SQL injection.
NEVER use f-strings or format() to build SQL queries with user input.
Always use parameterized queries with the database driver's built-in
binding mechanism.
Args:
db_path: Path to the SQLite database file.
role: The user role to filter by (e.g., 'admin', 'editor').
Returns:
List of matching user tuples (id, username, email, role).
Raises:
sqlite3.Error: If database access fails.
"""
if not isinstance(role, str) or len(role) > 50:
raise ValueError("Role must be a string of 50 characters or less")
# Parameterized query — the database driver handles escaping automatically
conn = sqlite3.connect(db_path)
try:
cursor = conn.execute(
"SELECT id, username, email, role FROM users WHERE role = ?",
(role,)
)
return cursor.fetchall()
finally:
conn.close()
❌ BAD: Command Injection via shell=True
import subprocess
# DANGEROUS: User-controlled string passed directly to shell
def find_files_bad(base_directory, pattern):
# Attacker passes pattern="; rm -rf / ; echo " — total system compromise
return subprocess.run(
f"find {base_directory} -name '{pattern}'",
shell=True # Never use with user input!
)
✅ GOOD: Safe File Search with pathlib (Secure)
import subprocess
from pathlib import Path
from typing import List
def find_files_safely(base_directory: str, pattern: str) -> List[str]:
"""Secure file search preventing command injection.
NEVER pass user input directly to os.system(), subprocess with shell=True,
or eval(). Use pathlib for safe path operations and subprocess with
explicit argument lists (no shell).
Args:
base_directory: The root directory to search within.
pattern: Glob pattern to match filenames against.
Returns:
List of absolute file paths matching the criteria.
Raises:
ValueError: If inputs contain path traversal sequences or are invalid.
"""
# Prevent directory traversal attacks
base_path = Path(base_directory).resolve()
if not base_path.is_dir():
raise ValueError(f"Invalid directory: {base_directory}")
# Validate pattern contains no path separators or shell metacharacters
if "/" in pattern or "\\" in pattern or any(c in pattern for c in "*;|&$`"):
raise ValueError("Pattern must not contain path separators or shell metacharacters")
# Use pathlib's built-in glob — safe, cross-platform, no shell involved
matches = []
for file_path in base_path.glob(pattern):
if file_path.is_file() and str(file_path.resolve()).startswith(str(base_path)):
matches.append(str(file_path))
return sorted(matches)
A07:2021 — Identification and Authentication Failures (Secure Session Handling)
❌ BAD: Insecure Session with Predictable Tokens and No Expiration
# DANGEROUS: Uses predictable tokens, no expiration, stored client-side
import uuid
import json
def create_session_bad(user_id: str) -> dict:
# UUID is not cryptographically random — attackers can predict session IDs
token = str(uuid.uuid4()) # Predictable on some implementations
# Storing sensitive data in the token itself breaks confidentiality
payload = {
"user_id": user_id,
"role": "admin", # NEVER embed roles in tokens without signing
"exp": None, # No expiration — sessions live forever
}
return {"token": token, "payload": json.dumps(payload)} # Leaks payload to client
✅ GOOD: Cryptographically Secure Session Manager (Secure)
import secrets
import time
from typing import Optional
from dataclasses import dataclass, field
@dataclass
class SessionConfig:
"""Configuration for secure session management."""
token_length: int = 64 # Bytes of entropy in session tokens
expiration_seconds: int = 3600 # 1 hour default TTL
max_concurrent_sessions: int = 5 # Limit per user
rotate_on_privilege_change: bool = True # New token on login/logout
secure_cookie: bool = True # HttpOnly + Secure + SameSite
class SecureSessionManager:
"""Implements secure session management with cryptographic tokens.
Follows OWASP Session Management Cheat Sheet:
- Generates unpredictable 256-bit+ tokens using secrets.token_bytes()
- Tokens are stored server-side (not in cookies) with expiration
- Sessions rotate on authentication state changes
- Rate limits login attempts to prevent brute force attacks
"""
def __init__(self, config: Optional[SessionConfig] = None):
self.config = config or SessionConfig()
self._sessions: dict[str, dict] = {}
self._failed_attempts: dict[str, int] = {} # IP -> count
self._attempt_timestamps: dict[str, list[float]] = {}
def create_session(self, user_id: str) -> str:
"""Create a new secure session token for an authenticated user.
Generates a cryptographically random token and stores session metadata
server-side. The token is returned — never embed sensitive data in it.
Args:
user_id: Unique identifier of the authenticated user.
Returns:
Cryptographically secure session token string (hex-encoded).
"""
# Generate 64 bytes (512 bits) of cryptographic randomness
token = secrets.token_hex(self.config.token_length)
now = time.time()
self._sessions[token] = {
"user_id": user_id,
"created_at": now,
"expires_at": now + self.config.expiration_seconds,
"ip_address": None, # Populate at first request
"user_agent": None,
}
return token
def validate_session(self, token: str) -> Optional[dict]:
"""Validate a session token and return user context if valid.
Checks expiration, revocation, and consistency. Returns None for
invalid or expired tokens.
Args:
token: The session token to validate.
Returns:
Session metadata dict if valid, None otherwise.
"""
session = self._sessions.get(token)
if not session:
return None
# Check expiration
if time.time() > session["expires_at"]:
del self._sessions[token]
return None
return session
def rotate_token(self, old_token: str, user_id: str) -> str:
"""Rotate a session token (called on login/logout/role change).
Invalidates the old token and creates a new one to prevent fixation attacks.
"""
# Invalidate old token first (never create new before invalidating old)
if old_token in self._sessions:
del self._sessions[old_token]
return self.create_session(user_id)
def record_failed_attempt(self, identifier: str) -> None:
"""Track failed login attempts for brute force detection."""
now = time.time()
# Clean attempts older than 15 minutes
window_start = now - 900
if identifier in self._attempt_timestamps:
self._attempt_timestamps[identifier] = [
t for t in self._attempt_timestamps[identifier] if t > window_start
]
else:
self._attempt_timestamps[identifier] = []
self._attempt_timestamps[identifier].append(now)
self._failed_attempts[identifier] = len(self._attempt_timestamps[identifier])
def is_rate_limited(self, identifier: str, max_attempts: int = 5) -> bool:
"""Check if an identifier is rate-limited due to too many failed attempts."""
count = self._failed_attempts.get(identifier, 0)
return count >= max_attempts
def clear_failed_attempts(self, identifier: str) -> None:
"""Clear failed attempt counters after successful authentication."""
self._failed_attempts.pop(identifier, None)
self._attempt_timestamps.pop(identifier, None)
Pattern 3: Zero-Trust Architecture — mTLS Configuration
Zero-trust networking requires mutual authentication for every service-to-service connection. This pattern shows production-ready mTLS using OpenSSL and Python's ssl module.
TLS Certificate Authority Setup Script
#!/usr/bin/env bash
# Set strict error handling for security-critical operations
set -euo pipefail
# ============================================================
# Zero-Trust: Generate CA + Service Certificates for mTLS
# ============================================================
# This script creates a minimal PKI hierarchy suitable for
# internal service mesh mutual TLS. For production, use
# cert-manager with Kubernetes or HashiCorp Vault PKI engine.
# ============================================================
readonly CERT_DIR="${1:-./tls-certs}"
readonly CA_CN="Internal Root CA"
readonly EXPIRY_YEARS=1
mkdir -p "${CERT_DIR}/ca"
mkdir -p "${CERT_DIR}/services"
generate_ca() {
local key_file="${CERT_DIR}/ca/ca.key"
local cert_file="${CERT_DIR}/ca/ca.crt"
echo "Generating Root CA..."
openssl genrsa -out "${key_file}" 4096 2>/dev/null
chmod 600 "${key_file}"
openssl req -x509 -new -nodes \
-key "${key_file}" \
-sha256 \
-days $((EXPIRY_YEARS * 365)) \
-out "${cert_file}" \
-subj "/C=US/ST=California/O=MyOrg/CN=${CA_CN}" \
-addext "basicConstraints=critical,CA:TRUE" \
-addext "keyUsage=critical,keyCertSign,cRLSign" 2>/dev/null
echo "Root CA created at ${cert_file}"
}
generate_service_cert() {
local service_name="$1"
local sans="$2" # Comma-separated SANs: dns:api.internal,dns:cache.internal,IP:10.0.0.5
local key_file="${CERT_DIR}/services/${service_name}.key"
local csr_file="${CERT_DIR}/services/${service_name}.csr"
local cert_file="${CERT_DIR}/services/${service_name}.crt"
echo "Generating certificate for ${service_name}..."
# Generate service key
openssl genrsa -out "${key_file}" 2048 2>/dev/null
chmod 600 "${key_file}"
# Build SAN configuration from comma-separated list
local san_config=""
IFS=',' read -ra parts <<< "${sans}"
for part in "${parts[@]}"; do
san_config+="extendedKeyUsage = serverAuth, clientAuth"$'\n'
case "$part" in
dns:*) san_config+="subjectAltName=DNS:${part#dns:}"$'\n' ;;
IP:*) san_config+="subjectAltName=IP:${part#IP:}"$'\n' ;;
esac
done
# Generate CSR
openssl req -new \
-key "${key_file}" \
-out "${csr_file}" \
-subj "/C=US/ST=California/O=MyOrg/CN=${service_name}.internal" 2>/dev/null
# Sign with CA
openssl x509 -req \
-in "${csr_file}" \
-CA "${CERT_DIR}/ca/ca.crt" \
-CAkey "${CERT_DIR}/ca/ca.key" \
-CAcreateserial \
-out "${cert_file}" \
-days $((EXPIRY_YEARS * 365)) \
-sha256 \
-extfile <(printf "%s" "basicConstraints=CA:FALSE"$'\n'"keyUsage=digitalSignature,keyEncipherment"$'\n'"extendedKeyUsage=serverAuth,clientAuth"$'\n'"${san_config}") 2>/dev/null
echo "Certificate for ${service_name} created at ${cert_file}"
rm -f "${csr_file}" # Clean up CSR — never ship CSRs to production
}
# Execute the PKI setup
generate_ca
generate_service_cert "api-gateway" "dns:api.internal,dns:api.internal.example.com,IP:10.0.1.5"
generate_service_cert "user-service" "dns:user.internal"
echo ""
echo "PKI generation complete."
echo "Copy ca.crt to all services as the trusted CA bundle."
Python mTLS Client with Certificate Verification
import ssl
import socket
from typing import Optional
from pathlib import Path
def create_mtls_context(
cert_file: str | Path,
key_file: str | Path,
ca_file: str | Path,
verify_mode: bool = True,
min_protocol_version: str = "TLSv1_3"
) -> ssl.SSLContext:
"""Create an SSL context configured for mutual TLS authentication.
This context requires both the client and server to present certificates,
enabling zero-trust service-to-service communication. The CA bundle is used
to verify the peer's certificate chain.
Args:
cert_file: Path to the client certificate file (PEM format).
key_file: Path to the client private key file (PEM format).
ca_file: Path to the CA certificate bundle for verifying the server.
verify_mode: If True, enforce full certificate chain verification.
Set False only for development — never in production.
min_protocol_version: Minimum TLS version. Must be TLSv1_3 in production.
Returns:
Configured SSLContext ready for mTLS connections.
Raises:
FileNotFoundError: If any certificate or key file does not exist.
ValueError: If the minimum protocol version is insecure (below TLS 1.2).
"""
cert_path = Path(cert_file)
key_path = Path(key_file)
ca_path = Path(ca_file)
# Validate all files exist before creating context
for path in (cert_path, key_path, ca_path):
if not path.exists():
raise FileNotFoundError(f"TLS certificate file not found: {path}")
# Enforce minimum TLS version — never allow TLS 1.0 or 1.1
protocol_map = {
"TLSv1_2": ssl.TLSVersion.TLSv1_2,
"TLSv1_3": ssl.TLSVersion.TLSv1_3,
}
min_version = protocol_map.get(min_protocol_version)
if not min_version:
raise ValueError(f"Unknown protocol version: {min_protocol_version}")
# Create context with modern security settings
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.minimum_version = min_version
context.maximum_version = ssl.TLSVersion.TLSv1_3
# Require certificate verification in production
if verify_mode:
context.verify_mode = ssl.CERT_REQUIRED
context.check_hostname = True
else:
context.verify_mode = ssl.CERT_OPTIONAL
# Load client certificate and key for mTLS authentication
context.load_cert_chain(certfile=str(cert_path), keyfile=str(key_path))
# Trust only the specified CA — not system defaults
context.load_verify_locations(str(ca_path))
# Enable OCSP stapling if supported
try:
context.minimum_version = min_version
except ValueError:
pass # Fallback to default if protocol not available in this Python build
# Security hardening: disable weak cipher suites
context.set_ciphers(
"ECDHE-ECDSA-AES256-GCM-SHA384:"
"ECDHE-RSA-AES256-GCM-SHA384:"
"ECDHE-ECDSA-CHACHA20-POLY1305:"
"ECDHE-RSA-CHACHA20-POLY1305:"
"ECDHE-ECDSA-AES128-GCM-SHA256:"
"ECDHE-RSA-AES128-GCM-SHA256"
)
return context
def connect_mtls(
host: str,
port: int,
cert_file: str,
key_file: str,
ca_file: str,
timeout: float = 10.0
) -> ssl.SSLSocket:
"""Establish a mutual TLS connection to a service.
Performs full certificate chain verification and returns
an authenticated SSL socket for bidirectional communication.
Args:
host: Target hostname or IP address.
port: Target TCP port (typically 443 or custom mTLS port).
cert_file: Client certificate path.
key_file: Client private key path.
ca_file: Trusted CA bundle path.
timeout: Connection timeout in seconds.
Returns:
Verified SSL socket connected to the target service.
Raises:
ssl.SSLCertVerificationError: If peer certificate verification fails.
socket.timeout: If connection times out.
"""
context = create_mtls_context(cert_file, key_file, ca_file)
with socket.create_connection((host, port), timeout=timeout) as sock:
with context.wrap_socket(sock, server_hostname=host) as ssock:
# Log certificate details for audit trail
cert = ssock.getpeercert(binary_form=True)
if cert:
subject = _extract_subject_from_cert(cert)
print(f"Connected to {host}:{port} — verified identity: {subject}")
return ssock
def _extract_subject_from_cert(cert_binary: bytes) -> str:
"""Extract the subject CN from a DER-encoded certificate.
Uses openssl CLI as a fallback since pure Python cert parsing
is verbose. In production, use cryptography library for direct parsing.
"""
import subprocess
result = subprocess.run(
["openssl", "x509", "-inform", "DER", "-noout", "-subject"],
input=cert_binary, capture_output=True
)
if result.returncode == 0:
return result.stdout.decode().strip()
return "<unknown>"
Pattern 4: Policy-as-Code with Open Policy Agent (OPA) / Rego
Enforce security policies declaratively. These Rego policies prevent insecure Kubernetes deployments and enforce security standards.
OPA Rego Policy — Enforce Container Security Standards
# Policy: containers must not run as root, must have resource limits,
# must use read-only filesystems, and must drop ALL capabilities
package kubernetes.admission.security.containers
# Reject pods running as root user or group
deny[msg] {
input.request.kind.kind == "Pod"
container := input.request.object.spec.containers[_]
container.securityContext.runAsUser == 0
msg := sprintf("Container '%v' must not run as root (runAsUser=0)", [container.name])
}
# Reject pods with privileged containers
deny[msg] {
input.request.kind.kind == "Pod"
container := input.request.object.spec.containers[_]
container.securityContext.privileged == true
msg := sprintf("Privileged containers are not allowed: '%v'", [container.name])
}
# Enforce resource limits on all containers
deny[msg] {
input.request.kind.kind == "Pod"
container := input.request.object.spec.containers[_]
not container.resources.limits.cpu
msg := sprintf("Container '%v' must have CPU resource limits defined", [container.name])
}
deny[msg] {
input.request.kind.kind == "Pod"
container := input.request.object.spec.containers[_]
not container.resources.limits.memory
msg := sprintf("Container '%v' must have memory resource limits defined", [container.name])
}
# Require read-only root filesystem
deny[msg] {
input.request.kind.kind == "Pod"
container := input.request.object.spec.containers[_]
not container.securityContext.readOnlyRootFilesystem
msg := sprintf("Container '%v' must enable readOnlyRootFilesystem", [container.name])
}
# Require dropping ALL capabilities and adding only NET_BIND_SERVICE
deny[msg] {
input.request.kind.kind == "Pod"
container := input.request.object.spec.containers[_]
caps := container.securityContext.capabilities.drop
not caps[_] == "ALL"
msg := sprintf("Container '%v' must drop ALL capabilities", [container.name])
}
# Enforce image digest pinning (not latest tag or mutable tag)
deny[msg] {
input.request.kind.kind == "Pod"
container := input.request.object.spec.containers[_]
image := container.image
# Reject images using :latest or no tag at all
endswith(image, ":latest")
msg := sprintf("Image '%v' must use a digest pin (@sha256:...) instead of :latest", [image])
}
deny[msg] {
input.request.kind.kind == "Pod"
container := input.request.object.spec.containers[_]
image := container.image
# Images with mutable tags (no @digest) are rejected
not contains(image, "@sha256:")
msg := sprintf("Image '%v' must use digest-pinned format (@sha256:...) for reproducible deployments", [image])
}
# Allow the admission if no violations found
allow { true }
OPA Rego Policy — Secret Detection in Git Operations
# Policy: Prevent committing secrets, API keys, and credentials to version control.
# This policy can be integrated with pre-commit hooks or CI gates via opa eval.
package security.secrets
# List of secret patterns — extend based on your organization's risk profile
secret_patterns := {
"aws_access_key": `AKIA[0-9A-Z]{16}`,
"private_key": `-+BEGIN (RSA |EC |OPENSSH )?PRIVATE KEY-+`,
"generic_secret": `[Aa][Pp][Ii]_?[Kk][Ee][Yy].*[A-Fa-f0-9]{20,}`,
"jwt_token": `eyJ[A-Za-z0-9_-]{10,}\.eyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}`,
"database_url": `(postgres|mysql|mongodb)://[^\s]+:[^\s]+@[^\s]+`,
}
# Deny if any tracked file contains a secret pattern
deny[msg] {
input.review ~= "add" # File is being added or modified
content := input.content # Content of the changed file
name := key(secret_patterns, k) # Iterate over pattern names
regex.match(secret_patterns[name], content)
msg := sprintf("Secret pattern '%v' detected in file: %s", [k, input.path])
}
# Deny if any diff contains a secret (for CI pipeline scanning)
deny[msg] {
input.type == "git.diffs"
change := input.changes[_]
name := key(secret_patterns, k)
regex.match(secret_patterns[name], change.new_content)
msg := sprintf("Secret pattern '%v' found in %s:%d", [k, change.file, change.line])
}
allow { true }
Pattern 5: CI/CD Security Automation Pipeline
This bash script implements a multi-stage security pipeline integrating SAST, DAST, dependency scanning, secret detection, and SBOM generation.
#!/usr/bin/env bash
# ============================================================
# CI/CD Security Pipeline — Multi-Stage Security Gates
# ============================================================
# Integrates: Semgrep (SAST), Grype (dependency vulns),
# Trivy (container scan), gitleaks (secrets), CycloneDX (SBOM)
# ============================================================
set -euo pipefail
readonly BUILD_DIR="${1:-.}"
readonly REPORT_DIR="./security-reports"
readonly FAIL_ON="critical" # Fail pipeline on: low|medium|high|critical
mkdir -p "${REPORT_DIR}"
log() { echo "[$(date +%FT%T)] $*"; }
run_sast() {
log "Running SAST scan with Semgrep..."
semgrep \
--config auto \
--json \
--output "${REPORT_DIR}/sast-results.json" \
--severity ERROR,WARNING \
--metrics off \
"${BUILD_DIR}" 2>/dev/null || true
# Check for critical/high findings
local high_count
high_count=$(python3 -c "
import json, sys
with open('${REPORT_DIR}/sast-results.json') as f:
data = json.load(f)
print(sum(1 for r in data.get('results', []) if r.get('extra',{}).get('message','').lower() in ['critical','high']))
" 2>/dev/null || echo "0")
if [[ "${high_count}" -gt 0 ]]; then
log "FAIL: ${high_count} critical/high SAST findings detected"
return 1
fi
log "PASS: SAST scan complete — no critical/high findings"
return 0
}
run_dep_scanning() {
log "Running dependency vulnerability scan with Grype..."
# Scan the lockfile (adjust for your language ecosystem)
if [[ -f "${BUILD_DIR}/requirements.txt" ]]; then
grype sbom "cyclonedx-sbom.json" --fail-at "${FAIL_ON}" \
--output json > "${REPORT_DIR}/dep-scans.json" 2>/dev/null || {
log "FAIL: Vulnerabilities found at or above ${FAIL_ON} severity"
return 1
}
elif [[ -f "${BUILD_DIR}/package-lock.json" ]]; then
grype dir:"${BUILD_DIR}" --fail-at "${FAIL_ON}" \
--output json > "${REPORT_DIR}/dep-scans.json" 2>/dev/null || {
log "FAIL: Vulnerabilities found at or above ${FAIL_ON} severity"
return 1
}
fi
log "PASS: Dependency scanning complete"
return 0
}
run_secret_detection() {
log "Running secret detection with gitleaks..."
# Scan the working directory for leaked secrets
gitleaks detect \
--source "${BUILD_DIR}" \
--report-path "${REPORT_DIR}/secrets-results.json" \
--report-format json \
--log-level warn 2>/dev/null || true
local secret_count
secret_count=$(python3 -c "
import json
with open('${REPORT_DIR}/secrets-results.json') as f:
data = json.load(f)
print(len(data.get('violations', [])))
" 2>/dev/null || echo "0")
if [[ "${secret_count}" -gt 0 ]]; then
log "FAIL: ${secret_count} potential secrets detected"
return 1
fi
log "PASS: Secret detection complete — no leaks found"
return 0
}
generate_sbom() {
log "Generating CycloneDX SBOM..."
# Use cyclonedx-npm, cyclonedx-py, or similar per ecosystem
if [[ -f "${BUILD_DIR}/package.json" ]]; then
cyclonedx-npm --output-file "${REPORT_DIR}/sbom-cyclonedx.json" \
--output-format json 2>/dev/null || log "WARN: SBOM generation skipped (no cyclonedx-npm)"
elif [[ -f "${BUILD_DIR}/requirements.txt" ]]; then
cyclonedx-py requirements "${BUILD_DIR}/requirements.txt" \
--output "${REPORT_DIR}/sbom-cyclonedx.json" 2>/dev/null || log "WARN: SBOM generation skipped (no cyclonedx-py)"
fi
if [[ -f "${REPORT_DIR}/sbom-cyclonedx.json" ]]; then
log "PASS: SBOM generated at ${REPORT_DIR}/sbom-cyclonedx.json"
else
log "WARN: SBOM generation not available for this project type"
fi
return 0
}
run_sca_policy_check() {
log "Running policy-as-code checks with OPA..."
# Run Rego policies against the current state
if [[ -f "policy/kubernetes/rego/security.rego" ]]; then
opa eval \
--data "policy/kubernetes/rego/" \
--input <(echo '{"request":{"kind":{"kind":"Pod"},"object":{}}}') \
"data.kubernetes.admission.security.containers.deny" 2>/dev/null || true
fi
log "PASS: Policy check complete"
return 0
}
# ============================================================
# Pipeline Execution
# ============================================================
log "========================================="
log "Starting Security Pipeline"
log "========================================="
FAILED_STAGES=()
run_sast || FAILED_STAGES+=("SAST")
run_dep_scanning || FAILED_STAGES+=("Dependency Scan")
run_secret_detection || FAILED_STAGES+=("Secret Detection")
generate_sbom # SBOM is advisory — does not block pipeline
run_sca_policy_check || FAILED_STAGES+=("Policy Check")
if [[ ${#FAILED_STAGES[@]} -gt 0 ]]; then
log "========================================="
log "SECURITY PIPELINE FAILED"
log "Failed stages: ${FAILED_STAGES[*]}"
log "Review reports in: ${REPORT_DIR}/"
log "========================================="
exit 1
fi
log "========================================="
log "SECURITY PIPELINE PASSED — All gates clear"
log "Reports available at: ${REPORT_DIR}/"
log "========================================="
exit 0
Pattern 6: HashiCorp Vault Integration for Secret Management
Application-level secret retrieval using the HashiCorp Vault API with automatic token renewal.
import time
import logging
from typing import Optional
from dataclasses import dataclass, field
try:
import hvac
except ImportError:
raise ImportError(
"Install vault client: pip install hvac"
)
logger = logging.getLogger(__name__)
@dataclass
class VaultConfig:
"""Configuration for HashiCorp Vault integration."""
address: str = "https://vault.example.com:8200"
namespace: Optional[str] = None
role_id: Optional[str] = None # AppRole authentication
secret_id: Optional[str] = None # AppRole authentication (or use k8s auth)
k8s_role: Optional[str] = None # Kubernetes service account authentication
k8s_token_path: str = "/var/run/secrets/kubernetes.io/serviceaccount/token"
mount_point: str = "secret" # KV secrets e
…(truncated)