Compliance Requirements Reference

Comprehensive guide for SOC 2, PCI-DSS, HIPAA, and GDPR compliance requirements.

Table of Contents

• SOC 2 Type II
• PCI-DSS
• HIPAA
• GDPR
• Compliance Automation
• Audit Preparation

SOC 2 Type II

Trust Service Criteria

Security Controls Checklist

## SOC 2 Security Controls

### CC1: Control Environment
- [ ] Security policies documented and approved
- [ ] Organizational structure defined
- [ ] Security roles and responsibilities assigned
- [ ] Background checks performed on employees
- [ ] Security awareness training completed annually

### CC2: Communication and Information
- [ ] Security policies communicated to employees
- [ ] Security incidents reported and tracked
- [ ] External communications about security controls
- [ ] Service level agreements documented

### CC3: Risk Assessment
- [ ] Annual risk assessment performed
- [ ] Risk register maintained
- [ ] Risk treatment plans documented
- [ ] Vendor risk assessments completed
- [ ] Business impact analysis current

### CC4: Monitoring Activities
- [ ] Security monitoring implemented
- [ ] Log aggregation and analysis
- [ ] Vulnerability scanning (weekly)
- [ ] Penetration testing (annual)
- [ ] Security metrics reviewed monthly

### CC5: Control Activities
- [ ] Access control policies enforced
- [ ] MFA enabled for all users
- [ ] Password policy enforced (12+ chars)
- [ ] Access reviews (quarterly)
- [ ] Least privilege principle applied

### CC6: Logical and Physical Access
- [ ] Identity management system
- [ ] Role-based access control
- [ ] Physical access controls
- [ ] Network segmentation
- [ ] Data center security

### CC7: System Operations
- [ ] Change management process
- [ ] Incident management process
- [ ] Problem management process
- [ ] Capacity management
- [ ] Backup and recovery tested

### CC8: Change Management
- [ ] Change control board
- [ ] Change approval workflow
- [ ] Testing requirements documented
- [ ] Rollback procedures
- [ ] Emergency change process

### CC9: Risk Mitigation
- [ ] Insurance coverage
- [ ] Business continuity plan
- [ ] Disaster recovery plan tested
- [ ] Vendor management program

Evidence Collection

def collect_soc2_evidence(period_start: str, period_end: str) -> dict:
    """
    Collect evidence for SOC 2 audit period.

    Returns dictionary organized by Trust Service Criteria.
    """
    evidence = {
        'period': {'start': period_start, 'end': period_end},
        'security': {
            'access_reviews': get_access_reviews(period_start, period_end),
            'vulnerability_scans': get_vulnerability_reports(period_start, period_end),
            'penetration_tests': get_pentest_reports(period_start, period_end),
            'security_incidents': get_incident_reports(period_start, period_end),
            'training_records': get_training_completion(period_start, period_end),
        },
        'availability': {
            'uptime_reports': get_uptime_metrics(period_start, period_end),
            'incident_reports': get_availability_incidents(period_start, period_end),
            'dr_tests': get_dr_test_results(period_start, period_end),
            'backup_tests': get_backup_test_results(period_start, period_end),
        },
        'processing_integrity': {
            'data_validation_logs': get_validation_logs(period_start, period_end),
            'error_reports': get_error_reports(period_start, period_end),
            'reconciliation_reports': get_reconciliation_reports(period_start, period_end),
        },
        'confidentiality': {
            'encryption_status': get_encryption_audit(period_start, period_end),
            'data_classification': get_data_inventory(),
            'access_logs': get_sensitive_data_access_logs(period_start, period_end),
        }
    }

    return evidence

PCI-DSS

PCI-DSS v4.0 Requirements

Cardholder Data Protection

# PCI-DSS compliant card data handling

import re
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64
import os

class PCIDataHandler:
    """Handle cardholder data per PCI-DSS requirements."""

    # PAN patterns (masked for display)
    PAN_PATTERN = re.compile(r'\b(?:\d{4}[-\s]?){3}\d{4}\b')

    def __init__(self, encryption_key: bytes):
        self.cipher = Fernet(encryption_key)

    @staticmethod
    def mask_pan(pan: str) -> str:
        """
        Mask PAN per PCI-DSS (show first 6, last 4 only).
        Requirement 3.4: Render PAN unreadable.
        """
        digits = re.sub(r'\D', '', pan)
        if len(digits) < 13:
            return '*' * len(digits)
        return f"{digits[:6]}{'*' * (len(digits) - 10)}{digits[-4:]}"

    def encrypt_pan(self, pan: str) -> str:
        """
        Encrypt PAN for storage.
        Requirement 3.5: Protect keys used to protect stored account data.
        """
        return self.cipher.encrypt(pan.encode()).decode()

    def decrypt_pan(self, encrypted_pan: str) -> str:
        """Decrypt PAN (requires authorization logging)."""
        return self.cipher.decrypt(encrypted_pan.encode()).decode()

    @staticmethod
    def validate_pan(pan: str) -> bool:
        """Validate PAN using Luhn algorithm."""
        digits = re.sub(r'\D', '', pan)
        if len(digits) < 13 or len(digits) > 19:
            return False

        # Luhn algorithm
        total = 0
        for i, digit in enumerate(reversed(digits)):
            d = int(digit)
            if i % 2 == 1:
                d *= 2
                if d > 9:
                    d -= 9
            total += d
        return total % 10 == 0

    def sanitize_logs(self, log_message: str) -> str:
        """
        Remove PAN from log messages.
        Requirement 3.3: Mask PAN when displayed.
        """
        def replace_pan(match):
            return self.mask_pan(match.group())

        return self.PAN_PATTERN.sub(replace_pan, log_message)

Network Segmentation

# PCI-DSS network segmentation example

# Cardholder Data Environment (CDE) firewall rules
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: cde-isolation
  namespace: payment-processing
spec:
  podSelector:
    matchLabels:
      pci-zone: cde
  policyTypes:
    - Ingress
    - Egress
  ingress:
    # Only allow from payment gateway
    - from:
        - namespaceSelector:
            matchLabels:
              pci-zone: dmz
        - podSelector:
            matchLabels:
              app: payment-gateway
      ports:
        - protocol: TCP
          port: 443
  egress:
    # Only allow to payment processor
    - to:
        - ipBlock:
            cidr: 10.0.100.0/24  # Payment processor network
      ports:
        - protocol: TCP
          port: 443
    # Allow DNS
    - to:
        - namespaceSelector: {}
          podSelector:
            matchLabels:
              k8s-app: kube-dns
      ports:
        - protocol: UDP
          port: 53

HIPAA

HIPAA Security Rule Requirements

PHI Handling

from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import hashlib
import logging

# Configure PHI audit logging
phi_logger = logging.getLogger('phi_access')
phi_logger.setLevel(logging.INFO)

@dataclass
class PHIAccessLog:
    """HIPAA-compliant PHI access logging."""
    timestamp: datetime
    user_id: str
    patient_id: str
    action: str  # view, create, update, delete, export
    reason: str
    data_elements: list
    source_ip: str
    success: bool

def log_phi_access(access: PHIAccessLog):
    """
    Log PHI access per HIPAA requirements.
    164.312(b): Audit controls.
    """
    phi_logger.info(
        f"PHI_ACCESS|"
        f"timestamp={access.timestamp.isoformat()}|"
        f"user={access.user_id}|"
        f"patient={access.patient_id}|"
        f"action={access.action}|"
        f"reason={access.reason}|"
        f"elements={','.join(access.data_elements)}|"
        f"ip={access.source_ip}|"
        f"success={access.success}"
    )

class HIPAACompliantStorage:
    """HIPAA-compliant PHI storage handler."""

    # Minimum Necessary Standard - only access needed data
    PHI_ELEMENTS = {
        'patient_name': 'high',
        'ssn': 'high',
        'medical_record_number': 'high',
        'diagnosis': 'medium',
        'treatment_plan': 'medium',
        'appointment_date': 'low',
        'provider_name': 'low'
    }

    def __init__(self, encryption_service, user_context):
        self.encryption = encryption_service
        self.user = user_context

    def access_phi(
        self,
        patient_id: str,
        elements: list,
        reason: str
    ) -> Optional[dict]:
        """
        Access PHI with HIPAA controls.

        Args:
            patient_id: Patient identifier
            elements: List of PHI elements to access
            reason: Business reason for access

        Returns:
            Requested PHI elements if authorized
        """
        # Verify minimum necessary - user only gets needed elements
        authorized_elements = self._check_authorization(elements)

        if not authorized_elements:
            log_phi_access(PHIAccessLog(
                timestamp=datetime.utcnow(),
                user_id=self.user.id,
                patient_id=patient_id,
                action='view',
                reason=reason,
                data_elements=elements,
                source_ip=self.user.ip_address,
                success=False
            ))
            raise PermissionError("Not authorized for requested PHI elements")

        # Retrieve and decrypt PHI
        phi_data = self._retrieve_phi(patient_id, authorized_elements)

        # Log successful access
        log_phi_access(PHIAccessLog(
            timestamp=datetime.utcnow(),
            user_id=self.user.id,
            patient_id=patient_id,
            action='view',
            reason=reason,
            data_elements=authorized_elements,
            source_ip=self.user.ip_address,
            success=True
        ))

        return phi_data

    def _check_authorization(self, requested_elements: list) -> list:
        """Check user authorization for PHI elements."""
        user_clearance = self.user.hipaa_clearance_level
        authorized = []

        for element in requested_elements:
            element_level = self.PHI_ELEMENTS.get(element, 'high')
            if self._clearance_allows(user_clearance, element_level):
                authorized.append(element)

        return authorized

GDPR

GDPR Principles

Data Subject Rights Implementation

from datetime import datetime, timedelta
from enum import Enum
from typing import Optional, List
import json

class DSRType(Enum):
    ACCESS = "access"           # Article 15
    RECTIFICATION = "rectification"  # Article 16
    ERASURE = "erasure"         # Article 17 (Right to be forgotten)
    RESTRICTION = "restriction"  # Article 18
    PORTABILITY = "portability"  # Article 20
    OBJECTION = "objection"     # Article 21

class DataSubjectRequest:
    """Handle GDPR Data Subject Requests."""

    # GDPR requires response within 30 days
    RESPONSE_DEADLINE_DAYS = 30

    def __init__(self, db, notification_service):
        self.db = db
        self.notifications = notification_service

    def submit_request(
        self,
        subject_email: str,
        request_type: DSRType,
        details: str
    ) -> dict:
        """
        Submit a Data Subject Request.

        Args:
            subject_email: Email of the data subject
            request_type: Type of GDPR request
            details: Additional request details

        Returns:
            Request tracking information
        """
        # Verify identity before processing
        verification_token = self._send_verification(subject_email)

        request = {
            'id': self._generate_request_id(),
            'subject_email': subject_email,
            'type': request_type.value,
            'details': details,
            'status': 'pending_verification',
            'submitted_at': datetime.utcnow().isoformat(),
            'deadline': (datetime.utcnow() + timedelta(days=self.RESPONSE_DEADLINE_DAYS)).isoformat(),
            'verification_token': verification_token
        }

        self.db.dsr_requests.insert(request)

        # Notify DPO
        self.notifications.notify_dpo(
            f"New DSR ({request_type.value}) received",
            request
        )

        return {
            'request_id': request['id'],
            'deadline': request['deadline'],
            'status': 'verification_sent'
        }

    def process_erasure_request(self, request_id: str) -> dict:
        """
        Process Article 17 erasure request (Right to be Forgotten).

        Returns:
            Erasure completion report
        """
        request = self.db.dsr_requests.find_one({'id': request_id})
        subject_email = request['subject_email']

        erasure_report = {
            'request_id': request_id,
            'subject': subject_email,
            'systems_processed': [],
            'data_deleted': [],
            'data_retained': [],  # With legal basis
            'completed_at': None
        }

        # Find all data for this subject
        data_inventory = self._find_subject_data(subject_email)

        for data_item in data_inventory:
            if self._can_delete(data_item):
                self._delete_data(data_item)
                erasure_report['data_deleted'].append({
                    'system': data_item['system'],
                    'data_type': data_item['type'],
                    'deleted_at': datetime.utcnow().isoformat()
                })
            else:
                erasure_report['data_retained'].append({
                    'system': data_item['system'],
                    'data_type': data_item['type'],
                    'retention_reason': data_item['legal_basis']
                })

        erasure_report['completed_at'] = datetime.utcnow().isoformat()

        # Update request status
        self.db.dsr_requests.update(
            {'id': request_id},
            {'status': 'completed', 'completion_report': erasure_report}
        )

        return erasure_report

    def generate_portability_export(self, request_id: str) -> dict:
        """
        Generate Article 20 data portability export.

        Returns machine-readable export in JSON format.
        """
        request = self.db.dsr_requests.find_one({'id': request_id})
        subject_email = request['subject_email']

        export_data = {
            'export_date': datetime.utcnow().isoformat(),
            'data_subject': subject_email,
            'format': 'JSON',
            'data': {}
        }

        # Collect data from all systems
        systems = ['user_accounts', 'orders', 'preferences', 'communications']

        for system in systems:
            system_data = self._extract_portable_data(system, subject_email)
            if system_data:
                export_data['data'][system] = system_data

        return export_data

Consent Management

class ConsentManager:
    """GDPR-compliant consent management."""

    def __init__(self, db):
        self.db = db

    def record_consent(
        self,
        user_id: str,
        purpose: str,
        consent_given: bool,
        consent_text: str
    ) -> dict:
        """
        Record consent per GDPR Article 7 requirements.

        Consent must be:
        - Freely given
        - Specific
        - Informed
        - Unambiguous
        """
        consent_record = {
            'user_id': user_id,
            'purpose': purpose,
            'consent_given': consent_given,
            'consent_text': consent_text,
            'timestamp': datetime.utcnow().isoformat(),
            'method': 'explicit_checkbox',  # Not pre-ticked
            'ip_address': self._get_user_ip(),
            'user_agent': self._get_user_agent(),
            'version': '1.0'  # Track consent version
        }

        self.db.consents.insert(consent_record)

        return consent_record

    def check_consent(self, user_id: str, purpose: str) -> bool:
        """Check if user has given consent for specific purpose."""
        latest_consent = self.db.consents.find_one(
            {'user_id': user_id, 'purpose': purpose},
            sort=[('timestamp', -1)]
        )

        return latest_consent and latest_consent.get('consent_given', False)

    def withdraw_consent(self, user_id: str, purpose: str) -> dict:
        """
        Process consent withdrawal.

        GDPR Article 7(3): Withdrawal must be as easy as giving consent.
        """
        withdrawal_record = {
            'user_id': user_id,
            'purpose': purpose,
            'consent_given': False,
            'timestamp': datetime.utcnow().isoformat(),
            'action': 'withdrawal'
        }

        self.db.consents.insert(withdrawal_record)

        # Trigger data processing stop for this purpose
        self._stop_processing(user_id, purpose)

        return withdrawal_record

Compliance Automation

Automated Compliance Checks

# compliance-checks.yml - GitHub Actions

name: Compliance Checks

on:
  push:
    branches: [main]
  pull_request:
  schedule:
    - cron: '0 0 * * *'  # Daily

jobs:
  soc2-checks:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Check for secrets in code
        run: |
          gitleaks detect --source . --report-format json --report-path gitleaks-report.json
          if [ -s gitleaks-report.json ]; then
            echo "Secrets detected in code!"
            exit 1
          fi

      - name: Verify encryption at rest
        run: |
          # Check database encryption configuration
          python scripts/compliance_checker.py --check encryption

      - name: Verify access controls
        run: |
          # Check RBAC configuration
          python scripts/compliance_checker.py --check access-control

      - name: Check logging configuration
        run: |
          # Verify audit logging enabled
          python scripts/compliance_checker.py --check audit-logging

  pci-checks:
    runs-on: ubuntu-latest
    if: contains(github.event.head_commit.message, '[pci]')
    steps:
      - uses: actions/checkout@v4

      - name: Scan for PAN in code
        run: |
          # Check for unencrypted card numbers
          python scripts/compliance_checker.py --check pci-pan-exposure

      - name: Verify TLS configuration
        run: |
          python scripts/compliance_checker.py --check tls-config

  gdpr-checks:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Check data retention policies
        run: |
          python scripts/compliance_checker.py --check data-retention

      - name: Verify consent mechanisms
        run: |
          python scripts/compliance_checker.py --check consent-management

Audit Preparation

Audit Readiness Checklist

## Pre-Audit Checklist

### 60 Days Before Audit
- [ ] Confirm audit scope and timeline
- [ ] Identify control owners
- [ ] Begin evidence collection
- [ ] Review previous audit findings
- [ ] Update policies and procedures

### 30 Days Before Audit
- [ ] Complete evidence collection
- [ ] Perform internal control testing
- [ ] Remediate any gaps identified
- [ ] Prepare executive summary
- [ ] Brief stakeholders

### 7 Days Before Audit
- [ ] Finalize evidence package
- [ ] Prepare interview schedules
- [ ] Set up secure evidence sharing
- [ ] Confirm auditor logistics
- [ ] Final gap assessment

### During Audit
- [ ] Daily status meetings
- [ ] Timely evidence delivery
- [ ] Document all requests
- [ ] Escalate issues promptly
- [ ] Maintain communication log

Evidence Repository Structure

evidence/
├── period_YYYY-MM/
│   ├── security/
│   │   ├── access_reviews/
│   │   ├── vulnerability_scans/
│   │   ├── penetration_tests/
│   │   └── security_training/
│   ├── availability/
│   │   ├── uptime_reports/
│   │   ├── incident_reports/
│   │   └── dr_tests/
│   ├── change_management/
│   │   ├── change_requests/
│   │   ├── approval_records/
│   │   └── deployment_logs/
│   ├── policies/
│   │   ├── current_policies/
│   │   └── acknowledgments/
│   └── index.json

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

Tool Recommendations

Vulnerability Management Guide

Complete workflow for vulnerability identification, assessment, prioritization, and remediation.

Table of Contents

• Vulnerability Lifecycle
• CVE Triage Process
• CVSS Scoring
• Remediation Workflows
• Dependency Scanning
• Security Incident Response

Vulnerability Lifecycle

Overview

┌─────────────┐    ┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│  DISCOVER   │ →  │   ASSESS    │ →  │ PRIORITIZE  │ →  │  REMEDIATE  │
│             │    │             │    │             │    │             │
│ - Scanning  │    │ - CVSS      │    │ - Risk      │    │ - Patch     │
│ - Reports   │    │ - Context   │    │ - Business  │    │ - Mitigate  │
│ - Audits    │    │ - Impact    │    │ - SLA       │    │ - Accept    │
└─────────────┘    └─────────────┘    └─────────────┘    └─────────────┘
                                                                │
                                                                ▼
                                                        ┌─────────────┐
                                                        │   VERIFY    │
                                                        │             │
                                                        │ - Retest    │
                                                        │ - Close     │
                                                        └─────────────┘

State Definitions

CVE Triage Process

Step 1: Initial Assessment

def triage_cve(cve_id: str, affected_systems: list) -> dict:
    """
    Perform initial triage of a CVE.

    Returns triage assessment with severity and recommended actions.
    """
    # Fetch CVE details from NVD
    cve_data = fetch_nvd_data(cve_id)

    assessment = {
        'cve_id': cve_id,
        'published': cve_data['published'],
        'base_cvss': cve_data['cvss_v3']['base_score'],
        'vector': cve_data['cvss_v3']['vector_string'],
        'description': cve_data['description'],
        'affected_systems': [],
        'exploitability': check_exploitability(cve_id),
        'recommendation': None
    }

    # Check which systems are actually affected
    for system in affected_systems:
        if is_system_vulnerable(system, cve_data):
            assessment['affected_systems'].append({
                'name': system.name,
                'version': system.version,
                'exposure': assess_exposure(system)
            })

    # Determine recommendation
    assessment['recommendation'] = determine_action(assessment)

    return assessment

Step 2: Severity Classification

Step 3: Context Analysis

## CVE Context Checklist

### Exposure Assessment
- [ ] Is the vulnerable component internet-facing?
- [ ] Is the vulnerable component in a DMZ?
- [ ] Does the component process sensitive data?
- [ ] Are there compensating controls in place?

### Exploitability Assessment
- [ ] Is there a public exploit available?
- [ ] Is exploitation being observed in the wild?
- [ ] What privileges are required to exploit?
- [ ] Does exploit require user interaction?

### Business Impact
- [ ] What business processes depend on affected systems?
- [ ] What is the potential data exposure?
- [ ] What are regulatory implications?
- [ ] What is the reputational risk?

Step 4: Triage Decision Matrix

CVSS Scoring

CVSS v3.1 Vector Components

CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
        │     │    │    │    │   │   │   │
        │     │    │    │    │   │   │   └── Availability Impact (H/L/N)
        │     │    │    │    │   │   └────── Integrity Impact (H/L/N)
        │     │    │    │    │   └────────── Confidentiality Impact (H/L/N)
        │     │    │    │    └────────────── Scope (C/U)
        │     │    │    └─────────────────── User Interaction (R/N)
        │     │    └──────────────────────── Privileges Required (H/L/N)
        │     └───────────────────────────── Attack Complexity (H/L)
        └─────────────────────────────────── Attack Vector (N/A/L/P)

Environmental Score Adjustments

def calculate_environmental_score(base_cvss: float, environment: dict) -> float:
    """
    Adjust CVSS base score based on environmental factors.

    Args:
        base_cvss: Base CVSS score from NVD
        environment: Dictionary with environmental modifiers

    Returns:
        Adjusted CVSS score for this environment
    """
    # Confidentiality Requirement (CR)
    cr_modifier = {
        'high': 1.5,
        'medium': 1.0,
        'low': 0.5
    }.get(environment.get('confidentiality_requirement', 'medium'))

    # Integrity Requirement (IR)
    ir_modifier = {
        'high': 1.5,
        'medium': 1.0,
        'low': 0.5
    }.get(environment.get('integrity_requirement', 'medium'))

    # Availability Requirement (AR)
    ar_modifier = {
        'high': 1.5,
        'medium': 1.0,
        'low': 0.5
    }.get(environment.get('availability_requirement', 'medium'))

    # Modified Attack Vector (reduce if not internet-facing)
    if not environment.get('internet_facing', True):
        base_cvss = max(0, base_cvss - 1.5)

    # Compensating controls reduce score
    if environment.get('waf_protected', False):
        base_cvss = max(0, base_cvss - 0.5)

    if environment.get('network_segmented', False):
        base_cvss = max(0, base_cvss - 0.5)

    return round(min(10.0, base_cvss), 1)

Remediation Workflows

Workflow 1: Emergency Patch (P0/Critical)

Timeline: 24 hours
Stakeholders: Security, DevOps, Engineering Lead, CISO

Hour 0-2: ASSESS
├── Confirm vulnerability affects production
├── Identify all affected systems
├── Assess active exploitation
└── Notify stakeholders

Hour 2-8: MITIGATE
├── Apply temporary mitigations (WAF rules, network blocks)
├── Enable enhanced monitoring
├── Prepare rollback plan
└── Begin patch development/testing

Hour 8-20: REMEDIATE
├── Test patch in staging
├── Security team validates fix
├── Change approval (emergency CAB)
└── Deploy to production (rolling)

Hour 20-24: VERIFY
├── Confirm vulnerability resolved
├── Monitor for issues
├── Update vulnerability tracker
└── Post-incident review scheduled

Workflow 2: Standard Patch (P1-P2)

# Remediation ticket template
REMEDIATION_TICKET = """
## Vulnerability Remediation

**CVE:** {cve_id}
**Severity:** {severity}
**CVSS:** {cvss_score}
**SLA:** {sla_date}

### Affected Components
{affected_components}

### Root Cause
{root_cause}

### Remediation Steps
1. Update {package} from {current_version} to {fixed_version}
2. Run security regression tests
3. Deploy to staging for validation
4. Security team approval required before production

### Testing Requirements
- [ ] Unit tests pass
- [ ] Integration tests pass
- [ ] Security scan shows vulnerability resolved
- [ ] No new vulnerabilities introduced

### Rollback Plan
{rollback_steps}

### Acceptance Criteria
- Vulnerability scan shows CVE resolved
- No functional regression
- Performance baseline maintained
"""

Workflow 3: Risk Acceptance

## Risk Acceptance Request

**Vulnerability:** CVE-XXXX-XXXXX
**Affected System:** [System Name]
**Requested By:** [Name]
**Date:** [Date]

### Business Justification
[Explain why the vulnerability cannot be remediated]

### Compensating Controls
- [ ] Control 1: [Description]
- [ ] Control 2: [Description]
- [ ] Control 3: [Description]

### Residual Risk Assessment
- **Likelihood:** [High/Medium/Low]
- **Impact:** [High/Medium/Low]
- **Residual Risk:** [Critical/High/Medium/Low]

### Review Schedule
- Next review date: [Date]
- Review frequency: [Monthly/Quarterly]

### Approvals
- [ ] Security Team Lead
- [ ] Engineering Manager
- [ ] CISO
- [ ] Business Owner

Dependency Scanning

Automated Scanning Pipeline

# .github/workflows/security-scan.yml
name: Security Scan

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]
  schedule:
    - cron: '0 6 * * *'  # Daily at 6 AM

jobs:
  dependency-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Run Snyk vulnerability scan
        uses: snyk/actions/node@master
        env:
          SNYK_TOKEN: ${{ secrets.SNYK_TOKEN }}
        with:
          args: --severity-threshold=high

      - name: Run npm audit
        run: npm audit --audit-level=high

      - name: Run Trivy filesystem scan
        uses: aquasecurity/trivy-action@master
        with:
          scan-type: 'fs'
          scan-ref: '.'
          severity: 'CRITICAL,HIGH'
          exit-code: '1'

  sast-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Run Semgrep
        uses: returntocorp/semgrep-action@v1
        with:
          config: >-
            p/security-audit
            p/secrets
            p/owasp-top-ten

Manual Dependency Review

# Node.js - Check for vulnerabilities
npm audit
npm audit --json > audit-report.json

# Python - Check for vulnerabilities
pip-audit
safety check -r requirements.txt

# Go - Check for vulnerabilities
govulncheck ./...

# Container images
trivy image myapp:latest
grype myapp:latest

Dependency Update Strategy

Security Incident Response

Incident Severity Levels

Incident Response Checklist

## Incident Response Checklist

### 1. DETECT & IDENTIFY (0-15 min)
- [ ] Alert received and acknowledged
- [ ] Initial severity assessment
- [ ] Incident commander assigned
- [ ] Communication channel established

### 2. CONTAIN (15-60 min)
- [ ] Affected systems identified
- [ ] Network isolation if needed
- [ ] Credentials rotated if compromised
- [ ] Preserve evidence (logs, memory dumps)

### 3. ERADICATE (1-4 hours)
- [ ] Root cause identified
- [ ] Malware/backdoors removed
- [ ] Vulnerabilities patched
- [ ] Systems hardened

### 4. RECOVER (4-24 hours)
- [ ] Systems restored from clean backup
- [ ] Services brought back online
- [ ] Enhanced monitoring enabled
- [ ] User access restored

### 5. POST-INCIDENT (24-72 hours)
- [ ] Incident timeline documented
- [ ] Root cause analysis complete
- [ ] Lessons learned documented
- [ ] Preventive measures implemented
- [ ] Report to stakeholders

Quick Reference

Vulnerability Response SLAs

Common Vulnerability Databases

Remediation Priority Formula

Priority Score = (CVSS × Exposure × Business_Impact) / Compensating_Controls

Where:
- CVSS: 0-10 (from NVD)
- Exposure: 1.0 (internal) to 2.0 (internet-facing)
- Business_Impact: 1.0 (low) to 2.0 (critical)
- Compensating_Controls: 1.0 (none) to 0.5 (multiple controls)