./browse/prompts

# mTLS Configuration Comprehensive guide to implementing mutual TLS for zero-trust service mesh communication. ## When to Use This Skill - Implementing zero-trust networking - Securing service-to-service communication - Certificate rotation and management - Debugging TLS handshake issues - Compliance requirements (PCI-DSS, HIPAA) - Multi-cluster secure communication ## Core Concepts ### 1. mTLS Flow ``` ┌─────────┐ ┌─────────┐ │ Service │ │ Service │ │ A │ │ B │ └────┬────┘ └────┬────┘ │ │ ┌────┴────┐ TLS Handshake ┌────┴────┐ │ Proxy │◄───────────────────────────►│ Proxy │ │(Sidecar)│ 1. ClientHello │(Sidecar)│ │ │ 2. ServerHello + Cert │ │ │ │ 3. Client Cert │ │ │ │ 4. Verify Both Certs │ │ │ │ 5. Encrypted Channel │ │ └─────────┘ └─────────┘ ``` ### 2. Certificate Hierarchy ``` Root CA (Self-signed, long-lived) │ ├── Intermediate CA (Cluster-level) │ │ │ ├── Workload Cert (Service A) │ └── Workload Cert (Service B) │ └── Intermediate CA (Multi-cluster) │ └── Cross-cluster certs ``` ## Templates ### Template 1: Istio mTLS (Strict Mode) ```yaml # Enable strict mTLS mesh-wide apiVersion: security.istio.io/v1beta1 kind: PeerAuthentication metadata: name: default namespace: istio-system spec: mtls: mode: STRICT --- # Namespace-level override (permissive for migration) apiVersion: security.istio.io/v1beta1 kind: PeerAuthentication metadata: name: default namespace: legacy-namespace spec: mtls: mode: PERMISSIVE --- # Workload-specific policy apiVersion: security.istio.io/v1beta1 kind: PeerAuthentication metadata: name: payment-service namespace: production spec: selector: matchLabels: app: payment-service mtls: mode: STRICT portLevelMtls: 8080: mode: STRICT 9090: mode: DISABLE # Metrics port, no mTLS ``` ### Template 2: Istio Destination Rule for mTLS ```yaml apiVersion: networking.istio.io/v1beta1 kind: DestinationRule metadata: name: default namespace: istio-system spec: host: "*.local" trafficPolicy: tls: mode: ISTIO_MUTUAL --- # TLS to external service apiVersion: networking.istio.io/v1beta1 kind: DestinationRule metadata: name: external-api spec: host: api.external.com trafficPolicy: tls: mode: SIMPLE caCertificates: /etc/certs/external-ca.pem --- # Mutual TLS to external service apiVersion: networking.istio.io/v1beta1 kind: DestinationRule metadata: name: partner-api spec: host: api.partner.com trafficPolicy: tls: mode: MUTUAL clientCertificate: /etc/certs/client.pem privateKey: /etc/certs/client-key.pem caCertificates: /etc/certs/partner-ca.pem ``` ### Template 3: Cert-Manager with Istio ```yaml # Install cert-manager issuer for Istio apiVersion: cert-manager.io/v1 kind: ClusterIssuer metadata: name: istio-ca spec: ca: secretName: istio-ca-secret --- # Create Istio CA secret apiVersion: v1 kind: Secret metadata: name: istio-ca-secret namespace: cert-manager type: kubernetes.io/tls data: tls.crt: <base64-encoded-ca-cert> tls.key: <base64-encoded-ca-key> --- # Certificate for workload apiVersion: cert-manager.io/v1 kind: Certificate metadata: name: my-service-cert namespace: my-namespace spec: secretName: my-service-tls duration: 24h renewBefore: 8h issuerRef: name: istio-ca kind: ClusterIssuer commonName: my-service.my-namespace.svc.cluster.local dnsNames: - my-service - my-service.my-namespace - my-service.my-namespace.svc - my-service.my-namespace.svc.cluster.local usages: - server auth - client auth ``` ### Template 4: SPIFFE/SPIRE Integration ```yaml # SPIRE Server configuration apiVersion: v1 kind: ConfigMap metadata: name: spire-server namespace: spire data: server.conf: | server { bind_address = "0.0.0.0" bind_port = "8081" trust_domain = "example.org" data_dir = "/run/spire/data" log_level = "INFO" ca_ttl = "168h" default_x509_svid_ttl = "1h" } plugins { DataStore "sql" { plugin_data { database_type = "sqlite3" connection_string = "/run/spire/data/datastore.sqlite3" } } NodeAttestor "k8s_psat" { plugin_data { clusters = { "demo-cluster" = { service_account_allow_list = ["spire:spire-agent"] } } } } KeyManager "memory" { plugin_data {} } UpstreamAuthority "disk" { plugin_data { key_file_path = "/run/spire/secrets/bootstrap.key" cert_file_path = "/run/spire/secrets/bootstrap.crt" } } } --- # SPIRE Agent DaemonSet (abbreviated) apiVersion: apps/v1 kind: DaemonSet metadata: name: spire-agent namespace: spire spec: selector: matchLabels: app: spire-agent template: spec: containers: - name: spire-agent image: ghcr.io/spiffe/spire-agent:1.8.0 volumeMounts: - name: spire-agent-socket mountPath: /run/spire/sockets volumes: - name: spire-agent-socket hostPath: path: /run/spire/sockets type: DirectoryOrCreate ``` ### Template 5: Linkerd mTLS (Automatic) ```yaml # Linkerd enables mTLS automatically # Verify with: # linkerd viz edges deployment -n my-namespace # For external services without mTLS apiVersion: policy.linkerd.io/v1beta1 kind: Server metadata: name: external-api namespace: my-namespace spec: podSelector: matchLabels: app: my-app port: external-api proxyProtocol: HTTP/1 # or TLS for passthrough --- # Skip TLS for specific port apiVersion: v1 kind: Service metadata: name: my-service annotations: config.linkerd.io/skip-outbound-ports: "3306" # MySQL ``` ## Certificate Rotation ```bash # Istio - Check certificate expiry istioctl proxy-config secret deploy/my-app -o json | \ jq '.dynamicActiveSecrets[0].secret.tlsCertificate.certificateChain.inlineBytes' | \ tr -d '"' | base64 -d | openssl x509 -text -noout # Force certificate rotation kubectl rollout restart deployment/my-app # Check Linkerd identity linkerd identity -n my-namespace ``` ## Debugging mTLS Issues ```bash # Istio - Check if mTLS is enabled istioctl authn tls-check my-service.my-namespace.svc.cluster.local # Verify peer authentication kubectl get peerauthentication --all-namespaces # Check destination rules kubectl get destinationrule --all-namespaces # Debug TLS handshake istioctl proxy-config log deploy/my-app --level debug kubectl logs deploy/my-app -c istio-proxy | grep -i tls # Linkerd - Check mTLS status linkerd viz edges deployment -n my-namespace linkerd viz tap deploy/my-app --to deploy/my-backend ``` ## Best Practices ### Do's - **Start with PERMISSIVE** - Migrate gradually to STRICT - **Monitor certificate expiry** - Set up alerts - **Use short-lived certs** - 24h or less for workloads - **Rotate CA periodically** - Plan for CA rotation - **Log TLS errors** - For debugging and audit ### Don'ts - **Don't disable mTLS** - For convenience in production - **Don't ignore cert expiry** - Automate rotation - **Don't use self-signed certs** - Use proper CA hierarchy - **Don't skip verification** - Verify the full chain ## Resources - [Istio Security](https://istio.io/latest/docs/concepts/security/) - [SPIFFE/SPIRE](https://spiffe.io/) - [cert-manager](https://cert-manager.io/) - [Zero Trust Architecture (NIST)](https://www.nist.gov/publications/zero-trust-architecture)

# PCI Compliance Master PCI DSS (Payment Card Industry Data Security Standard) compliance for secure payment processing and handling of cardholder data. ## When to Use This Skill - Building payment processing systems - Handling credit card information - Implementing secure payment flows - Conducting PCI compliance audits - Reducing PCI compliance scope - Implementing tokenization and encryption - Preparing for PCI DSS assessments ## PCI DSS Requirements (12 Core Requirements) ### Build and Maintain Secure Network 1. Install and maintain firewall configuration 2. Don't use vendor-supplied defaults for passwords ### Protect Cardholder Data 3. Protect stored cardholder data 4. Encrypt transmission of cardholder data across public networks ### Maintain Vulnerability Management 5. Protect systems against malware 6. Develop and maintain secure systems and applications ### Implement Strong Access Control 7. Restrict access to cardholder data by business need-to-know 8. Identify and authenticate access to system components 9. Restrict physical access to cardholder data ### Monitor and Test Networks 10. Track and monitor all access to network resources and cardholder data 11. Regularly test security systems and processes ### Maintain Information Security Policy 12. Maintain a policy that addresses information security ## Compliance Levels **Level 1**: > 6 million transactions/year (annual ROC required) **Level 2**: 1-6 million transactions/year (annual SAQ) **Level 3**: 20,000-1 million e-commerce transactions/year **Level 4**: < 20,000 e-commerce or < 1 million total transactions ## Data Minimization (Never Store) ```python # NEVER STORE THESE PROHIBITED_DATA = { 'full_track_data': 'Magnetic stripe data', 'cvv': 'Card verification code/value', 'pin': 'PIN or PIN block' } # CAN STORE (if encrypted) ALLOWED_DATA = { 'pan': 'Primary Account Number (card number)', 'cardholder_name': 'Name on card', 'expiration_date': 'Card expiration', 'service_code': 'Service code' } class PaymentData: """Safe payment data handling.""" def __init__(self): self.prohibited_fields = ['cvv', 'cvv2', 'cvc', 'pin'] def sanitize_log(self, data): """Remove sensitive data from logs.""" sanitized = data.copy() # Mask PAN if 'card_number' in sanitized: card = sanitized['card_number'] sanitized['card_number'] = f"{card[:6]}{'*' * (len(card) - 10)}{card[-4:]}" # Remove prohibited data for field in self.prohibited_fields: sanitized.pop(field, None) return sanitized def validate_no_prohibited_storage(self, data): """Ensure no prohibited data is being stored.""" for field in self.prohibited_fields: if field in data: raise SecurityError(f"Attempting to store prohibited field: {field}") ``` ## Tokenization ### Using Payment Processor Tokens ```python import stripe class TokenizedPayment: """Handle payments using tokens (no card data on server).""" @staticmethod def create_payment_method_token(card_details): """Create token from card details (client-side only).""" # THIS SHOULD ONLY BE DONE CLIENT-SIDE WITH STRIPE.JS # NEVER send card details to your server """ // Frontend JavaScript const stripe = Stripe('pk_...'); const {token, error} = await stripe.createToken({ card: { number: '4242424242424242', exp_month: 12, exp_year: 2024, cvc: '123' } }); // Send token.id to server (NOT card details) """ pass @staticmethod def charge_with_token(token_id, amount): """Charge using token (server-side).""" # Your server only sees the token, never the card number stripe.api_key = "sk_..." charge = stripe.Charge.create( amount=amount, currency="usd", source=token_id, # Token instead of card details description="Payment" ) return charge @staticmethod def store_payment_method(customer_id, payment_method_token): """Store payment method as token for future use.""" stripe.Customer.modify( customer_id, source=payment_method_token ) # Store only customer_id and payment_method_id in your database # NEVER store actual card details return { 'customer_id': customer_id, 'has_payment_method': True # DO NOT store: card number, CVV, etc. } ``` ### Custom Tokenization (Advanced) ```python import secrets from cryptography.fernet import Fernet class TokenVault: """Secure token vault for card data (if you must store it).""" def __init__(self, encryption_key): self.cipher = Fernet(encryption_key) self.vault = {} # In production: use encrypted database def tokenize(self, card_data): """Convert card data to token.""" # Generate secure random token token = secrets.token_urlsafe(32) # Encrypt card data encrypted = self.cipher.encrypt(json.dumps(card_data).encode()) # Store token -> encrypted data mapping self.vault[token] = encrypted return token def detokenize(self, token): """Retrieve card data from token.""" encrypted = self.vault.get(token) if not encrypted: raise ValueError("Token not found") # Decrypt decrypted = self.cipher.decrypt(encrypted) return json.loads(decrypted.decode()) def delete_token(self, token): """Remove token from vault.""" self.vault.pop(token, None) ``` ## Encryption ### Data at Rest ```python from cryptography.hazmat.primitives.ciphers.aead import AESGCM import os class EncryptedStorage: """Encrypt data at rest using AES-256-GCM.""" def __init__(self, encryption_key): """Initialize with 256-bit key.""" self.key = encryption_key # Must be 32 bytes def encrypt(self, plaintext): """Encrypt data.""" # Generate random nonce nonce = os.urandom(12) # Encrypt aesgcm = AESGCM(self.key) ciphertext = aesgcm.encrypt(nonce, plaintext.encode(), None) # Return nonce + ciphertext return nonce + ciphertext def decrypt(self, encrypted_data): """Decrypt data.""" # Extract nonce and ciphertext nonce = encrypted_data[:12] ciphertext = encrypted_data[12:] # Decrypt aesgcm = AESGCM(self.key) plaintext = aesgcm.decrypt(nonce, ciphertext, None) return plaintext.decode() # Usage storage = EncryptedStorage(os.urandom(32)) encrypted_pan = storage.encrypt("4242424242424242") # Store encrypted_pan in database ``` ### Data in Transit ```python # Always use TLS 1.2 or higher # Flask/Django example app.config['SESSION_COOKIE_SECURE'] = True # HTTPS only app.config['SESSION_COOKIE_HTTPONLY'] = True app.config['SESSION_COOKIE_SAMESITE'] = 'Strict' # Enforce HTTPS from flask_talisman import Talisman Talisman(app, force_https=True) ``` ## Access Control ```python from functools import wraps from flask import session def require_pci_access(f): """Decorator to restrict access to cardholder data.""" @wraps(f) def decorated_function(*args, **kwargs): user = session.get('user') # Check if user has PCI access role if not user or 'pci_access' not in user.get('roles', []): return {'error': 'Unauthorized access to cardholder data'}, 403 # Log access attempt audit_log( user=user['id'], action='access_cardholder_data', resource=f.__name__ ) return f(*args, **kwargs) return decorated_function @app.route('/api/payment-methods') @require_pci_access def get_payment_methods(): """Retrieve payment methods (restricted access).""" # Only accessible to users with pci_access role pass ``` ## Audit Logging ```python import logging from datetime import datetime class PCIAuditLogger: """PCI-compliant audit logging.""" def __init__(self): self.logger = logging.getLogger('pci_audit') # Configure to write to secure, append-only log def log_access(self, user_id, resource, action, result): """Log access to cardholder data.""" entry = { 'timestamp': datetime.utcnow().isoformat(), 'user_id': user_id, 'resource': resource, 'action': action, 'result': result, 'ip_address': request.remote_addr } self.logger.info(json.dumps(entry)) def log_authentication(self, user_id, success, method): """Log authentication attempt.""" entry = { 'timestamp': datetime.utcnow().isoformat(), 'user_id': user_id, 'event': 'authentication', 'success': success, 'method': method, 'ip_address': request.remote_addr } self.logger.info(json.dumps(entry)) # Usage audit = PCIAuditLogger() audit.log_access(user_id=123, resource='payment_methods', action='read', result='success') ``` ## Security Best Practices ### Input Validation ```python import re def validate_card_number(card_number): """Validate card number format (Luhn algorithm).""" # Remove spaces and dashes card_number = re.sub(r'[\s-]', '', card_number) # Check if all digits if not card_number.isdigit(): return False # Luhn algorithm def luhn_checksum(card_num): def digits_of(n): return [int(d) for d in str(n)] digits = digits_of(card_num) odd_digits = digits[-1::-2] even_digits = digits[-2::-2] checksum = sum(odd_digits) for d in even_digits: checksum += sum(digits_of(d * 2)) return checksum % 10 return luhn_checksum(card_number) == 0 def sanitize_input(user_input): """Sanitize user input to prevent injection.""" # Remove special characters # Validate against expected format # Escape for database queries pass ``` ## PCI DSS SAQ (Self-Assessment Questionnaire) ### SAQ A (Least Requirements) - E-commerce using hosted payment page - No card data on your systems - ~20 questions ### SAQ A-EP - E-commerce with embedded payment form - Uses JavaScript to handle card data - ~180 questions ### SAQ D (Most Requirements) - Store, process, or transmit card data - Full PCI DSS requirements - ~300 questions ## Compliance Checklist ```python PCI_COMPLIANCE_CHECKLIST = { 'network_security': [ 'Firewall configured and maintained', 'No vendor default passwords', 'Network segmentation implemented' ], 'data_protection': [ 'No storage of CVV, track data, or PIN', 'PAN encrypted when stored', 'PAN masked when displayed', 'Encryption keys properly managed' ], 'vulnerability_management': [ 'Anti-virus installed and updated', 'Secure development practices', 'Regular security patches', 'Vulnerability scanning performed' ], 'access_control': [ 'Access restricted by role', 'Unique IDs for all users', 'Multi-factor authentication', 'Physical security measures' ], 'monitoring': [ 'Audit logs enabled', 'Log review process', 'File integrity monitoring', 'Regular security testing' ], 'policy': [ 'Security policy documented', 'Risk assessment performed', 'Security awareness training', 'Incident response plan' ] } ``` ## Resources - **references/data-minimization.md**: Never store prohibited data - **references/tokenization.md**: Tokenization strategies - **references/encryption.md**: Encryption requirements - **references/access-control.md**: Role-based access - **references/audit-logging.md**: Comprehensive logging - **assets/pci-compliance-checklist.md**: Complete checklist - **assets/encrypted-storage.py**: Encryption utilities - **scripts/audit-payment-system.sh**: Compliance audit script ## Common Violations 1. **Storing CVV**: Never store card verification codes 2. **Unencrypted PAN**: Card numbers must be encrypted at rest 3. **Weak Encryption**: Use AES-256 or equivalent 4. **No Access Controls**: Restrict who can access cardholder data 5. **Missing Audit Logs**: Must log all access to payment data 6. **Insecure Transmission**: Always use TLS 1.2+ 7. **Default Passwords**: Change all default credentials 8. **No Security Testing**: Regular penetration testing required ## Reducing PCI Scope 1. **Use Hosted Payments**: Stripe Checkout, PayPal, etc. 2. **Tokenization**: Replace card data with tokens 3. **Network Segmentation**: Isolate cardholder data environment 4. **Outsource**: Use PCI-compliant payment processors 5. **No Storage**: Never store full card details By minimizing systems that touch card data, you reduce compliance burden significantly.

# Stripe Integration Master Stripe payment processing integration for robust, PCI-compliant payment flows including checkout, subscriptions, webhooks, and refunds. ## When to Use This Skill - Implementing payment processing in web/mobile applications - Setting up subscription billing systems - Handling one-time payments and recurring charges - Processing refunds and disputes - Managing customer payment methods - Implementing SCA (Strong Customer Authentication) for European payments - Building marketplace payment flows with Stripe Connect ## Core Concepts ### 1. Payment Flows **Checkout Sessions** - Recommended for most integrations - Supports all UI paths: - Stripe-hosted checkout page - Embedded checkout form - Custom UI with Elements (Payment Element, Express Checkout Element) using `ui_mode='custom'` - Provides built-in checkout capabilities (line items, discounts, tax, shipping, address collection, saved payment methods, and checkout lifecycle events) - Lower integration and maintenance burden than Payment Intents **Payment Intents (Bespoke control)** - You calculate the final amount with taxes, discounts, subscriptions, and currency conversion yourself. - More complex implementation and long-term maintenance burden - Requires Stripe.js for PCI compliance **Setup Intents (Save Payment Methods)** - Collect payment method without charging - Used for subscriptions and future payments - Requires customer confirmation ### 2. Webhooks **Critical Events:** - `payment_intent.succeeded`: Payment completed - `payment_intent.payment_failed`: Payment failed - `customer.subscription.updated`: Subscription changed - `customer.subscription.deleted`: Subscription canceled - `charge.refunded`: Refund processed - `invoice.payment_succeeded`: Subscription payment successful ### 3. Subscriptions **Components:** - **Product**: What you're selling - **Price**: How much and how often - **Subscription**: Customer's recurring payment - **Invoice**: Generated for each billing cycle ### 4. Customer Management - Create and manage customer records - Store multiple payment methods - Track customer metadata - Manage billing details ## Quick Start ```python import stripe stripe.api_key = "sk_test_..." # Create a checkout session session = stripe.checkout.Session.create( line_items=[{ 'price_data': { 'currency': 'usd', 'product_data': { 'name': 'Premium Subscription', }, 'unit_amount': 2000, # $20.00 'recurring': { 'interval': 'month', }, }, 'quantity': 1, }], mode='subscription', success_url='https://yourdomain.com/success?session_id={CHECKOUT_SESSION_ID}', cancel_url='https://yourdomain.com/cancel' ) # Redirect user to session.url print(session.url) ``` ## Payment Implementation Patterns ### Pattern 1: One-Time Payment (Hosted Checkout) ```python def create_checkout_session(amount, currency='usd'): """Create a one-time payment checkout session.""" try: session = stripe.checkout.Session.create( line_items=[{ 'price_data': { 'currency': currency, 'product_data': { 'name': 'Blue T-shirt', 'images': ['https://example.com/product.jpg'], }, 'unit_amount': amount, # Amount in cents }, 'quantity': 1, }], mode='payment', success_url='https://yourdomain.com/success?session_id={CHECKOUT_SESSION_ID}', cancel_url='https://yourdomain.com/cancel', metadata={ 'order_id': 'order_123', 'user_id': 'user_456' } ) return session except stripe.error.StripeError as e: # Handle error print(f"Stripe error: {e.user_message}") raise ``` ### Pattern 2: Elements with Checkout Sessions ```python def create_checkout_session_for_elements(amount, currency='usd'): """Create a checkout session configured for Payment Element.""" session = stripe.checkout.Session.create( mode='payment', ui_mode='custom', line_items=[{ 'price_data': { 'currency': currency, 'product_data': {'name': 'Blue T-shirt'}, 'unit_amount': amount, }, 'quantity': 1, }], return_url='https://yourdomain.com/complete?session_id={CHECKOUT_SESSION_ID}' ) return session.client_secret # Send to frontend ``` ```javascript const stripe = Stripe("pk_test_..."); const appearance = { theme: "stripe" }; const checkout = stripe.initCheckout({ clientSecret, elementsOptions: { appearance }, }); const loadActionsResult = await checkout.loadActions(); if (loadActionsResult.type === "success") { const { actions } = loadActionsResult; const session = actions.getSession(); const button = document.getElementById("pay-button"); const checkoutContainer = document.getElementById("checkout-container"); const emailInput = document.getElementById("email"); const emailErrors = document.getElementById("email-errors"); const errors = document.getElementById("confirm-errors"); // Display a formatted string representing the total amount checkoutContainer.append(`Total: ${session.total.total.amount}`); // Mount Payment Element const paymentElement = checkout.createPaymentElement(); paymentElement.mount("#payment-element"); // Store email for submission emailInput.addEventListener("blur", () => { actions.updateEmail(emailInput.value).then((result) => { if (result.error) emailErrors.textContent = result.error.message; }); }); // Handle form submission button.addEventListener("click", () => { actions.confirm().then((result) => { if (result.type === "error") errors.textContent = result.error.message; }); }); } ``` ### Pattern 3: Elements with Payment Intents Pattern 2 (Elements with Checkout Sessions) is Stripe's recommended approach, but you can also use Payment Intents as an alternative. ```python def create_payment_intent(amount, currency='usd', customer_id=None): """Create a payment intent for bespoke checkout UI with Payment Element.""" intent = stripe.PaymentIntent.create( amount=amount, currency=currency, customer=customer_id, automatic_payment_methods={ 'enabled': True, }, metadata={ 'integration_check': 'accept_a_payment' } ) return intent.client_secret # Send to frontend ``` ```javascript // Mount Payment Element and confirm via Payment Intents const stripe = Stripe("pk_test_..."); const appearance = { theme: "stripe" }; const elements = stripe.elements({ appearance, clientSecret }); const paymentElement = elements.create("payment"); paymentElement.mount("#payment-element"); document.getElementById("pay-button").addEventListener("click", async () => { const { error } = await stripe.confirmPayment({ elements, confirmParams: { return_url: "https://yourdomain.com/complete", }, }); if (error) { document.getElementById("errors").textContent = error.message; } }); ``` ### Pattern 4: Subscription Creation ```python def create_subscription(customer_id, price_id): """Create a subscription for a customer.""" try: subscription = stripe.Subscription.create( customer=customer_id, items=[{'price': price_id}], payment_behavior='default_incomplete', payment_settings={'save_default_payment_method': 'on_subscription'}, expand=['latest_invoice.payment_intent'], ) return { 'subscription_id': subscription.id, 'client_secret': subscription.latest_invoice.payment_intent.client_secret } except stripe.error.StripeError as e: print(f"Subscription creation failed: {e}") raise ``` ### Pattern 5: Customer Portal ```python def create_customer_portal_session(customer_id): """Create a portal session for customers to manage subscriptions.""" session = stripe.billing_portal.Session.create( customer=customer_id, return_url='https://yourdomain.com/account', ) return session.url # Redirect customer here ``` ## Webhook Handling ### Secure Webhook Endpoint ```python from flask import Flask, request import stripe app = Flask(__name__) endpoint_secret = 'whsec_...' @app.route('/webhook', methods=['POST']) def webhook(): payload = request.data sig_header = request.headers.get('Stripe-Signature') try: event = stripe.Webhook.construct_event( payload, sig_header, endpoint_secret ) except ValueError: # Invalid payload return 'Invalid payload', 400 except stripe.error.SignatureVerificationError: # Invalid signature return 'Invalid signature', 400 # Handle the event if event['type'] == 'payment_intent.succeeded': payment_intent = event['data']['object'] handle_successful_payment(payment_intent) elif event['type'] == 'payment_intent.payment_failed': payment_intent = event['data']['object'] handle_failed_payment(payment_intent) elif event['type'] == 'customer.subscription.deleted': subscription = event['data']['object'] handle_subscription_canceled(subscription) return 'Success', 200 def handle_successful_payment(payment_intent): """Process successful payment.""" customer_id = payment_intent.get('customer') amount = payment_intent['amount'] metadata = payment_intent.get('metadata', {}) # Update your database # Send confirmation email # Fulfill order print(f"Payment succeeded: {payment_intent['id']}") def handle_failed_payment(payment_intent): """Handle failed payment.""" error = payment_intent.get('last_payment_error', {}) print(f"Payment failed: {error.get('message')}") # Notify customer # Update order status def handle_subscription_canceled(subscription): """Handle subscription cancellation.""" customer_id = subscription['customer'] # Update user access # Send cancellation email print(f"Subscription canceled: {subscription['id']}") ``` ### Webhook Best Practices ```python import hashlib import hmac def verify_webhook_signature(payload, signature, secret): """Manually verify webhook signature.""" expected_sig = hmac.new( secret.encode('utf-8'), payload, hashlib.sha256 ).hexdigest() return hmac.compare_digest(signature, expected_sig) def handle_webhook_idempotently(event_id, handler): """Ensure webhook is processed exactly once.""" # Check if event already processed if is_event_processed(event_id): return # Process event try: handler() mark_event_processed(event_id) except Exception as e: log_error(e) # Stripe will retry failed webhooks raise ``` ## Customer Management ```python def create_customer(email, name, payment_method_id=None): """Create a Stripe customer.""" customer = stripe.Customer.create( email=email, name=name, payment_method=payment_method_id, invoice_settings={ 'default_payment_method': payment_method_id } if payment_method_id else None, metadata={ 'user_id': '12345' } ) return customer def attach_payment_method(customer_id, payment_method_id): """Attach a payment method to a customer.""" stripe.PaymentMethod.attach( payment_method_id, customer=customer_id ) # Set as default stripe.Customer.modify( customer_id, invoice_settings={ 'default_payment_method': payment_method_id } ) def list_customer_payment_methods(customer_id): """List all payment methods for a customer.""" payment_methods = stripe.PaymentMethod.list( customer=customer_id, type='card' ) return payment_methods.data ``` ## Refund Handling ```python def create_refund(payment_intent_id, amount=None, reason=None): """Create a refund.""" refund_params = { 'payment_intent': payment_intent_id } if amount: refund_params['amount'] = amount # Partial refund if reason: refund_params['reason'] = reason # 'duplicate', 'fraudulent', 'requested_by_customer' refund = stripe.Refund.create(**refund_params) return refund def handle_dispute(charge_id, evidence): """Update dispute with evidence.""" stripe.Dispute.modify( charge_id, evidence={ 'customer_name': evidence.get('customer_name'), 'customer_email_address': evidence.get('customer_email'), 'shipping_documentation': evidence.get('shipping_proof'), 'customer_communication': evidence.get('communication'), } ) ``` ## Testing ```python # Use test mode keys stripe.api_key = "sk_test_..." # Test card numbers TEST_CARDS = { 'success': '4242424242424242', 'declined': '4000000000000002', '3d_secure': '4000002500003155', 'insufficient_funds': '4000000000009995' } def test_payment_flow(): """Test complete payment flow.""" # Create test customer customer = stripe.Customer.create( email="test@example.com" ) # Create payment intent intent = stripe.PaymentIntent.create( amount=1000, automatic_payment_methods={ 'enabled': True }, currency='usd', customer=customer.id ) # Confirm with test card confirmed = stripe.PaymentIntent.confirm( intent.id, payment_method='pm_card_visa' # Test payment method ) assert confirmed.status == 'succeeded' ``` ## Resources - **references/checkout-flows.md**: Detailed checkout implementation - **references/webhook-handling.md**: Webhook security and processing - **references/subscription-management.md**: Subscription lifecycle - **references/customer-management.md**: Customer and payment method handling - **references/invoice-generation.md**: Invoicing and billing - **assets/stripe-client.py**: Production-ready Stripe client wrapper - **assets/webhook-handler.py**: Complete webhook processor - **assets/checkout-config.json**: Checkout configuration templates ## Best Practices 1. **Always Use Webhooks**: Don't rely solely on client-side confirmation 2. **Idempotency**: Handle webhook events idempotently 3. **Error Handling**: Gracefully handle all Stripe errors 4. **Test Mode**: Thoroughly test with test keys before production 5. **Metadata**: Use metadata to link Stripe objects to your database 6. **Monitoring**: Track payment success rates and errors 7. **PCI Compliance**: Never handle raw card data on your server 8. **SCA Ready**: Implement 3D Secure for European payments ## Common Pitfalls - **Not Verifying Webhooks**: Always verify webhook signatures - **Missing Webhook Events**: Handle all relevant webhook events - **Hardcoded Amounts**: Use cents/smallest currency unit - **No Retry Logic**: Implement retries for API calls - **Ignoring Test Mode**: Test all edge cases with test cards

# SAST Configuration Static Application Security Testing (SAST) tool setup, configuration, and custom rule creation for comprehensive security scanning across multiple programming languages. ## Overview This skill provides comprehensive guidance for setting up and configuring SAST tools including Semgrep, SonarQube, and CodeQL. Use this skill when you need to: - Set up SAST scanning in CI/CD pipelines - Create custom security rules for your codebase - Configure quality gates and compliance policies - Optimize scan performance and reduce false positives - Integrate multiple SAST tools for defense-in-depth ## Core Capabilities ### 1. Semgrep Configuration - Custom rule creation with pattern matching - Language-specific security rules (Python, JavaScript, Go, Java, etc.) - CI/CD integration (GitHub Actions, GitLab CI, Jenkins) - False positive tuning and rule optimization - Organizational policy enforcement ### 2. SonarQube Setup - Quality gate configuration - Security hotspot analysis - Code coverage and technical debt tracking - Custom quality profiles for languages - Enterprise integration with LDAP/SAML ### 3. CodeQL Analysis - GitHub Advanced Security integration - Custom query development - Vulnerability variant analysis - Security research workflows - SARIF result processing ## Quick Start ### Initial Assessment 1. Identify primary programming languages in your codebase 2. Determine compliance requirements (PCI-DSS, SOC 2, etc.) 3. Choose SAST tool based on language support and integration needs 4. Review baseline scan to understand current security posture ### Basic Setup ```bash # Semgrep quick start pip install semgrep semgrep --config=auto --error # SonarQube with Docker docker run -d --name sonarqube -p 9000:9000 sonarqube:latest # CodeQL CLI setup gh extension install github/gh-codeql codeql database create mydb --language=python ``` ## Reference Documentation - [Semgrep Rule Creation](references/semgrep-rules.md) - Pattern-based security rule development - [SonarQube Configuration](references/sonarqube-config.md) - Quality gates and profiles - [CodeQL Setup Guide](references/codeql-setup.md) - Query development and workflows ## Templates & Assets - [semgrep-config.yml](assets/semgrep-config.yml) - Production-ready Semgrep configuration - [sonarqube-settings.xml](assets/sonarqube-settings.xml) - SonarQube quality profile template - [run-sast.sh](scripts/run-sast.sh) - Automated SAST execution script ## Integration Patterns ### CI/CD Pipeline Integration ```yaml # GitHub Actions example - name: Run Semgrep uses: returntocorp/semgrep-action@v1 with: config: >- p/security-audit p/owasp-top-ten ``` ### Pre-commit Hook ```bash # .pre-commit-config.yaml - repo: https://github.com/returntocorp/semgrep rev: v1.45.0 hooks: - id: semgrep args: ['--config=auto', '--error'] ``` ## Best Practices 1. **Start with Baseline** - Run initial scan to establish security baseline - Prioritize critical and high severity findings - Create remediation roadmap 2. **Incremental Adoption** - Begin with security-focused rules - Gradually add code quality rules - Implement blocking only for critical issues 3. **False Positive Management** - Document legitimate suppressions - Create allow lists for known safe patterns - Regularly review suppressed findings 4. **Performance Optimization** - Exclude test files and generated code - Use incremental scanning for large codebases - Cache scan results in CI/CD 5. **Team Enablement** - Provide security training for developers - Create internal documentation for common patterns - Establish security champions program ## Common Use Cases ### New Project Setup ```bash ./scripts/run-sast.sh --setup --language python --tools semgrep,sonarqube ``` ### Custom Rule Development ```yaml # See references/semgrep-rules.md for detailed examples rules: - id: hardcoded-jwt-secret pattern: jwt.encode($DATA, "...", ...) message: JWT secret should not be hardcoded severity: ERROR ``` ### Compliance Scanning ```bash # PCI-DSS focused scan semgrep --config p/pci-dss --json -o pci-scan-results.json ``` ## Troubleshooting ### High False Positive Rate - Review and tune rule sensitivity - Add path filters to exclude test files - Use nostmt metadata for noisy patterns - Create organization-specific rule exceptions ### Performance Issues - Enable incremental scanning - Parallelize scans across modules - Optimize rule patterns for efficiency - Cache dependencies and scan results ### Integration Failures - Verify API tokens and credentials - Check network connectivity and proxy settings - Review SARIF output format compatibility - Validate CI/CD runner permissions ## Related Skills - [OWASP Top 10 Checklist](../owasp-top10-checklist/SKILL.md) - [Container Security](../container-security/SKILL.md) - [Dependency Scanning](../dependency-scanning/SKILL.md) ## Tool Comparison | Tool | Best For | Language Support | Cost | Integration | | --------- | ------------------------ | ---------------- | --------------- | ------------- | | Semgrep | Custom rules, fast scans | 30+ languages | Free/Enterprise | Excellent | | SonarQube | Code quality + security | 25+ languages | Free/Commercial | Good | | CodeQL | Deep analysis, research | 10+ languages | Free (OSS) | GitHub native | ## Next Steps 1. Complete initial SAST tool setup 2. Run baseline security scan 3. Create custom rules for organization-specific patterns 4. Integrate into CI/CD pipeline 5. Establish security gate policies 6. Train development team on findings and remediation

# Security Requirement Extraction Transform threat analysis into actionable security requirements. ## When to Use This Skill - Converting threat models to requirements - Writing security user stories - Creating security test cases - Building security acceptance criteria - Compliance requirement mapping - Security architecture documentation ## Core Concepts ### 1. Requirement Categories ``` Business Requirements → Security Requirements → Technical Controls ↓ ↓ ↓ "Protect customer "Encrypt PII at rest" "AES-256 encryption data" with KMS key rotation" ``` ### 2. Security Requirement Types | Type | Focus | Example | | ------------------ | ----------------------- | ------------------------------------- | | **Functional** | What system must do | "System must authenticate users" | | **Non-functional** | How system must perform | "Authentication must complete in <2s" | | **Constraint** | Limitations imposed | "Must use approved crypto libraries" | ### 3. Requirement Attributes | Attribute | Description | | ---------------- | --------------------------- | | **Traceability** | Links to threats/compliance | | **Testability** | Can be verified | | **Priority** | Business importance | | **Risk Level** | Impact if not met | ## Templates ### Template 1: Security Requirement Model ```python from dataclasses import dataclass, field from enum import Enum from typing import List, Dict, Optional, Set from datetime import datetime class RequirementType(Enum): FUNCTIONAL = "functional" NON_FUNCTIONAL = "non_functional" CONSTRAINT = "constraint" class Priority(Enum): CRITICAL = 1 HIGH = 2 MEDIUM = 3 LOW = 4 class SecurityDomain(Enum): AUTHENTICATION = "authentication" AUTHORIZATION = "authorization" DATA_PROTECTION = "data_protection" AUDIT_LOGGING = "audit_logging" INPUT_VALIDATION = "input_validation" ERROR_HANDLING = "error_handling" SESSION_MANAGEMENT = "session_management" CRYPTOGRAPHY = "cryptography" NETWORK_SECURITY = "network_security" AVAILABILITY = "availability" class ComplianceFramework(Enum): PCI_DSS = "pci_dss" HIPAA = "hipaa" GDPR = "gdpr" SOC2 = "soc2" NIST_CSF = "nist_csf" ISO_27001 = "iso_27001" OWASP = "owasp" @dataclass class SecurityRequirement: id: str title: str description: str req_type: RequirementType domain: SecurityDomain priority: Priority rationale: str = "" acceptance_criteria: List[str] = field(default_factory=list) test_cases: List[str] = field(default_factory=list) threat_refs: List[str] = field(default_factory=list) compliance_refs: List[str] = field(default_factory=list) dependencies: List[str] = field(default_factory=list) status: str = "draft" owner: str = "" created_date: datetime = field(default_factory=datetime.now) def to_user_story(self) -> str: """Convert to user story format.""" return f""" **{self.id}: {self.title}** As a security-conscious system, I need to {self.description.lower()}, So that {self.rationale.lower()}. **Acceptance Criteria:** {chr(10).join(f'- [ ] {ac}' for ac in self.acceptance_criteria)} **Priority:** {self.priority.name} **Domain:** {self.domain.value} **Threat References:** {', '.join(self.threat_refs)} """ def to_test_spec(self) -> str: """Convert to test specification.""" return f""" ## Test Specification: {self.id} ### Requirement {self.description} ### Test Cases {chr(10).join(f'{i+1}. {tc}' for i, tc in enumerate(self.test_cases))} ### Acceptance Criteria Verification {chr(10).join(f'- {ac}' for ac in self.acceptance_criteria)} """ @dataclass class RequirementSet: name: str version: str requirements: List[SecurityRequirement] = field(default_factory=list) def add(self, req: SecurityRequirement) -> None: self.requirements.append(req) def get_by_domain(self, domain: SecurityDomain) -> List[SecurityRequirement]: return [r for r in self.requirements if r.domain == domain] def get_by_priority(self, priority: Priority) -> List[SecurityRequirement]: return [r for r in self.requirements if r.priority == priority] def get_by_threat(self, threat_id: str) -> List[SecurityRequirement]: return [r for r in self.requirements if threat_id in r.threat_refs] def get_critical_requirements(self) -> List[SecurityRequirement]: return [r for r in self.requirements if r.priority == Priority.CRITICAL] def export_markdown(self) -> str: """Export all requirements as markdown.""" lines = [f"# Security Requirements: {self.name}\n"] lines.append(f"Version: {self.version}\n") for domain in SecurityDomain: domain_reqs = self.get_by_domain(domain) if domain_reqs: lines.append(f"\n## {domain.value.replace('_', ' ').title()}\n") for req in domain_reqs: lines.append(req.to_user_story()) return "\n".join(lines) def traceability_matrix(self) -> Dict[str, List[str]]: """Generate threat-to-requirement traceability.""" matrix = {} for req in self.requirements: for threat_id in req.threat_refs: if threat_id not in matrix: matrix[threat_id] = [] matrix[threat_id].append(req.id) return matrix ``` ### Template 2: Threat-to-Requirement Extractor ```python from dataclasses import dataclass from typing import List, Dict, Tuple @dataclass class ThreatInput: id: str category: str # STRIDE category title: str description: str target: str impact: str likelihood: str class RequirementExtractor: """Extract security requirements from threats.""" # Mapping of STRIDE categories to security domains and requirement patterns STRIDE_MAPPINGS = { "SPOOFING": { "domains": [SecurityDomain.AUTHENTICATION, SecurityDomain.SESSION_MANAGEMENT], "patterns": [ ("Implement strong authentication for {target}", "Ensure {target} authenticates all users before granting access"), ("Validate identity tokens for {target}", "All authentication tokens must be cryptographically verified"), ("Implement session management for {target}", "Sessions must be securely managed with proper expiration"), ] }, "TAMPERING": { "domains": [SecurityDomain.INPUT_VALIDATION, SecurityDomain.DATA_PROTECTION], "patterns": [ ("Validate all input to {target}", "All input must be validated against expected formats"), ("Implement integrity checks for {target}", "Data integrity must be verified using cryptographic signatures"), ("Protect {target} from modification", "Implement controls to prevent unauthorized data modification"), ] }, "REPUDIATION": { "domains": [SecurityDomain.AUDIT_LOGGING], "patterns": [ ("Log all security events for {target}", "Security-relevant events must be logged for audit purposes"), ("Implement non-repudiation for {target}", "Critical actions must have cryptographic proof of origin"), ("Protect audit logs for {target}", "Audit logs must be tamper-evident and protected"), ] }, "INFORMATION_DISCLOSURE": { "domains": [SecurityDomain.DATA_PROTECTION, SecurityDomain.CRYPTOGRAPHY], "patterns": [ ("Encrypt sensitive data in {target}", "Sensitive data must be encrypted at rest and in transit"), ("Implement access controls for {target}", "Data access must be restricted based on need-to-know"), ("Prevent information leakage from {target}", "Error messages and logs must not expose sensitive information"), ] }, "DENIAL_OF_SERVICE": { "domains": [SecurityDomain.AVAILABILITY, SecurityDomain.INPUT_VALIDATION], "patterns": [ ("Implement rate limiting for {target}", "Requests must be rate-limited to prevent resource exhaustion"), ("Ensure availability of {target}", "System must remain available under high load conditions"), ("Implement resource quotas for {target}", "Resource consumption must be bounded and monitored"), ] }, "ELEVATION_OF_PRIVILEGE": { "domains": [SecurityDomain.AUTHORIZATION], "patterns": [ ("Enforce authorization for {target}", "All actions must be authorized based on user permissions"), ("Implement least privilege for {target}", "Users must only have minimum necessary permissions"), ("Validate permissions for {target}", "Permission checks must be performed server-side"), ] }, } def extract_requirements( self, threats: List[ThreatInput], project_name: str ) -> RequirementSet: """Extract security requirements from threats.""" req_set = RequirementSet( name=f"{project_name} Security Requirements", version="1.0" ) req_counter = 1 for threat in threats: reqs = self._threat_to_requirements(threat, req_counter) for req in reqs: req_set.add(req) req_counter += len(reqs) return req_set def _threat_to_requirements( self, threat: ThreatInput, start_id: int ) -> List[SecurityRequirement]: """Convert a single threat to requirements.""" requirements = [] mapping = self.STRIDE_MAPPINGS.get(threat.category, {}) domains = mapping.get("domains", []) patterns = mapping.get("patterns", []) priority = self._calculate_priority(threat.impact, threat.likelihood) for i, (title_pattern, desc_pattern) in enumerate(patterns): req = SecurityRequirement( id=f"SR-{start_id + i:03d}", title=title_pattern.format(target=threat.target), description=desc_pattern.format(target=threat.target), req_type=RequirementType.FUNCTIONAL, domain=domains[i % len(domains)] if domains else SecurityDomain.DATA_PROTECTION, priority=priority, rationale=f"Mitigates threat: {threat.title}", threat_refs=[threat.id], acceptance_criteria=self._generate_acceptance_criteria( threat.category, threat.target ), test_cases=self._generate_test_cases( threat.category, threat.target ) ) requirements.append(req) return requirements def _calculate_priority(self, impact: str, likelihood: str) -> Priority: """Calculate requirement priority from threat attributes.""" score_map = {"LOW": 1, "MEDIUM": 2, "HIGH": 3, "CRITICAL": 4} impact_score = score_map.get(impact.upper(), 2) likelihood_score = score_map.get(likelihood.upper(), 2) combined = impact_score * likelihood_score if combined >= 12: return Priority.CRITICAL elif combined >= 6: return Priority.HIGH elif combined >= 3: return Priority.MEDIUM return Priority.LOW def _generate_acceptance_criteria( self, category: str, target: str ) -> List[str]: """Generate acceptance criteria for requirement.""" criteria_templates = { "SPOOFING": [ f"Users must authenticate before accessing {target}", "Authentication failures are logged and monitored", "Multi-factor authentication is available for sensitive operations", ], "TAMPERING": [ f"All input to {target} is validated", "Data integrity is verified before processing", "Modification attempts trigger alerts", ], "REPUDIATION": [ f"All actions on {target} are logged with user identity", "Logs cannot be modified by regular users", "Log retention meets compliance requirements", ], "INFORMATION_DISCLOSURE": [ f"Sensitive data in {target} is encrypted", "Access to sensitive data is logged", "Error messages do not reveal sensitive information", ], "DENIAL_OF_SERVICE": [ f"Rate limiting is enforced on {target}", "System degrades gracefully under load", "Resource exhaustion triggers alerts", ], "ELEVATION_OF_PRIVILEGE": [ f"Authorization is checked for all {target} operations", "Users cannot access resources beyond their permissions", "Privilege changes are logged and monitored", ], } return criteria_templates.get(category, []) def _generate_test_cases( self, category: str, target: str ) -> List[str]: """Generate test cases for requirement.""" test_templates = { "SPOOFING": [ f"Test: Unauthenticated access to {target} is denied", "Test: Invalid credentials are rejected", "Test: Session tokens cannot be forged", ], "TAMPERING": [ f"Test: Invalid input to {target} is rejected", "Test: Tampered data is detected and rejected", "Test: SQL injection attempts are blocked", ], "REPUDIATION": [ "Test: Security events are logged", "Test: Logs include sufficient detail for forensics", "Test: Log integrity is protected", ], "INFORMATION_DISCLOSURE": [ f"Test: {target} data is encrypted in transit", f"Test: {target} data is encrypted at rest", "Test: Error messages are sanitized", ], "DENIAL_OF_SERVICE": [ f"Test: Rate limiting on {target} works correctly", "Test: System handles burst traffic gracefully", "Test: Resource limits are enforced", ], "ELEVATION_OF_PRIVILEGE": [ f"Test: Unauthorized access to {target} is denied", "Test: Privilege escalation attempts are blocked", "Test: IDOR vulnerabilities are not present", ], } return test_templates.get(category, []) ``` ### Template 3: Compliance Mapping ```python from typing import Dict, List, Set class ComplianceMapper: """Map security requirements to compliance frameworks.""" FRAMEWORK_CONTROLS = { ComplianceFramework.PCI_DSS: { SecurityDomain.AUTHENTICATION: ["8.1", "8.2", "8.3"], SecurityDomain.AUTHORIZATION: ["7.1", "7.2"], SecurityDomain.DATA_PROTECTION: ["3.4", "3.5", "4.1"], SecurityDomain.AUDIT_LOGGING: ["10.1", "10.2", "10.3"], SecurityDomain.NETWORK_SECURITY: ["1.1", "1.2", "1.3"], SecurityDomain.CRYPTOGRAPHY: ["3.5", "3.6", "4.1"], }, ComplianceFramework.HIPAA: { SecurityDomain.AUTHENTICATION: ["164.312(d)"], SecurityDomain.AUTHORIZATION: ["164.312(a)(1)"], SecurityDomain.DATA_PROTECTION: ["164.312(a)(2)(iv)", "164.312(e)(2)(ii)"], SecurityDomain.AUDIT_LOGGING: ["164.312(b)"], }, ComplianceFramework.GDPR: { SecurityDomain.DATA_PROTECTION: ["Art. 32", "Art. 25"], SecurityDomain.AUDIT_LOGGING: ["Art. 30"], SecurityDomain.AUTHORIZATION: ["Art. 25"], }, ComplianceFramework.OWASP: { SecurityDomain.AUTHENTICATION: ["V2.1", "V2.2", "V2.3"], SecurityDomain.SESSION_MANAGEMENT: ["V3.1", "V3.2", "V3.3"], SecurityDomain.INPUT_VALIDATION: ["V5.1", "V5.2", "V5.3"], SecurityDomain.CRYPTOGRAPHY: ["V6.1", "V6.2"], SecurityDomain.ERROR_HANDLING: ["V7.1", "V7.2"], SecurityDomain.DATA_PROTECTION: ["V8.1", "V8.2", "V8.3"], SecurityDomain.AUDIT_LOGGING: ["V7.1", "V7.2"], }, } def map_requirement_to_compliance( self, requirement: SecurityRequirement, frameworks: List[ComplianceFramework] ) -> Dict[str, List[str]]: """Map a requirement to compliance controls.""" mapping = {} for framework in frameworks: controls = self.FRAMEWORK_CONTROLS.get(framework, {}) domain_controls = controls.get(requirement.domain, []) if domain_controls: mapping[framework.value] = domain_controls return mapping def get_requirements_for_control( self, requirement_set: RequirementSet, framework: ComplianceFramework, control_id: str ) -> List[SecurityRequirement]: """Find requirements that satisfy a compliance control.""" matching = [] framework_controls = self.FRAMEWORK_CONTROLS.get(framework, {}) for domain, controls in framework_controls.items(): if control_id in controls: matching.extend(requirement_set.get_by_domain(domain)) return matching def generate_compliance_matrix( self, requirement_set: RequirementSet, frameworks: List[ComplianceFramework] ) -> Dict[str, Dict[str, List[str]]]: """Generate compliance traceability matrix.""" matrix = {} for framework in frameworks: matrix[framework.value] = {} framework_controls = self.FRAMEWORK_CONTROLS.get(framework, {}) for domain, controls in framework_controls.items(): for control in controls: reqs = self.get_requirements_for_control( requirement_set, framework, control ) if reqs: matrix[framework.value][control] = [r.id for r in reqs] return matrix def gap_analysis( self, requirement_set: RequirementSet, framework: ComplianceFramework ) -> Dict[str, List[str]]: """Identify compliance gaps.""" gaps = {"missing_controls": [], "weak_coverage": []} framework_controls = self.FRAMEWORK_CONTROLS.get(framework, {}) for domain, controls in framework_controls.items(): domain_reqs = requirement_set.get_by_domain(domain) for

# Threat Mitigation Mapping Connect threats to controls for effective security planning. ## When to Use This Skill - Prioritizing security investments - Creating remediation roadmaps - Validating control coverage - Designing defense-in-depth - Security architecture review - Risk treatment planning ## Core Concepts ### 1. Control Categories ``` Preventive ────► Stop attacks before they occur │ (Firewall, Input validation) │ Detective ─────► Identify attacks in progress │ (IDS, Log monitoring) │ Corrective ────► Respond and recover from attacks (Incident response, Backup restore) ``` ### 2. Control Layers | Layer | Examples | | --------------- | ------------------------------------ | | **Network** | Firewall, WAF, DDoS protection | | **Application** | Input validation, authentication | | **Data** | Encryption, access controls | | **Endpoint** | EDR, patch management | | **Process** | Security training, incident response | ### 3. Defense in Depth ``` ┌──────────────────────┐ │ Perimeter │ ← Firewall, WAF │ ┌──────────────┐ │ │ │ Network │ │ ← Segmentation, IDS │ │ ┌────────┐ │ │ │ │ │ Host │ │ │ ← EDR, Hardening │ │ │ ┌────┐ │ │ │ │ │ │ │App │ │ │ │ ← Auth, Validation │ │ │ │Data│ │ │ │ ← Encryption │ │ │ └────┘ │ │ │ │ │ └────────┘ │ │ │ └──────────────┘ │ └──────────────────────┘ ``` ## Templates ### Template 1: Mitigation Model ```python from dataclasses import dataclass, field from enum import Enum from typing import List, Dict, Optional, Set from datetime import datetime class ControlType(Enum): PREVENTIVE = "preventive" DETECTIVE = "detective" CORRECTIVE = "corrective" class ControlLayer(Enum): NETWORK = "network" APPLICATION = "application" DATA = "data" ENDPOINT = "endpoint" PROCESS = "process" PHYSICAL = "physical" class ImplementationStatus(Enum): NOT_IMPLEMENTED = "not_implemented" PARTIAL = "partial" IMPLEMENTED = "implemented" VERIFIED = "verified" class Effectiveness(Enum): NONE = 0 LOW = 1 MEDIUM = 2 HIGH = 3 VERY_HIGH = 4 @dataclass class SecurityControl: id: str name: str description: str control_type: ControlType layer: ControlLayer effectiveness: Effectiveness implementation_cost: str # Low, Medium, High maintenance_cost: str status: ImplementationStatus = ImplementationStatus.NOT_IMPLEMENTED mitigates_threats: List[str] = field(default_factory=list) dependencies: List[str] = field(default_factory=list) technologies: List[str] = field(default_factory=list) compliance_refs: List[str] = field(default_factory=list) def coverage_score(self) -> float: """Calculate coverage score based on status and effectiveness.""" status_multiplier = { ImplementationStatus.NOT_IMPLEMENTED: 0.0, ImplementationStatus.PARTIAL: 0.5, ImplementationStatus.IMPLEMENTED: 0.8, ImplementationStatus.VERIFIED: 1.0, } return self.effectiveness.value * status_multiplier[self.status] @dataclass class Threat: id: str name: str category: str # STRIDE category description: str impact: str # Critical, High, Medium, Low likelihood: str risk_score: float @dataclass class MitigationMapping: threat: Threat controls: List[SecurityControl] residual_risk: str = "Unknown" notes: str = "" def calculate_coverage(self) -> float: """Calculate how well controls cover the threat.""" if not self.controls: return 0.0 total_score = sum(c.coverage_score() for c in self.controls) max_possible = len(self.controls) * Effectiveness.VERY_HIGH.value return (total_score / max_possible) * 100 if max_possible > 0 else 0 def has_defense_in_depth(self) -> bool: """Check if multiple layers are covered.""" layers = set(c.layer for c in self.controls if c.status != ImplementationStatus.NOT_IMPLEMENTED) return len(layers) >= 2 def has_control_diversity(self) -> bool: """Check if multiple control types are present.""" types = set(c.control_type for c in self.controls if c.status != ImplementationStatus.NOT_IMPLEMENTED) return len(types) >= 2 @dataclass class MitigationPlan: name: str threats: List[Threat] = field(default_factory=list) controls: List[SecurityControl] = field(default_factory=list) mappings: List[MitigationMapping] = field(default_factory=list) def get_unmapped_threats(self) -> List[Threat]: """Find threats without mitigations.""" mapped_ids = {m.threat.id for m in self.mappings} return [t for t in self.threats if t.id not in mapped_ids] def get_control_coverage(self) -> Dict[str, float]: """Get coverage percentage for each threat.""" return { m.threat.id: m.calculate_coverage() for m in self.mappings } def get_gaps(self) -> List[Dict]: """Identify mitigation gaps.""" gaps = [] for mapping in self.mappings: coverage = mapping.calculate_coverage() if coverage < 50: gaps.append({ "threat": mapping.threat.id, "threat_name": mapping.threat.name, "coverage": coverage, "issue": "Insufficient control coverage", "recommendation": "Add more controls or improve existing ones" }) if not mapping.has_defense_in_depth(): gaps.append({ "threat": mapping.threat.id, "threat_name": mapping.threat.name, "coverage": coverage, "issue": "No defense in depth", "recommendation": "Add controls at different layers" }) if not mapping.has_control_diversity(): gaps.append({ "threat": mapping.threat.id, "threat_name": mapping.threat.name, "coverage": coverage, "issue": "No control diversity", "recommendation": "Add detective/corrective controls" }) return gaps ``` ### Template 2: Control Library ```python class ControlLibrary: """Library of standard security controls.""" STANDARD_CONTROLS = { # Authentication Controls "AUTH-001": SecurityControl( id="AUTH-001", name="Multi-Factor Authentication", description="Require MFA for all user authentication", control_type=ControlType.PREVENTIVE, layer=ControlLayer.APPLICATION, effectiveness=Effectiveness.HIGH, implementation_cost="Medium", maintenance_cost="Low", mitigates_threats=["SPOOFING"], technologies=["TOTP", "WebAuthn", "SMS OTP"], compliance_refs=["PCI-DSS 8.3", "NIST 800-63B"] ), "AUTH-002": SecurityControl( id="AUTH-002", name="Account Lockout Policy", description="Lock accounts after failed authentication attempts", control_type=ControlType.PREVENTIVE, layer=ControlLayer.APPLICATION, effectiveness=Effectiveness.MEDIUM, implementation_cost="Low", maintenance_cost="Low", mitigates_threats=["SPOOFING"], technologies=["Custom implementation"], compliance_refs=["PCI-DSS 8.1.6"] ), # Input Validation Controls "VAL-001": SecurityControl( id="VAL-001", name="Input Validation Framework", description="Validate and sanitize all user input", control_type=ControlType.PREVENTIVE, layer=ControlLayer.APPLICATION, effectiveness=Effectiveness.HIGH, implementation_cost="Medium", maintenance_cost="Medium", mitigates_threats=["TAMPERING", "INJECTION"], technologies=["Joi", "Yup", "Pydantic"], compliance_refs=["OWASP ASVS V5"] ), "VAL-002": SecurityControl( id="VAL-002", name="Web Application Firewall", description="Deploy WAF to filter malicious requests", control_type=ControlType.PREVENTIVE, layer=ControlLayer.NETWORK, effectiveness=Effectiveness.MEDIUM, implementation_cost="Medium", maintenance_cost="Medium", mitigates_threats=["TAMPERING", "INJECTION", "DOS"], technologies=["AWS WAF", "Cloudflare", "ModSecurity"], compliance_refs=["PCI-DSS 6.6"] ), # Encryption Controls "ENC-001": SecurityControl( id="ENC-001", name="Data Encryption at Rest", description="Encrypt sensitive data in storage", control_type=ControlType.PREVENTIVE, layer=ControlLayer.DATA, effectiveness=Effectiveness.HIGH, implementation_cost="Medium", maintenance_cost="Low", mitigates_threats=["INFORMATION_DISCLOSURE"], technologies=["AES-256", "KMS", "HSM"], compliance_refs=["PCI-DSS 3.4", "GDPR Art. 32"] ), "ENC-002": SecurityControl( id="ENC-002", name="TLS Encryption", description="Encrypt data in transit using TLS 1.3", control_type=ControlType.PREVENTIVE, layer=ControlLayer.NETWORK, effectiveness=Effectiveness.HIGH, implementation_cost="Low", maintenance_cost="Low", mitigates_threats=["INFORMATION_DISCLOSURE", "TAMPERING"], technologies=["TLS 1.3", "Certificate management"], compliance_refs=["PCI-DSS 4.1", "HIPAA"] ), # Logging Controls "LOG-001": SecurityControl( id="LOG-001", name="Security Event Logging", description="Log all security-relevant events", control_type=ControlType.DETECTIVE, layer=ControlLayer.APPLICATION, effectiveness=Effectiveness.MEDIUM, implementation_cost="Low", maintenance_cost="Medium", mitigates_threats=["REPUDIATION"], technologies=["ELK Stack", "Splunk", "CloudWatch"], compliance_refs=["PCI-DSS 10.2", "SOC2"] ), "LOG-002": SecurityControl( id="LOG-002", name="Log Integrity Protection", description="Protect logs from tampering", control_type=ControlType.PREVENTIVE, layer=ControlLayer.DATA, effectiveness=Effectiveness.MEDIUM, implementation_cost="Medium", maintenance_cost="Low", mitigates_threats=["REPUDIATION", "TAMPERING"], technologies=["Immutable storage", "Log signing"], compliance_refs=["PCI-DSS 10.5"] ), # Access Control "ACC-001": SecurityControl( id="ACC-001", name="Role-Based Access Control", description="Implement RBAC for authorization", control_type=ControlType.PREVENTIVE, layer=ControlLayer.APPLICATION, effectiveness=Effectiveness.HIGH, implementation_cost="Medium", maintenance_cost="Medium", mitigates_threats=["ELEVATION_OF_PRIVILEGE", "INFORMATION_DISCLOSURE"], technologies=["RBAC", "ABAC", "Policy engines"], compliance_refs=["PCI-DSS 7.1", "SOC2"] ), # Availability Controls "AVL-001": SecurityControl( id="AVL-001", name="Rate Limiting", description="Limit request rates to prevent abuse", control_type=ControlType.PREVENTIVE, layer=ControlLayer.APPLICATION, effectiveness=Effectiveness.MEDIUM, implementation_cost="Low", maintenance_cost="Low", mitigates_threats=["DENIAL_OF_SERVICE"], technologies=["API Gateway", "Redis", "Token bucket"], compliance_refs=["OWASP API Security"] ), "AVL-002": SecurityControl( id="AVL-002", name="DDoS Protection", description="Deploy DDoS mitigation services", control_type=ControlType.PREVENTIVE, layer=ControlLayer.NETWORK, effectiveness=Effectiveness.HIGH, implementation_cost="High", maintenance_cost="Medium", mitigates_threats=["DENIAL_OF_SERVICE"], technologies=["Cloudflare", "AWS Shield", "Akamai"], compliance_refs=["NIST CSF"] ), } def get_controls_for_threat(self, threat_category: str) -> List[SecurityControl]: """Get all controls that mitigate a threat category.""" return [ c for c in self.STANDARD_CONTROLS.values() if threat_category in c.mitigates_threats ] def get_controls_by_layer(self, layer: ControlLayer) -> List[SecurityControl]: """Get controls for a specific layer.""" return [c for c in self.STANDARD_CONTROLS.values() if c.layer == layer] def get_control(self, control_id: str) -> Optional[SecurityControl]: """Get a specific control by ID.""" return self.STANDARD_CONTROLS.get(control_id) def recommend_controls( self, threat: Threat, existing_controls: List[str] ) -> List[SecurityControl]: """Recommend additional controls for a threat.""" available = self.get_controls_for_threat(threat.category) return [c for c in available if c.id not in existing_controls] ``` ### Template 3: Mitigation Analysis ```python class MitigationAnalyzer: """Analyze and optimize mitigation strategies.""" def __init__(self, plan: MitigationPlan, library: ControlLibrary): self.plan = plan self.library = library def calculate_overall_risk_reduction(self) -> float: """Calculate overall risk reduction percentage.""" if not self.plan.mappings: return 0.0 weighted_coverage = 0 total_weight = 0 for mapping in self.plan.mappings: # Weight by threat risk score weight = mapping.threat.risk_score coverage = mapping.calculate_coverage() weighted_coverage += weight * coverage total_weight += weight return weighted_coverage / total_weight if total_weight > 0 else 0 def get_critical_gaps(self) -> List[Dict]: """Find critical gaps that need immediate attention.""" gaps = self.plan.get_gaps() critical_threats = {t.id for t in self.plan.threats if t.impact == "Critical"} return [g for g in gaps if g["threat"] in critical_threats] def optimize_budget( self, budget: float, cost_map: Dict[str, float] ) -> List[SecurityControl]: """Select controls that maximize risk reduction within budget.""" # Simple greedy approach - can be replaced with optimization algorithm recommended = [] remaining_budget = budget unmapped = self.plan.get_unmapped_threats() # Sort controls by effectiveness/cost ratio all_controls = list(self.library.STANDARD_CONTROLS.values()) controls_with_value = [] for control in all_controls: if control.status == ImplementationStatus.NOT_IMPLEMENTED: cost = cost_map.get(control.id, float('inf')) if cost <= remaining_budget: # Calculate value as threats covered * effectiveness / cost threats_covered = len([ t for t in unmapped if t.category in control.mitigates_threats ]) if threats_covered > 0: value = (threats_covered * control.effectiveness.value) / cost controls_with_value.append((control, value, cost)) # Sort by value (higher is better) controls_with_value.sort(key=lambda x: x[1], reverse=True) for control, value, cost in controls_with_value: if cost <= remaining_budget: recommended.append(control) remaining_budget -= cost return recommended def generate_roadmap(self) -> List[Dict]: """Generate implementation roadmap by priority.""" roadmap = [] gaps = self.plan.get_gaps() # Phase 1: Critical threats with low coverage phase1 = [] for gap in gaps: mapping = next( (m for m in self.plan.mappings if m.threat.id == gap["threat"]), None ) if mapping and mapping.threat.impact == "Critical": controls = self.library.get_controls_for_threat(mapping.threat.category) phase1.extend([ { "threat": gap["threat"], "control": c.id, "control_name": c.name, "phase": 1, "priority": "Critical" } for c in controls if c.status == ImplementationStatus.NOT_IMPLEMENTED ]) roadmap.extend(phase1[:5]) # Top 5 for phase 1 # Phase 2: High impact threats phase2 = [] for gap in gaps: mapping = next( (m for m in self.plan.mappings if m.threat.id == gap["threat"]), None ) if mapping and mapping.threat.impact == "High": controls = self.library.get_controls_for_threat(mapping.threat.category) phase2.extend([ { "threat": gap["threat"], "control": c.id, "control_name": c.name, "phase": 2, "priority": "High" } for c in controls if c.status == ImplementationStatus.NOT_IMPLEMENTED ]) roadmap.extend(phase2[:5]) # Top 5 for phase 2 return roadmap def defense_in_depth_analysis(self) -> Dict[str, List[str]]: """Analyze defense in depth coverage.""" layer_coverage = {layer.value: [] for layer in ControlLayer} for mapping in self.plan.mappings: for control in mapping.controls: if control.status in [ImplementationStatus.IMPLEMENTED, ImplementationStatus.VERIFIED]: layer_coverage[control.layer.value].append(control.id) return layer_coverage def generate_report(self) -> str: """Generate comprehensive m