Security Standards Reference
Comprehensive security standards and secure coding practices for application security.
Table of Contents
- OWASP Top 10
- Secure Coding Practices
- Authentication Standards
- API Security
- Secrets Management
- 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:
# BAD - No authorization check
@app.route('/admin/users/<user_id>')
def get_user(user_id):
return User.query.get(user_id).to_dict()
# GOOD - Authorization enforced
@app.route('/admin/users/<user_id>')
@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:
# 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:
# 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:
# 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:
# BAD - Direct HTML insertion (VULNERABLE)
return f"<div>Welcome, {username}</div>"
# GOOD - HTML escaping
from markupsafe import escape
return f"<div>Welcome, {escape(username)}</div>"
# 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:
# 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:
# 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
# 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
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
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
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
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
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
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
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
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
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
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
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 |