# Security Standards Reference Comprehensive security standards and secure coding practices for application security. --- ## Table of Contents - [OWASP Top 10](#owasp-top-10) - [Secure Coding Practices](#secure-coding-practices) - [Authentication Standards](#authentication-standards) - [API Security](#api-security) - [Secrets Management](#secrets-management) - [Security Headers](#security-headers) --- ## OWASP Top 10 ### A01:2021 - Broken Access Control **Description:** Access control enforces policy such that users cannot act outside of their intended permissions. **Prevention:** ```python # BAD - No authorization check @app.route('/admin/users/') def get_user(user_id): return User.query.get(user_id).to_dict() # GOOD - Authorization enforced @app.route('/admin/users/') @requires_role('admin') def get_user(user_id): user = User.query.get_or_404(user_id) if not current_user.can_access(user): abort(403) return user.to_dict() ``` **Checklist:** - [ ] Deny access by default (allowlist approach) - [ ] Implement RBAC or ABAC consistently - [ ] Validate object-level authorization (IDOR prevention) - [ ] Disable directory listing - [ ] Log access control failures and alert on repeated failures ### A02:2021 - Cryptographic Failures **Description:** Failures related to cryptography which often lead to exposure of sensitive data. **Prevention:** ```python # BAD - Weak hashing import hashlib password_hash = hashlib.md5(password.encode()).hexdigest() # GOOD - Strong password hashing from argon2 import PasswordHasher ph = PasswordHasher( time_cost=3, memory_cost=65536, parallelism=4 ) password_hash = ph.hash(password) # Verify password try: ph.verify(stored_hash, password) except argon2.exceptions.VerifyMismatchError: raise InvalidCredentials() ``` **Checklist:** - [ ] Use TLS 1.2+ for all data in transit - [ ] Use AES-256-GCM for encryption at rest - [ ] Use Argon2id, bcrypt, or scrypt for passwords - [ ] Never use MD5, SHA1 for security purposes - [ ] Rotate encryption keys regularly ### A03:2021 - Injection **Description:** Untrusted data sent to an interpreter as part of a command or query. **SQL Injection Prevention:** ```python # BAD - String concatenation (VULNERABLE) query = f"SELECT * FROM users WHERE id = {user_id}" cursor.execute(query) # GOOD - Parameterized queries cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,)) # GOOD - ORM with parameter binding user = User.query.filter_by(id=user_id).first() ``` **Command Injection Prevention:** ```python # BAD - Shell execution with user input (VULNERABLE) # NEVER use: os.system(f"ping {user_input}") # GOOD - Use subprocess with shell=False and validated input import subprocess def safe_ping(hostname: str) -> str: # Validate hostname format first if not is_valid_hostname(hostname): raise ValueError("Invalid hostname") result = subprocess.run( ["ping", "-c", "4", hostname], shell=False, capture_output=True, text=True ) return result.stdout ``` **XSS Prevention:** ```python # BAD - Direct HTML insertion (VULNERABLE) return f"
Welcome, {username}
" # GOOD - HTML escaping from markupsafe import escape return f"
Welcome, {escape(username)}
" # GOOD - Template auto-escaping (Jinja2) # {{ username }} is auto-escaped by default ``` ### A04:2021 - Insecure Design **Description:** Risks related to design and architectural flaws. **Prevention Patterns:** ```python # Threat modeling categories (STRIDE) THREATS = { 'Spoofing': 'Authentication controls', 'Tampering': 'Integrity controls', 'Repudiation': 'Audit logging', 'Information Disclosure': 'Encryption, access control', 'Denial of Service': 'Rate limiting, resource limits', 'Elevation of Privilege': 'Authorization controls' } # Defense in depth - multiple layers class SecurePaymentFlow: def process_payment(self, payment_data): # Layer 1: Input validation self.validate_input(payment_data) # Layer 2: Authentication check self.verify_user_authenticated() # Layer 3: Authorization check self.verify_user_can_pay(payment_data.amount) # Layer 4: Rate limiting self.check_rate_limit() # Layer 5: Fraud detection self.check_fraud_signals(payment_data) # Layer 6: Secure processing return self.execute_payment(payment_data) ``` ### A05:2021 - Security Misconfiguration **Description:** Missing or incorrect security hardening. **Prevention:** ```yaml # Kubernetes pod security apiVersion: v1 kind: Pod spec: securityContext: runAsNonRoot: true runAsUser: 1000 fsGroup: 1000 containers: - name: app securityContext: allowPrivilegeEscalation: false readOnlyRootFilesystem: true capabilities: drop: - ALL ``` ```python # Flask security configuration app.config.update( SESSION_COOKIE_SECURE=True, SESSION_COOKIE_HTTPONLY=True, SESSION_COOKIE_SAMESITE='Lax', PERMANENT_SESSION_LIFETIME=timedelta(hours=1), ) ``` --- ## Secure Coding Practices ### Input Validation ```python from pydantic import BaseModel, validator, constr from typing import Optional import re class UserInput(BaseModel): username: constr(min_length=3, max_length=50, regex=r'^[a-zA-Z0-9_]+$') email: str age: Optional[int] = None @validator('email') def validate_email(cls, v): # Use proper email validation 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.lower() @validator('age') def validate_age(cls, v): if v is not None and (v < 0 or v > 150): raise ValueError('Age must be between 0 and 150') return v ``` ### Output Encoding ```python import html import json from urllib.parse import quote def encode_for_html(data: str) -> str: """Encode data for safe HTML output.""" return html.escape(data) def encode_for_javascript(data: str) -> str: """Encode data for safe JavaScript string.""" return json.dumps(data) def encode_for_url(data: str) -> str: """Encode data for safe URL parameter.""" return quote(data, safe='') def encode_for_css(data: str) -> str: """Encode data for safe CSS value.""" return ''.join( c if c.isalnum() else f'\\{ord(c):06x}' for c in data ) ``` ### Error Handling ```python import logging from typing import Dict, Any logger = logging.getLogger(__name__) class SecurityException(Exception): """Base exception for security-related errors.""" def __init__(self, message: str, internal_details: str = None): # User-facing message (safe to display) self.message = message # Internal details (for logging only) self.internal_details = internal_details super().__init__(message) def handle_request(): try: process_sensitive_data() except DatabaseError as e: # Log full details internally logger.error(f"Database error: {e}", exc_info=True) # Return generic message to user raise SecurityException( "An error occurred processing your request", internal_details=str(e) ) except Exception as e: logger.error(f"Unexpected error: {e}", exc_info=True) raise SecurityException("An unexpected error occurred") ``` --- ## Authentication Standards ### Password Requirements ```python import re from typing import Tuple def validate_password(password: str) -> Tuple[bool, str]: """ Validate password against security requirements. Requirements: - Minimum 12 characters - At least one uppercase letter - At least one lowercase letter - At least one digit - At least one special character - Not in common password list """ if len(password) < 12: return False, "Password must be at least 12 characters" if not re.search(r'[A-Z]', password): return False, "Password must contain uppercase letter" if not re.search(r'[a-z]', password): return False, "Password must contain lowercase letter" if not re.search(r'\d', password): return False, "Password must contain a digit" if not re.search(r'[!@#$%^&*(),.?":{}|<>]', password): return False, "Password must contain special character" # Check against common passwords (use haveibeenpwned API in production) common_passwords = {'password123', 'qwerty123456', 'admin123456'} if password.lower() in common_passwords: return False, "Password is too common" return True, "Password meets requirements" ``` ### JWT Best Practices ```python import jwt from datetime import datetime, timedelta from typing import Dict, Optional class JWTManager: def __init__(self, secret_key: str, algorithm: str = 'HS256'): self.secret_key = secret_key self.algorithm = algorithm self.access_token_expiry = timedelta(minutes=15) self.refresh_token_expiry = timedelta(days=7) def create_access_token(self, user_id: str, roles: list) -> str: payload = { 'sub': user_id, 'roles': roles, 'type': 'access', 'iat': datetime.utcnow(), 'exp': datetime.utcnow() + self.access_token_expiry, 'jti': self._generate_jti() # Unique token ID for revocation } return jwt.encode(payload, self.secret_key, algorithm=self.algorithm) def verify_token(self, token: str) -> Optional[Dict]: try: payload = jwt.decode( token, self.secret_key, algorithms=[self.algorithm], options={ 'require': ['exp', 'iat', 'sub', 'jti'], 'verify_exp': True } ) # Check if token is revoked if self._is_token_revoked(payload['jti']): return None return payload except jwt.ExpiredSignatureError: return None except jwt.InvalidTokenError: return None ``` ### MFA Implementation ```python import pyotp import qrcode from io import BytesIO import base64 class TOTPManager: def __init__(self, issuer: str = "MyApp"): self.issuer = issuer def generate_secret(self) -> str: """Generate a new TOTP secret for a user.""" return pyotp.random_base32() def get_provisioning_uri(self, secret: str, email: str) -> str: """Generate URI for QR code.""" totp = pyotp.TOTP(secret) return totp.provisioning_uri(name=email, issuer_name=self.issuer) def generate_qr_code(self, provisioning_uri: str) -> str: """Generate base64-encoded QR code image.""" qr = qrcode.QRCode(version=1, box_size=10, border=5) qr.add_data(provisioning_uri) qr.make(fit=True) img = qr.make_image(fill_color="black", back_color="white") buffer = BytesIO() img.save(buffer, format='PNG') return base64.b64encode(buffer.getvalue()).decode() def verify_totp(self, secret: str, code: str) -> bool: """Verify TOTP code with time window tolerance.""" totp = pyotp.TOTP(secret) # Allow 1 period before/after for clock skew return totp.verify(code, valid_window=1) ``` --- ## API Security ### Rate Limiting ```python from functools import wraps from flask import request, jsonify import time from collections import defaultdict import threading class RateLimiter: def __init__(self, requests_per_minute: int = 60): self.requests_per_minute = requests_per_minute self.requests = defaultdict(list) self.lock = threading.Lock() def is_rate_limited(self, identifier: str) -> bool: with self.lock: now = time.time() minute_ago = now - 60 # Clean old requests self.requests[identifier] = [ req_time for req_time in self.requests[identifier] if req_time > minute_ago ] if len(self.requests[identifier]) >= self.requests_per_minute: return True self.requests[identifier].append(now) return False rate_limiter = RateLimiter(requests_per_minute=100) def rate_limit(f): @wraps(f) def decorated_function(*args, **kwargs): identifier = request.remote_addr if rate_limiter.is_rate_limited(identifier): return jsonify({ 'error': 'Rate limit exceeded', 'retry_after': 60 }), 429 return f(*args, **kwargs) return decorated_function ``` ### API Key Validation ```python import hashlib import secrets from datetime import datetime from typing import Optional, Dict class APIKeyManager: def __init__(self, db): self.db = db def generate_api_key(self, user_id: str, name: str, scopes: list) -> Dict: """Generate a new API key.""" # Generate key with prefix for identification raw_key = f"sk_live_{secrets.token_urlsafe(32)}" # Store hash only key_hash = hashlib.sha256(raw_key.encode()).hexdigest() api_key_record = { 'id': secrets.token_urlsafe(16), 'user_id': user_id, 'name': name, 'key_hash': key_hash, 'key_prefix': raw_key[:12], # Store prefix for identification 'scopes': scopes, 'created_at': datetime.utcnow(), 'last_used_at': None } self.db.api_keys.insert(api_key_record) # Return raw key only once return { 'key': raw_key, 'id': api_key_record['id'], 'scopes': scopes } def validate_api_key(self, raw_key: str) -> Optional[Dict]: """Validate an API key and return associated data.""" key_hash = hashlib.sha256(raw_key.encode()).hexdigest() api_key = self.db.api_keys.find_one({'key_hash': key_hash}) if not api_key: return None # Update last used timestamp self.db.api_keys.update( {'id': api_key['id']}, {'last_used_at': datetime.utcnow()} ) return { 'user_id': api_key['user_id'], 'scopes': api_key['scopes'] } ``` --- ## Secrets Management ### Environment Variables ```python import os from typing import Optional from dataclasses import dataclass @dataclass class AppSecrets: database_url: str jwt_secret: str api_key: str encryption_key: str def load_secrets() -> AppSecrets: """Load secrets from environment with validation.""" def get_required(name: str) -> str: value = os.environ.get(name) if not value: raise ValueError(f"Required environment variable {name} is not set") return value return AppSecrets( database_url=get_required('DATABASE_URL'), jwt_secret=get_required('JWT_SECRET'), api_key=get_required('API_KEY'), encryption_key=get_required('ENCRYPTION_KEY') ) # Never log secrets import logging class SecretFilter(logging.Filter): """Filter to redact secrets from logs.""" def __init__(self, secrets: list): super().__init__() self.secrets = secrets def filter(self, record): message = record.getMessage() for secret in self.secrets: if secret in message: record.msg = record.msg.replace(secret, '[REDACTED]') return True ``` ### HashiCorp Vault Integration ```python import hvac from typing import Dict, Optional class VaultClient: def __init__(self, url: str, token: str = None, role_id: str = None, secret_id: str = None): self.client = hvac.Client(url=url) if token: self.client.token = token elif role_id and secret_id: # AppRole authentication self.client.auth.approle.login( role_id=role_id, secret_id=secret_id ) def get_secret(self, path: str, key: str) -> Optional[str]: """Retrieve a secret from Vault.""" try: response = self.client.secrets.kv.v2.read_secret_version(path=path) return response['data']['data'].get(key) except hvac.exceptions.InvalidPath: return None def get_database_credentials(self, role: str) -> Dict[str, str]: """Get dynamic database credentials.""" response = self.client.secrets.database.generate_credentials(name=role) return { 'username': response['data']['username'], 'password': response['data']['password'], 'lease_id': response['lease_id'], 'lease_duration': response['lease_duration'] } ``` --- ## Security Headers ### HTTP Security Headers ```python from flask import Flask, Response def add_security_headers(response: Response) -> Response: """Add security headers to HTTP response.""" # Prevent clickjacking response.headers['X-Frame-Options'] = 'DENY' # Enable XSS filter response.headers['X-XSS-Protection'] = '1; mode=block' # Prevent MIME type sniffing response.headers['X-Content-Type-Options'] = 'nosniff' # Referrer policy response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin' # Content Security Policy response.headers['Content-Security-Policy'] = ( "default-src 'self'; " "script-src 'self' 'unsafe-inline'; " "style-src 'self' 'unsafe-inline'; " "img-src 'self' data: https:; " "font-src 'self'; " "frame-ancestors 'none'; " "form-action 'self'" ) # HSTS (enable only with valid HTTPS) response.headers['Strict-Transport-Security'] = ( 'max-age=31536000; includeSubDomains; preload' ) # Permissions Policy response.headers['Permissions-Policy'] = ( 'geolocation=(), microphone=(), camera=()' ) return response app = Flask(__name__) app.after_request(add_security_headers) ``` --- ## Quick Reference ### Security Checklist | Category | Check | Priority | |----------|-------|----------| | Authentication | MFA enabled | Critical | | Authentication | Password policy enforced | Critical | | Authorization | RBAC implemented | Critical | | Input | All inputs validated | Critical | | Injection | Parameterized queries | Critical | | Crypto | TLS 1.2+ enforced | Critical | | Secrets | No hardcoded secrets | Critical | | Headers | Security headers set | High | | Logging | Security events logged | High | | Dependencies | No known vulnerabilities | High | ### Tool Recommendations | Purpose | Tool | Usage | |---------|------|-------| | SAST | Semgrep | `semgrep --config auto .` | | SAST | Bandit (Python) | `bandit -r src/` | | Secrets | Gitleaks | `gitleaks detect --source .` | | Dependencies | Snyk | `snyk test` | | Container | Trivy | `trivy image myapp:latest` | | DAST | OWASP ZAP | Dynamic scanning |