# Access control decorator in Flask
from functools import wraps
from flask import abort, g

def require_role(role):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if not g.current_user or g.current_user.role != role:
                abort(403)
            return f(*args, **kwargs)
        return decorated_function
    return decorator

@app.route('/admin/users')
@require_role('admin')
def admin_users():
    return get_all_users()

# Never do this - vulnerable to SQL Injection
query = "SELECT * FROM users WHERE id = " + user_input

# Use parameterized queries
cursor.execute("SELECT * FROM users WHERE id = %s", (user_input,))

# Use an ORM (SQLAlchemy)
user = User.query.filter_by(id=user_input).first()

# npm security audit
npm audit

# Python dependency audit
pip-audit

# Go vulnerability check
govulncheck ./...

import logging
import json
from datetime import datetime, timezone

security_logger = logging.getLogger('security')

def log_security_event(event_type, user_id, details):
    event = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "event_type": event_type,
        "user_id": user_id,
        "ip_address": request.remote_addr,
        "user_agent": request.headers.get('User-Agent'),
        "details": details
    }
    security_logger.warning(json.dumps(event))

# Usage
log_security_event("LOGIN_FAILED", "user123", "Invalid password attempt 3")
log_security_event("PRIVILEGE_ESCALATION", "user456", "Attempted admin access")

# Argon2 - currently the most recommended hashing algorithm
from argon2 import PasswordHasher

ph = PasswordHasher(
    time_cost=3,       # iterations
    memory_cost=65536,  # 64MB memory
    parallelism=4,      # parallel threads
    hash_len=32,        # hash length
    salt_len=16         # salt length
)

# Hashing
hashed = ph.hash("user_password")
# Example output: $argon2id$v=19$m=65536,t=3,p=4$...

# Verification
try:
    ph.verify(hashed, "user_password")
    print("Password matches")
except Exception:
    print("Password does not match")

# bcrypt - a widely used alternative
import bcrypt

password = b"user_password"
salt = bcrypt.gensalt(rounds=12)
hashed = bcrypt.hashpw(password, salt)

# Verification
if bcrypt.checkpw(password, hashed):
    print("Password matches")

# TOTP (Time-based One-Time Password) implementation
import pyotp

# Generate secret key
secret = pyotp.random_base32()

# Generate QR code URI
totp_uri = pyotp.totp.TOTP(secret).provisioning_uri(
    name="user@example.com",
    issuer_name="MyApp"
)

# Verification
totp = pyotp.TOTP(secret)
is_valid = totp.verify("123456")  # user-entered code

// Passkey registration (browser side)
const credential = await navigator.credentials.create({
  publicKey: {
    challenge: serverChallenge,
    rp: { name: "MyApp", id: "myapp.com" },
    user: {
      id: userId,
      name: "user@example.com",
      displayName: "User"
    },
    pubKeyCredParams: [
      { alg: -7, type: "public-key" },   // ES256
      { alg: -257, type: "public-key" }  // RS256
    ],
    authenticatorSelection: {
      authenticatorAttachment: "platform",
      residentKey: "required",
      userVerification: "required"
    }
  }
});

# RBAC implementation example
PERMISSIONS = {
    "admin": ["read", "write", "delete", "manage_users"],
    "editor": ["read", "write"],
    "viewer": ["read"],
}

def check_permission(user_role, required_permission):
    role_permissions = PERMISSIONS.get(user_role, [])
    return required_permission in role_permissions

# Usage
if check_permission(current_user.role, "delete"):
    delete_resource(resource_id)
else:
    raise PermissionError("You do not have delete permissions")

# ABAC policy example
def evaluate_policy(user, resource, action, environment):
    # Allow access only during business hours
    if environment["hour"] < 9 or environment["hour"] > 18:
        return False

    # Only allow access to documents in the same department
    if user["department"] != resource["department"]:
        return False

    # Managers can perform all actions, regular employees can only read
    if action == "write" and user["level"] < 3:
        return False

    return True

# OAuth 2.0 scope definition example
scopes:
  read:profile: "Read user profile"
  write:profile: "Update user profile"
  read:orders: "View order history"
  admin:users: "User management (admin only)"

import jwt
from datetime import datetime, timedelta, timezone

# JWT creation (RS256 recommended)
def create_token(user_id, role):
    payload = {
        "sub": user_id,
        "role": role,
        "iat": datetime.now(timezone.utc),
        "exp": datetime.now(timezone.utc) + timedelta(minutes=15),
        "iss": "myapp.com",
        "aud": "myapp.com"
    }
    return jwt.encode(payload, PRIVATE_KEY, algorithm="RS256")

# JWT verification
def verify_token(token):
    try:
        payload = jwt.decode(
            token,
            PUBLIC_KEY,
            algorithms=["RS256"],  # Explicitly specify algorithm (prevent none attack)
            issuer="myapp.com",
            audience="myapp.com"
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise AuthError("Token has expired")
    except jwt.InvalidTokenError:
        raise AuthError("Invalid token")

from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import os

# Key generation (256-bit)
key = AESGCM.generate_key(bit_length=256)

# Encryption
aesgcm = AESGCM(key)
nonce = os.urandom(12)  # 96-bit nonce
plaintext = b"Sensitive data here"
aad = b"additional authenticated data"

ciphertext = aesgcm.encrypt(nonce, plaintext, aad)

# Decryption
decrypted = aesgcm.decrypt(nonce, ciphertext, aad)

from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes, serialization

# ECDSA key pair generation
private_key = ec.generate_private_key(ec.SECP256R1())
public_key = private_key.public_key()

# Signing
from cryptography.hazmat.primitives.asymmetric import utils
signature = private_key.sign(
    b"message to sign",
    ec.ECDSA(hashes.SHA256())
)

# Verification
public_key.verify(
    signature,
    b"message to sign",
    ec.ECDSA(hashes.SHA256())
)

# Nginx TLS configuration
server {
    listen 443 ssl http2;

    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256;
    ssl_prefer_server_ciphers on;

    # HSTS header
    add_header Strict-Transport-Security "max-age=63072000; includeSubDomains; preload" always;

    # OCSP Stapling
    ssl_stapling on;
    ssl_stapling_verify on;
}

import re
from pydantic import BaseModel, validator, constr

class UserInput(BaseModel):
    username: constr(min_length=3, max_length=30, pattern=r'^[a-zA-Z0-9_]+$')
    email: str
    age: int

    @validator('email')
    def validate_email(cls, v):
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(pattern, v):
            raise ValueError('Invalid email format')
        return v

    @validator('age')
    def validate_age(cls, v):
        if v < 0 or v > 150:
            raise ValueError('Invalid age')
        return v

# Escape HTML output
from markupsafe import escape

user_input = '<script>alert("XSS")</script>'
safe_output = escape(user_input)
# Result: &lt;script&gt;alert("XSS")&lt;/script&gt;

# Using SQLAlchemy ORM
from sqlalchemy import text

# Safe approach
result = db.session.execute(
    text("SELECT * FROM users WHERE email = :email"),
    {"email": user_email}
)

# Flask-CORS configuration
from flask_cors import CORS

CORS(app, resources={
    r"/api/*": {
        "origins": ["https://myapp.com", "https://admin.myapp.com"],
        "methods": ["GET", "POST", "PUT", "DELETE"],
        "allow_headers": ["Content-Type", "Authorization"],
        "max_age": 86400
    }
})

# CSP header configuration
@app.after_request
def set_csp(response):
    response.headers['Content-Security-Policy'] = (
        "default-src 'self'; "
        "script-src 'self' 'nonce-abc123'; "
        "style-src 'self' 'unsafe-inline'; "
        "img-src 'self' data: https:; "
        "connect-src 'self' https://api.myapp.com; "
        "frame-ancestors 'none'; "
        "base-uri 'self'"
    )
    return response

# Using Flask-Limiter
from flask_limiter import Limiter

limiter = Limiter(
    app,
    default_limits=["200 per day", "50 per hour"]
)

@app.route("/api/login", methods=["POST"])
@limiter.limit("5 per minute")
def login():
    # login logic
    pass

@app.route("/api/data")
@limiter.limit("100 per hour")
def get_data():
    # data retrieval logic
    pass

# mTLS configuration on Kubernetes Ingress
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  annotations:
    nginx.ingress.kubernetes.io/auth-tls-verify-client: "on"
    nginx.ingress.kubernetes.io/auth-tls-secret: "default/ca-secret"
spec:
  tls:
    - hosts:
        - api.myapp.com
      secretName: tls-secret

import hmac
import hashlib
import time

def create_hmac_signature(secret_key, method, path, body, timestamp):
    message = f"{method}\n{path}\n{timestamp}\n{body}"
    signature = hmac.new(
        secret_key.encode(),
        message.encode(),
        hashlib.sha256
    ).hexdigest()
    return signature

# Verification (with timestamp check)
def verify_request(secret_key, method, path, body, timestamp, signature):
    # Only allow requests within 5 minutes (replay attack prevention)
    if abs(time.time() - float(timestamp)) > 300:
        return False

    expected = create_hmac_signature(secret_key, method, path, body, timestamp)
    return hmac.compare_digest(expected, signature)

# Scan container image vulnerabilities with Trivy
trivy image myapp:latest

# Filter by severity
trivy image --severity HIGH,CRITICAL myapp:latest

# Use in CI/CD (fail build if vulnerabilities found)
trivy image --exit-code 1 --severity CRITICAL myapp:latest

# Multi-stage build
FROM node:20-alpine AS builder
WORKDIR /app
COPY package*.json ./
RUN npm ci --only=production
COPY . .
RUN npm run build

# Production image
FROM node:20-alpine
RUN addgroup -g 1001 -S appgroup && \
    adduser -S appuser -u 1001 -G appgroup
WORKDIR /app
COPY --from=builder --chown=appuser:appgroup /app/dist ./dist
COPY --from=builder --chown=appuser:appgroup /app/node_modules ./node_modules

# Run as non-root user
USER appuser
EXPOSE 3000
CMD ["node", "dist/server.js"]

# Kubernetes Pod Security Context
apiVersion: v1
kind: Pod
metadata:
  name: secure-app
spec:
  securityContext:
    runAsNonRoot: true
    runAsUser: 1001
    fsGroup: 1001
  containers:
    - name: app
      image: myapp:latest
      securityContext:
        allowPrivilegeEscalation: false
        readOnlyRootFilesystem: true
        capabilities:
          drop:
            - ALL

# Policy to deny privileged containers
apiVersion: constraints.gatekeeper.sh/v1beta1
kind: K8sPSPPrivilegedContainer
metadata:
  name: deny-privileged
spec:
  match:
    kinds:
      - apiGroups: [""]
        kinds: ["Pod"]
    excludedNamespaces:
      - kube-system

# Run SonarQube in GitHub Actions
name: Security Scan
on: [push, pull_request]

jobs:
  sonarqube:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: SonarQube Scan
        uses: sonarsource/sonarqube-scan-action@master
        env:
          SONAR_TOKEN: SONAR_TOKEN_PLACEHOLDER
          SONAR_HOST_URL: SONAR_HOST_PLACEHOLDER

# Dynamic security scanning with OWASP ZAP
docker run -t zaproxy/zap-stable zap-baseline.py \
  -t https://myapp.com \
  -r report.html

# Dependabot configuration (.github/dependabot.yml)
version: 2
updates:
  - package-ecosystem: "npm"
    directory: "/"
    schedule:
      interval: "weekly"
    open-pull-requests-limit: 10
  - package-ecosystem: "pip"
    directory: "/"
    schedule:
      interval: "weekly"

# Generate SBOM with Syft
syft myapp:latest -o spdx-json > sbom.json

# Scan vulnerabilities based on SBOM with Grype
grype sbom:sbom.json

# .pre-commit-config.yaml
repos:
  - repo: https://github.com/Yelp/detect-secrets
    rev: v1.4.0
    hooks:
      - id: detect-secrets
        args: ['--baseline', '.secrets.baseline']
  - repo: https://github.com/zricethezav/gitleaks
    rev: v8.18.0
    hooks:
      - id: gitleaks

# Emergency response example: revoke leaked API keys
# 1. Immediately revoke affected keys
aws iam delete-access-key --user-name compromised-user --access-key-id AKIA...

# 2. Invalidate all related sessions
aws iam put-user-policy --user-name compromised-user \
  --policy-name DenyAll --policy-document file://deny-all.json

# 3. Issue new credentials and distribute securely