Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

28 NODES
šŸ“text•3 hours ago

Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering

Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering User centric prompting for analyzing inspiration pages or instructing on internal analysis cycles

architecture
⭐1
# Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering You are an **Expert Enterprise Architect, Product Director, and Lead Engineer**. Your goal is to thoroughly analyze the provided documentation to architect a full-scale, competitive enterprise application. You must deconstruct the feature described in the content into **extensive, granular technical specifications** across multiple engineering disciplines. **CRITICAL INSTRUCTIONS**: - **DO NOT SUMMARIZE**. Be exhaustive. - For every category below, aim to list **10+ specific items** if possible. - Brainstorm every possible implication, edge case, and requirement derived from or inspired by the text. - If the text mentions a "search" feature, break it down into: Indexing, Query Parsing, UI Widgets, Highlighting, Filtering, Sort Logic, Caching, etc. Please output the analysis in the following Markdown format: # 1. Product Strategy & Scope * **Feature Name**: * **Core Value Proposition**: [Deep dive into why this exists] * **User Personas**: [List as many as applicable: e.g. Admin, Power User, Viewer, Auditor, API Consumer...] * **User Stories**: [Extensive list of 10+ granular user stories e.g. "As a User, I want to..."] * **Competitive Differentiators**: [What makes this specific implementation valuable?] # 2. Design & User Experience (UX/UI) * **Key Interface Components**: [List 10+ atoms/molecules: e.g. Data Grid, Filter Chips, Modals, Tooltips, Empty States, Toasts, Dropdowns...] * **Interaction Patterns**: [List 10+ patterns: e.g. Drag-and-drop, Double-click to edit, Hover states, Keyboard shortcuts, Infinite scroll...] * **Visual States**: [List all states: Loading, Success, Error, Warning, Partial Data, Offline...] * **Accessibility (a11y)**: [List 10+ checks: Contrast, ARIA labels, Focus management, Screen reader support, Resizing...] # 3. Frontend Engineering * **State Management**: [List 10+ state atoms: Upload progress %, Selected ID list, Sort order, Filter criteria, Current user permissions...] * **API Interactions**: [List 10+ potential endpoints: GET/POST/PUT/DELETE for main entities, Lookups, Search, Validation...] * **Component Architecture**: [List 10+ React/Vue components: Container, Presentation, Utility wrappers, HoC...] * **Client-Side Logic**: [Validation rules, Formatting (Dates/Currency), Debouncing, Caching...] # 4. Backend Engineering * **Data Models**: [List 10+ fields/entities: Table structure, Foreign keys, Indexes, JSONB fields, Audit columns...] * **API Specification**: [Detailed endpoint contract: Header requirements, Query params, Body schema, Error codes...] * **Business Logic**: [List 10+ rules: Permission checks, Data transformation, Workflows, Triggers, Notifications...] * **Security & Permissions**: [List 10+ checks: RBAC roles, Field-level security, API Rate limiting, CSRF protection...] # 5. Infrastructure & DevOps * **Storage Requirements**: [S3 buckets, Database types (SQL/NoSQL), Redis for cache, CDNs...] * **Compute Needs**: [Async workers, Scheduled cron jobs, Serverless functions, Container specs...] * **Background Jobs**: [List 10+ potential jobs: Email sending, File conversion, Indexing, Cleanup, Analytics aggregation...] * **Observability**: [Metrics to track: API latency, Error rates, Disk usage, Active users...] # 6. Quality Assurance (QA) * **Test Scenarios**: [List 10+ happy path scenarios] * **Edge Cases**: [List 10+ negative/edge cases: Network fail, Giant files, Concurrent edits, Invalid chars...] * **Performance Metrics**: [Specific SLAs: <200ms API, <1s Page load, 99.9% Uptime...] * **Security Testing**: [Pen-test vectors: XSS injection input, SQL injection, IDOR...] # 7. Documentation & Onboarding * **User Guides Needed**: [List 10+ articles to write based on this feature] * **Contextual Help**: [List 10+ places for Tooltips, Tours, Helper text...] * **API Documentation**: [Swagger/OpenAPI requirements] # 8. Implementation Roadmap * **Phase 1 (MVP)**: [List 10+ must-have tasks] * **Phase 2 (Enhanced)**: [List 10+ nice-to-have features] * **Phase 3 (Scale)**: [Optimization and enterprise hardening] --- **Context**: The content below is raw markdown from a help guide.
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ“text•3 hours ago

Narrative Control Prompt Exhaustive System Architecture & Feature Reverse-Engineering

Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering User concentric prompting for analyzing inspiration pages or instructing on internal analysis cycles

architecture
⭐1
# Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering You are an **Expert Enterprise Architect, Product Director, and Lead Engineer**. Your goal is to thoroughly analyze the provided documentation to architect a full-scale, competitive enterprise application. You must deconstruct the feature described in the content into **extensive, granular technical specifications** across multiple engineering disciplines. **CRITICAL INSTRUCTIONS**: - **DO NOT SUMMARIZE**. Be exhaustive. - For every category below, aim to list **10+ specific items** if possible. - Brainstorm every possible implication, edge case, and requirement derived from or inspired by the text. - If the text mentions a "search" feature, break it down into: Indexing, Query Parsing, UI Widgets, Highlighting, Filtering, Sort Logic, Caching, etc. Please output the analysis in the following Markdown format: # 1. Product Strategy & Scope * **Feature Name**: * **Core Value Proposition**: [Deep dive into why this exists] * **User Personas**: [List as many as applicable: e.g. Admin, Power User, Viewer, Auditor, API Consumer...] * **User Stories**: [Extensive list of 10+ granular user stories e.g. "As a User, I want to..."] * **Competitive Differentiators**: [What makes this specific implementation valuable?] # 2. Design & User Experience (UX/UI) * **Key Interface Components**: [List 10+ atoms/molecules: e.g. Data Grid, Filter Chips, Modals, Tooltips, Empty States, Toasts, Dropdowns...] * **Interaction Patterns**: [List 10+ patterns: e.g. Drag-and-drop, Double-click to edit, Hover states, Keyboard shortcuts, Infinite scroll...] * **Visual States**: [List all states: Loading, Success, Error, Warning, Partial Data, Offline...] * **Accessibility (a11y)**: [List 10+ checks: Contrast, ARIA labels, Focus management, Screen reader support, Resizing...] # 3. Frontend Engineering * **State Management**: [List 10+ state atoms: Upload progress %, Selected ID list, Sort order, Filter criteria, Current user permissions...] * **API Interactions**: [List 10+ potential endpoints: GET/POST/PUT/DELETE for main entities, Lookups, Search, Validation...] * **Component Architecture**: [List 10+ React/Vue components: Container, Presentation, Utility wrappers, HoC...] * **Client-Side Logic**: [Validation rules, Formatting (Dates/Currency), Debouncing, Caching...] # 4. Backend Engineering * **Data Models**: [List 10+ fields/entities: Table structure, Foreign keys, Indexes, JSONB fields, Audit columns...] * **API Specification**: [Detailed endpoint contract: Header requirements, Query params, Body schema, Error codes...] * **Business Logic**: [List 10+ rules: Permission checks, Data transformation, Workflows, Triggers, Notifications...] * **Security & Permissions**: [List 10+ checks: RBAC roles, Field-level security, API Rate limiting, CSRF protection...] # 5. Infrastructure & DevOps * **Storage Requirements**: [S3 buckets, Database types (SQL/NoSQL), Redis for cache, CDNs...] * **Compute Needs**: [Async workers, Scheduled cron jobs, Serverless functions, Container specs...] * **Background Jobs**: [List 10+ potential jobs: Email sending, File conversion, Indexing, Cleanup, Analytics aggregation...] * **Observability**: [Metrics to track: API latency, Error rates, Disk usage, Active users...] # 6. Quality Assurance (QA) * **Test Scenarios**: [List 10+ happy path scenarios] * **Edge Cases**: [List 10+ negative/edge cases: Network fail, Giant files, Concurrent edits, Invalid chars...] * **Performance Metrics**: [Specific SLAs: <200ms API, <1s Page load, 99.9% Uptime...] * **Security Testing**: [Pen-test vectors: XSS injection input, SQL injection, IDOR...] # 7. Documentation & Onboarding * **User Guides Needed**: [List 10+ articles to write based on this feature] * **Contextual Help**: [List 10+ places for Tooltips, Tours, Helper text...] * **API Documentation**: [Swagger/OpenAPI requirements] # 8. Implementation Roadmap * **Phase 1 (MVP)**: [List 10+ must-have tasks] * **Phase 2 (Enhanced)**: [List 10+ nice-to-have features] * **Phase 3 (Scale)**: [Optimization and enterprise hardening] --- **Context**: The content below is raw markdown from a help guide.
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ“text•6 months ago

ARCH-AEP Tiered Remediation Cycle

Run a scope-locked remediation cycle that normalizes findings, revalidates them in parallel, and clears tiers with verification evidence.

architecture
⭐1
# ARCH-AEP Tiered Remediation Cycle Imported from curated first-party documentation sources. ## What this covers Use this workflow when you need an orchestrated engineering review cycle with strict scope lock, tiered execution, and audit-grade evidence. ## Use this when - Large remediation efforts spanning multiple PRs - Parallel specialist review with shared backlog authority - Tracking verification evidence through closure ## Expected outcomes - Scope changes are controlled instead of creeping mid-cycle - Severity tiers drive remediation order predictably - Verification evidence is treated as part of done ## Source synthesis - CHELATEDAI/docs/ARCH AGENTIC ENGINEERING AND PLANNING/workflow.md (https://github.com/mattmre/CHELATEDAI/blob/main/docs/ARCH%20AGENTIC%20ENGINEERING%20AND%20PLANNING/workflow.md) - CHELATEDAI/docs/ARCH AGENTIC ENGINEERING AND PLANNING/orchestrator-briefing.md (https://github.com/mattmre/CHELATEDAI/blob/main/docs/ARCH%20AGENTIC%20ENGINEERING%20AND%20PLANNING/orchestrator-briefing.md) ## Dedupe notes Condenses the core ARCH-AEP workflow and briefing docs into a single remediation-cycle entry instead of importing both separately. ## Source excerpts ### CHELATEDAI/docs/ARCH AGENTIC ENGINEERING AND PLANNING/workflow.md ## Workflow (ARCH phase) 1. Scope lock - Define PR range and time window. - Freeze inputs (refinement report + framework + PR list). 2. Discovery + normalization - Orchestrator ingests `agentic-review-framework.md` and the latest refinement report. - Normalize all findings into a single backlog with unique IDs. - De-duplicate, merge overlaps, and assign provisional severity. - Record backlog in `docs/ARCH AGENTIC ENGINEERING AND PLANNING/backlog-YYYY-MM-DD.md`. 3. Parallel re-validation - Spawn specialist agents to re-validate findings against current main. - Each agent must attach exact file paths and line ranges. - Propose the smallest safe fix, acceptance criteria, and effort sizing (S/M/L). 4. Architecture and planning synthesis - Orchestrator merges validated findings into a master backlog. - Re-score and re-rank using the sorting model. - Identify dependency chains and blockers. ### CHELATEDAI/docs/ARCH AGENTIC ENGINEERING AND PLANNING/orchestrator-briefing.md Purpose: Quick reference for the ARCH-AEP documentation set, plus the narrative to start a new session. Full file index: `docs/INDEX.md` ## What Exists In This Folder - `README.md`: narrative overview and intent for ARCH-AEP. - `workflow.md`: end-to-end workflow specification with phases, guardrails, and enhancements. - `templates.md`: ID and branch conventions, tracker table format, and status log. - `backlog-template.md`: template for the master backlog file. - `backlog-index.md`: index of backlog files. - `next-session.md`: short handoff checklist for resuming work. - `phase-planning.md`: long-running planning record for the current cycle. - `schedule-and-tracking.md`: cadence, gates, and milestone tracking. - `agent-learning.md`: cross-session learnings and reusable patterns. - `change-log.md`: scope/defer decisions with audit context. - `risk-memo-template.md`: Critical/High risk memo template. - `risk-memos/README.md`: storage location for risk memos. - `phase-summary-template.md`: template for per-PR or per-phase summaries. - `phase-summaries/README.md`: storage location for phase summaries. - `verification-log.md`: test/build evidence mirror for PRs. - `tracker-pointer.md`: pointer to the active tracker file. - `tracker-index.md`: index of tracker files. - `scope-lock-t ...
šŸ‘0
šŸ‘ļø0
docs
šŸ“text•6 months ago

Improvement Cycle Review Wizard

Run a live improvement-cycle workflow that links review artifacts, approvals, tool requests, and operator checkpoints.

productivity
⭐1
# Improvement Cycle Review Wizard Imported from curated first-party documentation sources. ## What this covers Use this workflow when an operator needs a guided improvement cycle with live status, explicit approvals, and linked artifacts. ## Use this when - Operational review sessions with live progress - Approval-heavy improvement cycles - Coordinating review artifacts with execution steps ## Expected outcomes - Live workflow execution remains visible to operators - Artifacts, approvals, and tool requests stay connected - Improvement loops become easier to audit and rerun ## Source synthesis - AGENT33/docs/phase25-26-live-review-walkthrough.md (https://github.com/mattmre/AGENT33/blob/main/docs/phase25-26-live-review-walkthrough.md) - AGENT33/docs/operator-improvement-cycle-and-jupyter.md (https://github.com/mattmre/AGENT33/blob/main/docs/operator-improvement-cycle-and-jupyter.md) ## Dedupe notes Merges AGENT33 live review walkthrough details with the shorter operator-focused improvement-cycle guide. ## Source excerpts ### AGENT33/docs/phase25-26-live-review-walkthrough.md This guide documents the operator flow introduced by the Phase 25 live workflow transport and the Phase 26/27 improvement-cycle review wizard stack. ## Scope - Live workflow execution with run-scoped graph refresh - WebSocket-first status streaming with authenticated SSE fallback - Improvement-cycle preset creation - Explanation artifact generation for `plan_review` and `diff_review` - Linked review creation, risk assessment, L1/L2 signoff, and final approval - Pending tool-approval triage inside the same workflow domain > Note > This walkthrough reflects the review stack built on `codex/session58-phase26-wizard` and validated in `codex/session60-phase22-docs-validation`. If the related PRs are still open, `main` may not yet expose every surface described here. ## Operator Flow ```mermaid sequenceDiagram participant UI as "Frontend Control Plane" participant WF as "Workflow APIs" participant VIZ as "Visualization APIs" participant EXP as "Explanation APIs" participant REV as "Review APIs" participant HITL as "Tool Approval APIs" UI->>WF: "POST /v1/workflows/{name}/execute (single mode, caller run_id)" UI->>VIZ: "GET /v1/visualizations/workflows/{workflow_id}/graph?run_id=..." UI->>WF: "WS /v1/workflows/{run_id}/ws" alt "WebSocket unav ... ### AGENT33/docs/operator-improvement-cycle-and-jupyter.md This guide covers the merged operator surfaces for: - the Phase 26 improvement-cycle review wizard - the Phase 27 canonical workflow presets - the Phase 38 Docker-backed Jupyter kernel workflow Use it when you want the shortest current path from UI entry point to a real workflow run. ## 1. Improvement-Cycle Wizard The wizard is mounted inside the frontend control plane under the `Workflows` domain. ### Entry path 1. Open the frontend at `http://localhost:3000` 2. Authenticate with a bearer token or API key 3. Open `Advanced Settings` 4. Select the `Workflows` domain 5. Use the `Improvement Cycle Wizard` panel at the top of the page ### What the wizard does The wizard stitches together the backend surfaces that previously had to be called manually: - plan review / diff review generation - review creation and risk assessment - L1 and L2 review submission - tool approval request review and decision capture Reference implementation: - frontend: `frontend/src/features/improvement-cycle/ImprovementCycleWizard.tsx` - tests: `frontend/src/features/improvement-cycle/ImprovementCycleWizard.test.tsx` For the detailed review flow, keep using: - [`phase25-26-live-review-walkthrough.md`](phase25-26-live-review-walkthrough.md) ## 2. Canonical Workflow Presets The `Workflows` doma ...
šŸ‘0
šŸ‘ļø0
docs
šŸ“text•6 months ago

Guardrailed Code Review Pipeline

Run a staged code review workflow with explicit risk assessment, reviewer handoffs, and merge gates.

coding
⭐1
# Guardrailed Code Review Pipeline Imported from curated first-party documentation sources. ## What this covers Use this workflow when a change needs structured review evidence, clear approval stages, and a rollback-aware path to merge. ## Use this when - Multi-step L1 and L2 review signoff - High-risk changes that need audit trails - Standardizing reviewer decisions before merge ## Expected outcomes - Review states and approval checkpoints stay explicit - Risk assessment and evidence are captured before merge - The workflow remains small enough to reuse across repositories ## Source synthesis - AGENT33/docs/functionality-and-workflows.md (https://github.com/mattmre/AGENT33/blob/main/docs/functionality-and-workflows.md) - AGENT33/docs/walkthroughs.md (https://github.com/mattmre/AGENT33/blob/main/docs/walkthroughs.md) ## Dedupe notes Combines AGENT33 lifecycle mapping and operator walkthrough material into one review-oriented site entry. ## Source excerpts ### AGENT33/docs/functionality-and-workflows.md ### 4.1 Review Lifecycle States: - `draft -> ready -> l1-review -> l1-approved -> (optional l2-review -> l2-approved) -> approved -> merged` Main APIs: - `/v1/reviews/{id}/assess` - `/v1/reviews/{id}/assign-l1` - `/v1/reviews/{id}/l1` - `/v1/reviews/{id}/assign-l2` - `/v1/reviews/{id}/l2` - `/v1/reviews/{id}/approve` - `/v1/reviews/{id}/merge` ### AGENT33/docs/walkthroughs.md ## 4. Review Lifecycle (Two-Layer Signoff) Create review: ```bash curl -X POST http://localhost:8000/v1/reviews/ \ -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{"task_id":"TASK-101","branch":"feat/docs-refresh","pr_number":12}' ``` Assess risk: ```bash curl -X POST http://localhost:8000/v1/reviews/<review_id>/assess \ -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{"triggers":["api-public","security"]}' ``` Move to ready and assign L1: ```bash curl -X POST http://localhost:8000/v1/reviews/<review_id>/ready -H "Authorization: Bearer $TOKEN" curl -X POST http://localhost:8000/v1/reviews/<review_id>/assign-l1 -H "Authorization: Bearer $TOKEN" ``` Submit L1 decision: ```bash curl -X POST http://localhost:8000/v1/reviews/<review_id>/l1 \ -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{"decision":"approved","issues":[],"comments":"L1 pass"}' ``` If L2 required, continue: ```bash curl -X POST http://localhost:8000/v1/reviews/<review_id>/assign-l2 -H "Authorization: Bearer $TOKEN" curl -X POST http://localhost:8000/v1/reviews/<review_id>/l2 \ -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{"decision":"approved","issues":[],"commen ...
šŸ‘0
šŸ‘ļø0
docs
šŸ¤–system prompt•7 months ago

hybrid-cloud-networking

Configure secure, high-performance connectivity between on-premises

architecture
⭐1
# Hybrid Cloud Networking Configure secure, high-performance connectivity between on-premises and cloud environments using VPN, Direct Connect, and ExpressRoute. ## Purpose Establish secure, reliable network connectivity between on-premises data centers and cloud providers (AWS, Azure, GCP). ## When to Use - Connect on-premises to cloud - Extend datacenter to cloud - Implement hybrid active-active setups - Meet compliance requirements - Migrate to cloud gradually ## Connection Options ### AWS Connectivity #### 1. Site-to-Site VPN - IPSec VPN over internet - Up to 1.25 Gbps per tunnel - Cost-effective for moderate bandwidth - Higher latency, internet-dependent ```hcl resource "aws_vpn_gateway" "main" { vpc_id = aws_vpc.main.id tags = { Name = "main-vpn-gateway" } } resource "aws_customer_gateway" "main" { bgp_asn = 65000 ip_address = "203.0.113.1" type = "ipsec.1" } resource "aws_vpn_connection" "main" { vpn_gateway_id = aws_vpn_gateway.main.id customer_gateway_id = aws_customer_gateway.main.id type = "ipsec.1" static_routes_only = false } ``` #### 2. AWS Direct Connect - Dedicated network connection - 1 Gbps to 100 Gbps - Lower latency, consistent bandwidth - More expensive, setup time required **Reference:** See `references/direct-connect.md` ### Azure Connectivity #### 1. Site-to-Site VPN ```hcl resource "azurerm_virtual_network_gateway" "vpn" { name = "vpn-gateway" location = azurerm_resource_group.main.location resource_group_name = azurerm_resource_group.main.name type = "Vpn" vpn_type = "RouteBased" sku = "VpnGw1" ip_configuration { name = "vnetGatewayConfig" public_ip_address_id = azurerm_public_ip.vpn.id private_ip_address_allocation = "Dynamic" subnet_id = azurerm_subnet.gateway.id } } ``` #### 2. Azure ExpressRoute - Private connection via connectivity provider - Up to 100 Gbps - Low latency, high reliability - Premium for global connectivity ### GCP Connectivity #### 1. Cloud VPN - IPSec VPN (Classic or HA VPN) - HA VPN: 99.99% SLA - Up to 3 Gbps per tunnel #### 2. Cloud Interconnect - Dedicated (10 Gbps, 100 Gbps) - Partner (50 Mbps to 50 Gbps) - Lower latency than VPN ## Hybrid Network Patterns ### Pattern 1: Hub-and-Spoke ``` On-Premises Datacenter ↓ VPN/Direct Connect ↓ Transit Gateway (AWS) / vWAN (Azure) ↓ ā”œā”€ Production VPC/VNet ā”œā”€ Staging VPC/VNet └─ Development VPC/VNet ``` ### Pattern 2: Multi-Region Hybrid ``` On-Premises ā”œā”€ Direct Connect → us-east-1 └─ Direct Connect → us-west-2 ↓ Cross-Region Peering ``` ### Pattern 3: Multi-Cloud Hybrid ``` On-Premises Datacenter ā”œā”€ Direct Connect → AWS ā”œā”€ ExpressRoute → Azure └─ Interconnect → GCP ``` ## Routing Configuration ### BGP Configuration ``` On-Premises Router: - AS Number: 65000 - Advertise: 10.0.0.0/8 Cloud Router: - AS Number: 64512 (AWS), 65515 (Azure) - Advertise: Cloud VPC/VNet CIDRs ``` ### Route Propagation - Enable route propagation on route tables - Use BGP for dynamic routing - Implement route filtering - Monitor route advertisements ## Security Best Practices 1. **Use private connectivity** (Direct Connect/ExpressRoute) 2. **Implement encryption** for VPN tunnels 3. **Use VPC endpoints** to avoid internet routing 4. **Configure network ACLs** and security groups 5. **Enable VPC Flow Logs** for monitoring 6. **Implement DDoS protection** 7. **Use PrivateLink/Private Endpoints** 8. **Monitor connections** with CloudWatch/Monitor 9. **Implement redundancy** (dual tunnels) 10. **Regular security audits** ## High Availability ### Dual VPN Tunnels ```hcl resource "aws_vpn_connection" "primary" { vpn_gateway_id = aws_vpn_gateway.main.id customer_gateway_id = aws_customer_gateway.primary.id type = "ipsec.1" } resource "aws_vpn_connection" "secondary" { vpn_gateway_id = aws_vpn_gateway.main.id customer_gateway_id = aws_customer_gateway.secondary.id type = "ipsec.1" } ``` ### Active-Active Configuration - Multiple connections from different locations - BGP for automatic failover - Equal-cost multi-path (ECMP) routing - Monitor health of all connections ## Monitoring and Troubleshooting ### Key Metrics - Tunnel status (up/down) - Bytes in/out - Packet loss - Latency - BGP session status ### Troubleshooting ```bash # AWS VPN aws ec2 describe-vpn-connections aws ec2 get-vpn-connection-telemetry # Azure VPN az network vpn-connection show az network vpn-connection show-device-config-script ``` ## Cost Optimization 1. **Right-size connections** based on traffic 2. **Use VPN for low-bandwidth** workloads 3. **Consolidate traffic** through fewer connections 4. **Minimize data transfer** costs 5. **Use Direct Connect** for high bandwidth 6. **Implement caching** to reduce traffic ## Reference Files - `references/vpn-setup.md` - VPN configuration guide - `references/direct-connect.md` - Direct Connect setup ## Related Skills - `multi-cloud-architecture` - For architecture decisions - `terraform-module-library` - For IaC implementation
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

mtls-configuration

Configure mutual TLS (mTLS) for zero-trust service-to-service

architecture
⭐1
# 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)
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

service-mesh-observability

Implement comprehensive observability for service meshes including

architecture
⭐1
# Service Mesh Observability Complete guide to observability patterns for Istio, Linkerd, and service mesh deployments. ## When to Use This Skill - Setting up distributed tracing across services - Implementing service mesh metrics and dashboards - Debugging latency and error issues - Defining SLOs for service communication - Visualizing service dependencies - Troubleshooting mesh connectivity ## Core Concepts ### 1. Three Pillars of Observability ``` ā”Œā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā” │ Observability │ ā”œā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”¬ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”¬ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”¤ │ Metrics │ Traces │ Logs │ │ │ │ │ │ • Request rate │ • Span context │ • Access logs │ │ • Error rate │ • Latency │ • Error details │ │ • Latency P50 │ • Dependencies │ • Debug info │ │ • Saturation │ • Bottlenecks │ • Audit trail │ ā””ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”“ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”“ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”˜ ``` ### 2. Golden Signals for Mesh | Signal | Description | Alert Threshold | | -------------- | ------------------------- | ----------------- | | **Latency** | Request duration P50, P99 | P99 > 500ms | | **Traffic** | Requests per second | Anomaly detection | | **Errors** | 5xx error rate | > 1% | | **Saturation** | Resource utilization | > 80% | ## Templates ### Template 1: Istio with Prometheus & Grafana ```yaml # Install Prometheus apiVersion: v1 kind: ConfigMap metadata: name: prometheus namespace: istio-system data: prometheus.yml: | global: scrape_interval: 15s scrape_configs: - job_name: 'istio-mesh' kubernetes_sd_configs: - role: endpoints namespaces: names: - istio-system relabel_configs: - source_labels: [__meta_kubernetes_service_name] action: keep regex: istio-telemetry --- # ServiceMonitor for Prometheus Operator apiVersion: monitoring.coreos.com/v1 kind: ServiceMonitor metadata: name: istio-mesh namespace: istio-system spec: selector: matchLabels: app: istiod endpoints: - port: http-monitoring interval: 15s ``` ### Template 2: Key Istio Metrics Queries ```promql # Request rate by service sum(rate(istio_requests_total{reporter="destination"}[5m])) by (destination_service_name) # Error rate (5xx) sum(rate(istio_requests_total{reporter="destination", response_code=~"5.."}[5m])) / sum(rate(istio_requests_total{reporter="destination"}[5m])) * 100 # P99 latency histogram_quantile(0.99, sum(rate(istio_request_duration_milliseconds_bucket{reporter="destination"}[5m])) by (le, destination_service_name)) # TCP connections sum(istio_tcp_connections_opened_total{reporter="destination"}) by (destination_service_name) # Request size histogram_quantile(0.99, sum(rate(istio_request_bytes_bucket{reporter="destination"}[5m])) by (le, destination_service_name)) ``` ### Template 3: Jaeger Distributed Tracing ```yaml # Jaeger installation for Istio apiVersion: install.istio.io/v1alpha1 kind: IstioOperator spec: meshConfig: enableTracing: true defaultConfig: tracing: sampling: 100.0 # 100% in dev, lower in prod zipkin: address: jaeger-collector.istio-system:9411 --- # Jaeger deployment apiVersion: apps/v1 kind: Deployment metadata: name: jaeger namespace: istio-system spec: selector: matchLabels: app: jaeger template: metadata: labels: app: jaeger spec: containers: - name: jaeger image: jaegertracing/all-in-one:1.50 ports: - containerPort: 5775 # UDP - containerPort: 6831 # Thrift - containerPort: 6832 # Thrift - containerPort: 5778 # Config - containerPort: 16686 # UI - containerPort: 14268 # HTTP - containerPort: 14250 # gRPC - containerPort: 9411 # Zipkin env: - name: COLLECTOR_ZIPKIN_HOST_PORT value: ":9411" ``` ### Template 4: Linkerd Viz Dashboard ```bash # Install Linkerd viz extension linkerd viz install | kubectl apply -f - # Access dashboard linkerd viz dashboard # CLI commands for observability # Top requests linkerd viz top deploy/my-app # Per-route metrics linkerd viz routes deploy/my-app --to deploy/backend # Live traffic inspection linkerd viz tap deploy/my-app --to deploy/backend # Service edges (dependencies) linkerd viz edges deployment -n my-namespace ``` ### Template 5: Grafana Dashboard JSON ```json { "dashboard": { "title": "Service Mesh Overview", "panels": [ { "title": "Request Rate", "type": "graph", "targets": [ { "expr": "sum(rate(istio_requests_total{reporter=\"destination\"}[5m])) by (destination_service_name)", "legendFormat": "{{destination_service_name}}" } ] }, { "title": "Error Rate", "type": "gauge", "targets": [ { "expr": "sum(rate(istio_requests_total{response_code=~\"5..\"}[5m])) / sum(rate(istio_requests_total[5m])) * 100" } ], "fieldConfig": { "defaults": { "thresholds": { "steps": [ { "value": 0, "color": "green" }, { "value": 1, "color": "yellow" }, { "value": 5, "color": "red" } ] } } } }, { "title": "P99 Latency", "type": "graph", "targets": [ { "expr": "histogram_quantile(0.99, sum(rate(istio_request_duration_milliseconds_bucket{reporter=\"destination\"}[5m])) by (le, destination_service_name))", "legendFormat": "{{destination_service_name}}" } ] }, { "title": "Service Topology", "type": "nodeGraph", "targets": [ { "expr": "sum(rate(istio_requests_total{reporter=\"destination\"}[5m])) by (source_workload, destination_service_name)" } ] } ] } } ``` ### Template 6: Kiali Service Mesh Visualization ```yaml # Kiali installation apiVersion: kiali.io/v1alpha1 kind: Kiali metadata: name: kiali namespace: istio-system spec: auth: strategy: anonymous # or openid, token deployment: accessible_namespaces: - "**" external_services: prometheus: url: http://prometheus.istio-system:9090 tracing: url: http://jaeger-query.istio-system:16686 grafana: url: http://grafana.istio-system:3000 ``` ### Template 7: OpenTelemetry Integration ```yaml # OpenTelemetry Collector for mesh apiVersion: v1 kind: ConfigMap metadata: name: otel-collector-config data: config.yaml: | receivers: otlp: protocols: grpc: endpoint: 0.0.0.0:4317 http: endpoint: 0.0.0.0:4318 zipkin: endpoint: 0.0.0.0:9411 processors: batch: timeout: 10s exporters: jaeger: endpoint: jaeger-collector:14250 tls: insecure: true prometheus: endpoint: 0.0.0.0:8889 service: pipelines: traces: receivers: [otlp, zipkin] processors: [batch] exporters: [jaeger] metrics: receivers: [otlp] processors: [batch] exporters: [prometheus] --- # Istio Telemetry v2 with OTel apiVersion: telemetry.istio.io/v1alpha1 kind: Telemetry metadata: name: mesh-default namespace: istio-system spec: tracing: - providers: - name: otel randomSamplingPercentage: 10 ``` ## Alerting Rules ```yaml apiVersion: monitoring.coreos.com/v1 kind: PrometheusRule metadata: name: mesh-alerts namespace: istio-system spec: groups: - name: mesh.rules rules: - alert: HighErrorRate expr: | sum(rate(istio_requests_total{response_code=~"5.."}[5m])) by (destination_service_name) / sum(rate(istio_requests_total[5m])) by (destination_service_name) > 0.05 for: 5m labels: severity: critical annotations: summary: "High error rate for {{ $labels.destination_service_name }}" - alert: HighLatency expr: | histogram_quantile(0.99, sum(rate(istio_request_duration_milliseconds_bucket[5m])) by (le, destination_service_name)) > 1000 for: 5m labels: severity: warning annotations: summary: "High P99 latency for {{ $labels.destination_service_name }}" - alert: MeshCertExpiring expr: | (certmanager_certificate_expiration_timestamp_seconds - time()) / 86400 < 7 labels: severity: warning annotations: summary: "Mesh certificate expiring in less than 7 days" ``` ## Best Practices ### Do's - **Sample appropriately** - 100% in dev, 1-10% in prod - **Use trace context** - Propagate headers consistently - **Set up alerts** - For golden signals - **Correlate metrics/traces** - Use exemplars - **Retain strategically** - Hot/cold storage tiers ### Don'ts - **Don't over-sample** - Storage costs add up - **Don't ignore cardinality** - Limit label values - **Don't skip dashboards** - Visualize dependencies - **Don't forget costs** - Monitor observability costs ## Resources - [Istio Observability](https://istio.io/latest/docs/tasks/observability/) - [Linkerd Observability](https://linkerd.io/2.14/features/dashboard/) - [OpenTelemetry](https://opentelemetry.io/) - [Kiali](https://kiali.io/)
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

workflow-patterns

Use this skill when implementing tasks according to Conductor's TDD

coding
⭐1
# Workflow Patterns Guide for implementing tasks using Conductor's TDD workflow, managing phase checkpoints, handling git commits, and executing the verification protocol that ensures quality throughout implementation. ## When to Use This Skill - Implementing tasks from a track's plan.md - Following TDD red-green-refactor cycle - Completing phase checkpoints - Managing git commits and notes - Understanding quality assurance gates - Handling verification protocols - Recording progress in plan files ## TDD Task Lifecycle Follow these 11 steps for each task: ### Step 1: Select Next Task Read plan.md and identify the next pending `[ ]` task. Select tasks in order within the current phase. Do not skip ahead to later phases. ### Step 2: Mark as In Progress Update plan.md to mark the task as `[~]`: ```markdown - [~] **Task 2.1**: Implement user validation ``` Commit this status change separately from implementation. ### Step 3: RED - Write Failing Tests Write tests that define the expected behavior before writing implementation: - Create test file if needed - Write test cases covering happy path - Write test cases covering edge cases - Write test cases covering error conditions - Run tests - they should FAIL Example: ```python def test_validate_user_email_valid(): user = User(email="test@example.com") assert user.validate_email() is True def test_validate_user_email_invalid(): user = User(email="invalid") assert user.validate_email() is False ``` ### Step 4: GREEN - Implement Minimum Code Write the minimum code necessary to make tests pass: - Focus on making tests green, not perfection - Avoid premature optimization - Keep implementation simple - Run tests - they should PASS ### Step 5: REFACTOR - Improve Clarity With green tests, improve the code: - Extract common patterns - Improve naming - Remove duplication - Simplify logic - Run tests after each change - they should remain GREEN ### Step 6: Verify Coverage Check test coverage meets the 80% target: ```bash pytest --cov=module --cov-report=term-missing ``` If coverage is below 80%: - Identify uncovered lines - Add tests for missing paths - Re-run coverage check ### Step 7: Document Deviations If implementation deviated from plan or introduced new dependencies: - Update tech-stack.md with new dependencies - Note deviations in plan.md task comments - Update spec.md if requirements changed ### Step 8: Commit Implementation Create a focused commit for the task: ```bash git add -A git commit -m "feat(user): implement email validation - Add validate_email method to User class - Handle empty and malformed emails - Add comprehensive test coverage Task: 2.1 Track: user-auth_20250115" ``` Commit message format: - Type: feat, fix, refactor, test, docs, chore - Scope: affected module or component - Summary: imperative, present tense - Body: bullet points of changes - Footer: task and track references ### Step 9: Attach Git Notes Add rich task summary as git note: ```bash git notes add -m "Task 2.1: Implement user validation Summary: - Added email validation using regex pattern - Handles edge cases: empty, no @, no domain - Coverage: 94% on validation module Files changed: - src/models/user.py (modified) - tests/test_user.py (modified) Decisions: - Used simple regex over email-validator library - Reason: No external dependency for basic validation" ``` ### Step 10: Update Plan with SHA Update plan.md to mark task complete with commit SHA: ```markdown - [x] **Task 2.1**: Implement user validation `abc1234` ``` ### Step 11: Commit Plan Update Commit the plan status update: ```bash git add conductor/tracks/*/plan.md git commit -m "docs: update plan - task 2.1 complete Track: user-auth_20250115" ``` ## Phase Completion Protocol When all tasks in a phase are complete, execute the verification protocol: ### Identify Changed Files List all files modified since the last checkpoint: ```bash git diff --name-only <last-checkpoint-sha>..HEAD ``` ### Ensure Test Coverage For each modified file: 1. Identify corresponding test file 2. Verify tests exist for new/changed code 3. Run coverage for modified modules 4. Add tests if coverage < 80% ### Run Full Test Suite Execute complete test suite: ```bash pytest -v --tb=short ``` All tests must pass before proceeding. ### Generate Manual Verification Steps Create checklist of manual verifications: ```markdown ## Phase 1 Verification Checklist - [ ] User can register with valid email - [ ] Invalid email shows appropriate error - [ ] Database stores user correctly - [ ] API returns expected response codes ``` ### WAIT for User Approval Present verification checklist to user: ``` Phase 1 complete. Please verify: 1. [ ] Test suite passes (automated) 2. [ ] Coverage meets target (automated) 3. [ ] Manual verification items (requires human) Respond with 'approved' to continue, or note issues. ``` Do NOT proceed without explicit approval. ### Create Checkpoint Commit After approval, create checkpoint commit: ```bash git add -A git commit -m "checkpoint: phase 1 complete - user-auth_20250115 Verified: - All tests passing - Coverage: 87% - Manual verification approved Phase 1 tasks: - [x] Task 1.1: Setup database schema - [x] Task 1.2: Implement user model - [x] Task 1.3: Add validation logic" ``` ### Record Checkpoint SHA Update plan.md checkpoints table: ```markdown ## Checkpoints | Phase | Checkpoint SHA | Date | Status | | ------- | -------------- | ---------- | -------- | | Phase 1 | def5678 | 2025-01-15 | verified | | Phase 2 | | | pending | ``` ## Quality Assurance Gates Before marking any task complete, verify these gates: ### Passing Tests - All existing tests pass - New tests pass - No test regressions ### Coverage >= 80% - New code has 80%+ coverage - Overall project coverage maintained - Critical paths fully covered ### Style Compliance - Code follows style guides - Linting passes - Formatting correct ### Documentation - Public APIs documented - Complex logic explained - README updated if needed ### Type Safety - Type hints present (if applicable) - Type checker passes - No type: ignore without reason ### No Linting Errors - Zero linter errors - Warnings addressed or justified - Static analysis clean ### Mobile Compatibility If applicable: - Responsive design verified - Touch interactions work - Performance acceptable ### Security Audit - No secrets in code - Input validation present - Authentication/authorization correct - Dependencies vulnerability-free ## Git Integration ### Commit Message Format ``` <type>(<scope>): <subject> <body> <footer> ``` Types: - `feat`: New feature - `fix`: Bug fix - `refactor`: Code change without feature/fix - `test`: Adding tests - `docs`: Documentation - `chore`: Maintenance ### Git Notes for Rich Summaries Attach detailed notes to commits: ```bash git notes add -m "<detailed summary>" ``` View notes: ```bash git log --show-notes ``` Benefits: - Preserves context without cluttering commit message - Enables semantic queries across commits - Supports track-based operations ### SHA Recording in plan.md Always record the commit SHA when completing tasks: ```markdown - [x] **Task 1.1**: Setup schema `abc1234` - [x] **Task 1.2**: Add model `def5678` ``` This enables: - Traceability from plan to code - Semantic revert operations - Progress auditing ## Verification Checkpoints ### Why Checkpoints Matter Checkpoints create restore points for semantic reversion: - Revert to end of any phase - Maintain logical code state - Enable safe experimentation ### When to Create Checkpoints Create checkpoint after: - All phase tasks complete - All phase verifications pass - User approval received ### Checkpoint Commit Content Include in checkpoint commit: - All uncommitted changes - Updated plan.md - Updated metadata.json - Any documentation updates ### How to Use Checkpoints For reverting: ```bash # Revert to end of Phase 1 git revert --no-commit <phase-2-commits>... git commit -m "revert: rollback to phase 1 checkpoint" ``` For review: ```bash # See what changed in Phase 2 git diff <phase-1-sha>..<phase-2-sha> ``` ## Handling Deviations During implementation, deviations from the plan may occur. Handle them systematically: ### Types of Deviations **Scope Addition** Discovered requirement not in original spec. - Document in spec.md as new requirement - Add tasks to plan.md - Note addition in task comments **Scope Reduction** Feature deemed unnecessary during implementation. - Mark tasks as `[-]` (skipped) with reason - Update spec.md scope section - Document decision rationale **Technical Deviation** Different implementation approach than planned. - Note deviation in task completion comment - Update tech-stack.md if dependencies changed - Document why original approach was unsuitable **Requirement Change** Understanding of requirement changes during work. - Update spec.md with corrected requirement - Adjust plan.md tasks if needed - Re-verify acceptance criteria ### Deviation Documentation Format When completing a task with deviation: ```markdown - [x] **Task 2.1**: Implement validation `abc1234` - DEVIATION: Used library instead of custom code - Reason: Better edge case handling - Impact: Added email-validator to dependencies ``` ## Error Recovery ### Failed Tests After GREEN If tests fail after reaching GREEN: 1. Do NOT proceed to REFACTOR 2. Identify which test started failing 3. Check if refactoring broke something 4. Revert to last known GREEN state 5. Re-approach the implementation ### Checkpoint Rejection If user rejects a checkpoint: 1. Note rejection reason in plan.md 2. Create tasks to address issues 3. Complete remediation tasks 4. Request checkpoint approval again ### Blocked by Dependency If task cannot proceed: 1. Mark task as `[!]` with blocker description 2. Check if other tasks can proceed 3. Document expected resolution timeline 4. Consider creating dependency resolution track ## TDD Variations by Task Type ### Data Model Tasks ``` RED: Write test for model creation and validation GREEN: Implement model class with fields REFACTOR: Add computed properties, improve types ``` ### API Endpoint Tasks ``` RED: Write test for request/response contract GREEN: Implement endpoint handler REFACTOR: Extract validation, improve error handling ``` ### Integration Tasks ``` RED: Write test for component interaction GREEN: Wire components together REFACTOR: Improve error propagation, add logging ``` ### Refactoring Tasks ``` RED: Add characterization tests for current behavior GREEN: Apply refactoring (tests should stay green) REFACTOR: Clean up any introduced complexity ``` ## Working with Existing Tests When modifying code with existing tests: ### Extend, Don't Replace - Keep existing tests passing - Add new tests for new behavior - Update tests only when requirements change ### Test Migration When refactoring changes test structure: 1. Run existing tests (should pass) 2. Add new tests for refactored code 3. Migrate test cases to new structure 4. Remove old tests only after new tests pass ### Regression Prevention After any change: 1. Run full test suite 2. Check for unexpected failures 3. Investigate any new failures 4. Fix regressions before proceeding ## Checkpoint Verification Details ### Automated Verification Run before requesting approval: ```bash # Test suite pytest -v --tb=short # Coverage pytest --cov=src --cov-report=term-missing # Linting ruff check src/ tests/ # Type checking (if applicable) mypy src/ ``` ### Manual Verification Guidance For manual items, provide specific instructions: ```markdown ## Manual Verification Steps ### User Registration 1. Navigate to /register 2. Enter valid email: test@example.com 3. Enter password meeting requirements 4. Click Submit 5. Verify success message appears 6. Verify user appears in database ### Error Handling 1. Enter invalid email: "notanemail" 2. Verify error message shows 3. Verify form retains other entered data ``` ## Performance Considerations ### Test Suite Performance Keep test suite fast: - Use fixtures to avoid redundant setup - Mock slow external calls - Run subset during development, full suite at checkpoints ### Commit Performance Keep commits atomic: - One logical change per commit - Complete thought, not work-in-progress - Tests should pass after every commit ## Best Practices 1. **Never skip RED**: Always write failing tests first 2. **Small commits**: One logical change per commit 3. **Immediate updates**: Update plan.md right after task completion 4. **Wait for approval**: Never skip checkpoint verification 5. **Rich git notes**: Include context that helps future understanding 6. **Coverage discipline**: Don't accept coverage below target 7. **Quality gates**: Check all gates before marking complete 8. **Sequential phases**: Complete phases in order 9. **Document deviations**: Note any changes from original plan 10. **Clean state**: Each commit should leave code in working state 11. **Fast feedback**: Run relevant tests frequently during development 12. **Clear blockers**: Address blockers promptly, don't work around them
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

postgresql-table-design

Design a PostgreSQL-specific schema. Covers best-practices, data

data
⭐1
# PostgreSQL Table Design ## Core Rules - Define a **PRIMARY KEY** for reference tables (users, orders, etc.). Not always needed for time-series/event/log data. When used, prefer `BIGINT GENERATED ALWAYS AS IDENTITY`; use `UUID` only when global uniqueness/opacity is needed. - **Normalize first (to 3NF)** to eliminate data redundancy and update anomalies; denormalize **only** for measured, high-ROI reads where join performance is proven problematic. Premature denormalization creates maintenance burden. - Add **NOT NULL** everywhere it’s semantically required; use **DEFAULT**s for common values. - Create **indexes for access paths you actually query**: PK/unique (auto), **FK columns (manual!)**, frequent filters/sorts, and join keys. - Prefer **TIMESTAMPTZ** for event time; **NUMERIC** for money; **TEXT** for strings; **BIGINT** for integer values, **DOUBLE PRECISION** for floats (or `NUMERIC` for exact decimal arithmetic). ## PostgreSQL ā€œGotchasā€ - **Identifiers**: unquoted → lowercased. Avoid quoted/mixed-case names. Convention: use `snake_case` for table/column names. - **Unique + NULLs**: UNIQUE allows multiple NULLs. Use `UNIQUE (...) NULLS NOT DISTINCT` (PG15+) to restrict to one NULL. - **FK indexes**: PostgreSQL **does not** auto-index FK columns. Add them. - **No silent coercions**: length/precision overflows error out (no truncation). Example: inserting 999 into `NUMERIC(2,0)` fails with error, unlike some databases that silently truncate or round. - **Sequences/identity have gaps** (normal; don't "fix"). Rollbacks, crashes, and concurrent transactions create gaps in ID sequences (1, 2, 5, 6...). This is expected behavior—don't try to make IDs consecutive. - **Heap storage**: no clustered PK by default (unlike SQL Server/MySQL InnoDB); `CLUSTER` is one-off reorganization, not maintained on subsequent inserts. Row order on disk is insertion order unless explicitly clustered. - **MVCC**: updates/deletes leave dead tuples; vacuum handles them—design to avoid hot wide-row churn. ## Data Types - **IDs**: `BIGINT GENERATED ALWAYS AS IDENTITY` preferred (`GENERATED BY DEFAULT` also fine); `UUID` when merging/federating/used in a distributed system or for opaque IDs. Generate with `uuidv7()` (preferred if using PG18+) or `gen_random_uuid()` (if using an older PG version). - **Integers**: prefer `BIGINT` unless storage space is critical; `INTEGER` for smaller ranges; avoid `SMALLINT` unless constrained. - **Floats**: prefer `DOUBLE PRECISION` over `REAL` unless storage space is critical. Use `NUMERIC` for exact decimal arithmetic. - **Strings**: prefer `TEXT`; if length limits needed, use `CHECK (LENGTH(col) <= n)` instead of `VARCHAR(n)`; avoid `CHAR(n)`. Use `BYTEA` for binary data. Large strings/binary (>2KB default threshold) automatically stored in TOAST with compression. TOAST storage: `PLAIN` (no TOAST), `EXTENDED` (compress + out-of-line), `EXTERNAL` (out-of-line, no compress), `MAIN` (compress, keep in-line if possible). Default `EXTENDED` usually optimal. Control with `ALTER TABLE tbl ALTER COLUMN col SET STORAGE strategy` and `ALTER TABLE tbl SET (toast_tuple_target = 4096)` for threshold. Case-insensitive: for locale/accent handling use non-deterministic collations; for plain ASCII use expression indexes on `LOWER(col)` (preferred unless column needs case-insensitive PK/FK/UNIQUE) or `CITEXT`. - **Money**: `NUMERIC(p,s)` (never float). - **Time**: `TIMESTAMPTZ` for timestamps; `DATE` for date-only; `INTERVAL` for durations. Avoid `TIMESTAMP` (without timezone). Use `now()` for transaction start time, `clock_timestamp()` for current wall-clock time. - **Booleans**: `BOOLEAN` with `NOT NULL` constraint unless tri-state values are required. - **Enums**: `CREATE TYPE ... AS ENUM` for small, stable sets (e.g. US states, days of week). For business-logic-driven and evolving values (e.g. order statuses) → use TEXT (or INT) + CHECK or lookup table. - **Arrays**: `TEXT[]`, `INTEGER[]`, etc. Use for ordered lists where you query elements. Index with **GIN** for containment (`@>`, `<@`) and overlap (`&&`) queries. Access: `arr[1]` (1-indexed), `arr[1:3]` (slicing). Good for tags, categories; avoid for relations—use junction tables instead. Literal syntax: `'{val1,val2}'` or `ARRAY[val1,val2]`. - **Range types**: `daterange`, `numrange`, `tstzrange` for intervals. Support overlap (`&&`), containment (`@>`), operators. Index with **GiST**. Good for scheduling, versioning, numeric ranges. Pick a bounds scheme and use it consistently; prefer `[)` (inclusive/exclusive) by default. - **Network types**: `INET` for IP addresses, `CIDR` for network ranges, `MACADDR` for MAC addresses. Support network operators (`<<`, `>>`, `&&`). - **Geometric types**: `POINT`, `LINE`, `POLYGON`, `CIRCLE` for 2D spatial data. Index with **GiST**. Consider **PostGIS** for advanced spatial features. - **Text search**: `TSVECTOR` for full-text search documents, `TSQUERY` for search queries. Index `tsvector` with **GIN**. Always specify language: `to_tsvector('english', col)` and `to_tsquery('english', 'query')`. Never use single-argument versions. This applies to both index expressions and queries. - **Domain types**: `CREATE DOMAIN email AS TEXT CHECK (VALUE ~ '^[^@]+@[^@]+$')` for reusable custom types with validation. Enforces constraints across tables. - **Composite types**: `CREATE TYPE address AS (street TEXT, city TEXT, zip TEXT)` for structured data within columns. Access with `(col).field` syntax. - **JSONB**: preferred over JSON; index with **GIN**. Use only for optional/semi-structured attrs. ONLY use JSON if the original ordering of the contents MUST be preserved. - **Vector types**: `vector` type by `pgvector` for vector similarity search for embeddings. ### Do not use the following data types - DO NOT use `timestamp` (without time zone); DO use `timestamptz` instead. - DO NOT use `char(n)` or `varchar(n)`; DO use `text` instead. - DO NOT use `money` type; DO use `numeric` instead. - DO NOT use `timetz` type; DO use `timestamptz` instead. - DO NOT use `timestamptz(0)` or any other precision specification; DO use `timestamptz` instead - DO NOT use `serial` type; DO use `generated always as identity` instead. ## Table Types - **Regular**: default; fully durable, logged. - **TEMPORARY**: session-scoped, auto-dropped, not logged. Faster for scratch work. - **UNLOGGED**: persistent but not crash-safe. Faster writes; good for caches/staging. ## Row-Level Security Enable with `ALTER TABLE tbl ENABLE ROW LEVEL SECURITY`. Create policies: `CREATE POLICY user_access ON orders FOR SELECT TO app_users USING (user_id = current_user_id())`. Built-in user-based access control at the row level. ## Constraints - **PK**: implicit UNIQUE + NOT NULL; creates a B-tree index. - **FK**: specify `ON DELETE/UPDATE` action (`CASCADE`, `RESTRICT`, `SET NULL`, `SET DEFAULT`). Add explicit index on referencing column—speeds up joins and prevents locking issues on parent deletes/updates. Use `DEFERRABLE INITIALLY DEFERRED` for circular FK dependencies checked at transaction end. - **UNIQUE**: creates a B-tree index; allows multiple NULLs unless `NULLS NOT DISTINCT` (PG15+). Standard behavior: `(1, NULL)` and `(1, NULL)` are allowed. With `NULLS NOT DISTINCT`: only one `(1, NULL)` allowed. Prefer `NULLS NOT DISTINCT` unless you specifically need duplicate NULLs. - **CHECK**: row-local constraints; NULL values pass the check (three-valued logic). Example: `CHECK (price > 0)` allows NULL prices. Combine with `NOT NULL` to enforce: `price NUMERIC NOT NULL CHECK (price > 0)`. - **EXCLUDE**: prevents overlapping values using operators. `EXCLUDE USING gist (room_id WITH =, booking_period WITH &&)` prevents double-booking rooms. Requires appropriate index type (often GiST). ## Indexing - **B-tree**: default for equality/range queries (`=`, `<`, `>`, `BETWEEN`, `ORDER BY`) - **Composite**: order matters—index used if equality on leftmost prefix (`WHERE a = ? AND b > ?` uses index on `(a,b)`, but `WHERE b = ?` does not). Put most selective/frequently filtered columns first. - **Covering**: `CREATE INDEX ON tbl (id) INCLUDE (name, email)` - includes non-key columns for index-only scans without visiting table. - **Partial**: for hot subsets (`WHERE status = 'active'` → `CREATE INDEX ON tbl (user_id) WHERE status = 'active'`). Any query with `status = 'active'` can use this index. - **Expression**: for computed search keys (`CREATE INDEX ON tbl (LOWER(email))`). Expression must match exactly in WHERE clause: `WHERE LOWER(email) = 'user@example.com'`. - **GIN**: JSONB containment/existence, arrays (`@>`, `?`), full-text search (`@@`) - **GiST**: ranges, geometry, exclusion constraints - **BRIN**: very large, naturally ordered data (time-series)—minimal storage overhead. Effective when row order on disk correlates with indexed column (insertion order or after `CLUSTER`). ## Partitioning - Use for very large tables (>100M rows) where queries consistently filter on partition key (often time/date). - Alternate use: use for tables where data maintenance tasks dictates e.g. data pruned or bulk replaced periodically - **RANGE**: common for time-series (`PARTITION BY RANGE (created_at)`). Create partitions: `CREATE TABLE logs_2024_01 PARTITION OF logs FOR VALUES FROM ('2024-01-01') TO ('2024-02-01')`. **TimescaleDB** automates time-based or ID-based partitioning with retention policies and compression. - **LIST**: for discrete values (`PARTITION BY LIST (region)`). Example: `FOR VALUES IN ('us-east', 'us-west')`. - **HASH**: for even distribution when no natural key (`PARTITION BY HASH (user_id)`). Creates N partitions with modulus. - **Constraint exclusion**: requires `CHECK` constraints on partitions for query planner to prune. Auto-created for declarative partitioning (PG10+). - Prefer declarative partitioning or hypertables. Do NOT use table inheritance. - **Limitations**: no global UNIQUE constraints—include partition key in PK/UNIQUE. FKs from partitioned tables not supported; use triggers. ## Special Considerations ### Update-Heavy Tables - **Separate hot/cold columns**—put frequently updated columns in separate table to minimize bloat. - **Use `fillfactor=90`** to leave space for HOT updates that avoid index maintenance. - **Avoid updating indexed columns**—prevents beneficial HOT updates. - **Partition by update patterns**—separate frequently updated rows in a different partition from stable data. ### Insert-Heavy Workloads - **Minimize indexes**—only create what you query; every index slows inserts. - **Use `COPY` or multi-row `INSERT`** instead of single-row inserts. - **UNLOGGED tables** for rebuildable staging data—much faster writes. - **Defer index creation** for bulk loads—>drop index, load data, recreate indexes. - **Partition by time/hash** to distribute load. **TimescaleDB** automates partitioning and compression of insert-heavy data. - **Use a natural key for primary key** such as a (timestamp, device_id) if enforcing global uniqueness is important many insert-heavy tables don't need a primary key at all. - If you do need a surrogate key, **Prefer `BIGINT GENERATED ALWAYS AS IDENTITY` over `UUID`**. ### Upsert-Friendly Design - **Requires UNIQUE index** on conflict target columns—`ON CONFLICT (col1, col2)` needs exact matching unique index (partial indexes don't work). - **Use `EXCLUDED.column`** to reference would-be-inserted values; only update columns that actually changed to reduce write overhead. - **`DO NOTHING` faster** than `DO UPDATE` when no actual update needed. ### Safe Schema Evolution - **Transactional DDL**: most DDL operations can run in transactions and be rolled back—`BEGIN; ALTER TABLE...; ROLLBACK;` for safe testing. - **Concurrent index creation**: `CREATE INDEX CONCURRENTLY` avoids blocking writes but can't run in transactions. - **Volatile defaults cause rewrites**: adding `NOT NULL` columns with volatile defaults (e.g., `now()`, `gen_random_uuid()`) rewrites entire table. Non-volatile defaults are fast. - **Drop constraints before columns**: `ALTER TABLE DROP CONSTRAINT` then `DROP COLUMN` to avoid dependency issues. - **Function signature changes**: `CREATE OR REPLACE` with different arguments creates overloads, not replacements. DROP old version if no overload desired. ## Generated Columns - `... GENERATED ALWAYS AS (<expr>) STORED` for computed, indexable fields. PG18+ adds `VIRTUAL` columns (computed on read, not stored). ## Extensions - **`pgcrypto`**: `crypt()` for password hashing. - **`uuid-ossp`**: alternative UUID functions; prefer `pgcrypto` for new projects. - **`pg_trgm`**: fuzzy text search with `%` operator, `similarity()` function. Index with GIN for `LIKE '%pattern%'` acceleration. - **`citext`**: case-insensitive text type. Prefer expression indexes on `LOWER(col)` unless you need case-insensitive constraints. - **`btree_gin`/`btree_gist`**: enable mixed-type indexes (e.g., GIN index on both JSONB and text columns). - **`hstore`**: key-value pairs; mostly superseded by JSONB but useful for simple string mappings. - **`timescaledb`**: essential for time-series—automated partitioning, retention, compression, continuous aggregates. - **`postgis`**: comprehensive geospatial support beyond basic geometric types—essential for location-based applications. - **`pgvector`**: vector similarity search for embeddings. - **`pgaudit`**: audit logging for all database activity. ## JSONB Guidance - Prefer `JSONB` with **GIN** index. - Default: `CREATE INDEX ON tbl USING GIN (jsonb_col);` → accelerates: - **Containment** `jsonb_col @> '{"k":"v"}'` - **Key existence** `jsonb_col ? 'k'`, **any/all keys** `?\|`, `?&` - **Path containment** on nested docs - **Disjunction** `jsonb_col @> ANY(ARRAY['{"status":"active"}', '{"status":"pending"}'])` - Heavy `@>` workloads: consider opclass `jsonb_path_ops` for smaller/faster containment-only indexes: - `CREATE INDEX ON tbl USING GIN (jsonb_col jsonb_path_ops);` - **Trade-off**: loses support for key existence (`?`, `?|`, `?&`) queries—only supports containment (`@>`) - Equality/range on a specific scalar field: extract and index with B-tree (generated column or expression): - `ALTER TABLE tbl ADD COLUMN price INT GENERATED ALWAYS AS ((jsonb_col->>'price')::INT) STORED;` - `CREATE INDEX ON tbl (price);` - Prefer queries like `WHERE price BETWEEN 100 AND 500` (uses B-tree) over `WHERE (jsonb_col->>'price')::INT BETWEEN 100 AND 500` without index. - Arrays inside JSONB: use GIN + `@>` for containment (e.g., tags). Consider `jsonb_path_ops` if only doing containment. - Keep core relations in tables; use JSONB for optional/variable attributes. - Use constraints to limit allowed JSONB values in a column e.g. `config JSONB NOT NULL CHECK(jsonb_typeof(config) = 'object')` ## Examples ### Users ```sql CREATE TABLE users ( user_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, email TEXT NOT NULL UNIQUE, name TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE UNIQUE INDEX ON users (LOWER(email)); CREATE INDEX ON users (created_at); ``` ### Orders ```sql CREATE TABLE orders ( order_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, user_id BIGINT NOT NULL REFERENCES users(user_id), status TEXT NOT NULL DEFAULT 'PENDING' CHECK (status IN ('PENDING','PAID','CANCELED')), total NUMERIC(10,2) NOT NULL CHECK (total > 0), created_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE INDEX ON orders (user_id); CREATE INDEX ON orders (created_at); ``` ### JSONB ```sql CREATE TABLE profiles ( user_id BIGINT PRIMARY KEY REFERENCES users(user_id), attrs JSONB NOT NULL DEFAULT '{}', theme TEXT GENERATED ALWAYS AS (attrs->>'theme') STORED ); CREATE INDEX profiles_attrs_gin ON profiles USING GIN (attrs); ```
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

architecture-decision-records

Write and maintain Architecture Decision Records (ADRs) following

coding
⭐1
# Architecture Decision Records Comprehensive patterns for creating, maintaining, and managing Architecture Decision Records (ADRs) that capture the context and rationale behind significant technical decisions. ## When to Use This Skill - Making significant architectural decisions - Documenting technology choices - Recording design trade-offs - Onboarding new team members - Reviewing historical decisions - Establishing decision-making processes ## Core Concepts ### 1. What is an ADR? An Architecture Decision Record captures: - **Context**: Why we needed to make a decision - **Decision**: What we decided - **Consequences**: What happens as a result ### 2. When to Write an ADR | Write ADR | Skip ADR | | -------------------------- | ---------------------- | | New framework adoption | Minor version upgrades | | Database technology choice | Bug fixes | | API design patterns | Implementation details | | Security architecture | Routine maintenance | | Integration patterns | Configuration changes | ### 3. ADR Lifecycle ``` Proposed → Accepted → Deprecated → Superseded ↓ Rejected ``` ## Templates ### Template 1: Standard ADR (MADR Format) ```markdown # ADR-0001: Use PostgreSQL as Primary Database ## Status Accepted ## Context We need to select a primary database for our new e-commerce platform. The system will handle: - ~10,000 concurrent users - Complex product catalog with hierarchical categories - Transaction processing for orders and payments - Full-text search for products - Geospatial queries for store locator The team has experience with MySQL, PostgreSQL, and MongoDB. We need ACID compliance for financial transactions. ## Decision Drivers - **Must have ACID compliance** for payment processing - **Must support complex queries** for reporting - **Should support full-text search** to reduce infrastructure complexity - **Should have good JSON support** for flexible product attributes - **Team familiarity** reduces onboarding time ## Considered Options ### Option 1: PostgreSQL - **Pros**: ACID compliant, excellent JSON support (JSONB), built-in full-text search, PostGIS for geospatial, team has experience - **Cons**: Slightly more complex replication setup than MySQL ### Option 2: MySQL - **Pros**: Very familiar to team, simple replication, large community - **Cons**: Weaker JSON support, no built-in full-text search (need Elasticsearch), no geospatial without extensions ### Option 3: MongoDB - **Pros**: Flexible schema, native JSON, horizontal scaling - **Cons**: No ACID for multi-document transactions (at decision time), team has limited experience, requires schema design discipline ## Decision We will use **PostgreSQL 15** as our primary database. ## Rationale PostgreSQL provides the best balance of: 1. **ACID compliance** essential for e-commerce transactions 2. **Built-in capabilities** (full-text search, JSONB, PostGIS) reduce infrastructure complexity 3. **Team familiarity** with SQL databases reduces learning curve 4. **Mature ecosystem** with excellent tooling and community support The slight complexity in replication is outweighed by the reduction in additional services (no separate Elasticsearch needed). ## Consequences ### Positive - Single database handles transactions, search, and geospatial queries - Reduced operational complexity (fewer services to manage) - Strong consistency guarantees for financial data - Team can leverage existing SQL expertise ### Negative - Need to learn PostgreSQL-specific features (JSONB, full-text search syntax) - Vertical scaling limits may require read replicas sooner - Some team members need PostgreSQL-specific training ### Risks - Full-text search may not scale as well as dedicated search engines - Mitigation: Design for potential Elasticsearch addition if needed ## Implementation Notes - Use JSONB for flexible product attributes - Implement connection pooling with PgBouncer - Set up streaming replication for read replicas - Use pg_trgm extension for fuzzy search ## Related Decisions - ADR-0002: Caching Strategy (Redis) - complements database choice - ADR-0005: Search Architecture - may supersede if Elasticsearch needed ## References - [PostgreSQL JSON Documentation](https://www.postgresql.org/docs/current/datatype-json.html) - [PostgreSQL Full Text Search](https://www.postgresql.org/docs/current/textsearch.html) - Internal: Performance benchmarks in `/docs/benchmarks/database-comparison.md` ``` ### Template 2: Lightweight ADR ```markdown # ADR-0012: Adopt TypeScript for Frontend Development **Status**: Accepted **Date**: 2024-01-15 **Deciders**: @alice, @bob, @charlie ## Context Our React codebase has grown to 50+ components with increasing bug reports related to prop type mismatches and undefined errors. PropTypes provide runtime-only checking. ## Decision Adopt TypeScript for all new frontend code. Migrate existing code incrementally. ## Consequences **Good**: Catch type errors at compile time, better IDE support, self-documenting code. **Bad**: Learning curve for team, initial slowdown, build complexity increase. **Mitigations**: TypeScript training sessions, allow gradual adoption with `allowJs: true`. ``` ### Template 3: Y-Statement Format ```markdown # ADR-0015: API Gateway Selection In the context of **building a microservices architecture**, facing **the need for centralized API management, authentication, and rate limiting**, we decided for **Kong Gateway** and against **AWS API Gateway and custom Nginx solution**, to achieve **vendor independence, plugin extensibility, and team familiarity with Lua**, accepting that **we need to manage Kong infrastructure ourselves**. ``` ### Template 4: ADR for Deprecation ```markdown # ADR-0020: Deprecate MongoDB in Favor of PostgreSQL ## Status Accepted (Supersedes ADR-0003) ## Context ADR-0003 (2021) chose MongoDB for user profile storage due to schema flexibility needs. Since then: - MongoDB's multi-document transactions remain problematic for our use case - Our schema has stabilized and rarely changes - We now have PostgreSQL expertise from other services - Maintaining two databases increases operational burden ## Decision Deprecate MongoDB and migrate user profiles to PostgreSQL. ## Migration Plan 1. **Phase 1** (Week 1-2): Create PostgreSQL schema, dual-write enabled 2. **Phase 2** (Week 3-4): Backfill historical data, validate consistency 3. **Phase 3** (Week 5): Switch reads to PostgreSQL, monitor 4. **Phase 4** (Week 6): Remove MongoDB writes, decommission ## Consequences ### Positive - Single database technology reduces operational complexity - ACID transactions for user data - Team can focus PostgreSQL expertise ### Negative - Migration effort (~4 weeks) - Risk of data issues during migration - Lose some schema flexibility ## Lessons Learned Document from ADR-0003 experience: - Schema flexibility benefits were overestimated - Operational cost of multiple databases was underestimated - Consider long-term maintenance in technology decisions ``` ### Template 5: Request for Comments (RFC) Style ```markdown # RFC-0025: Adopt Event Sourcing for Order Management ## Summary Propose adopting event sourcing pattern for the order management domain to improve auditability, enable temporal queries, and support business analytics. ## Motivation Current challenges: 1. Audit requirements need complete order history 2. "What was the order state at time X?" queries are impossible 3. Analytics team needs event stream for real-time dashboards 4. Order state reconstruction for customer support is manual ## Detailed Design ### Event Store ``` OrderCreated { orderId, customerId, items[], timestamp } OrderItemAdded { orderId, item, timestamp } OrderItemRemoved { orderId, itemId, timestamp } PaymentReceived { orderId, amount, paymentId, timestamp } OrderShipped { orderId, trackingNumber, timestamp } ``` ### Projections - **CurrentOrderState**: Materialized view for queries - **OrderHistory**: Complete timeline for audit - **DailyOrderMetrics**: Analytics aggregation ### Technology - Event Store: EventStoreDB (purpose-built, handles projections) - Alternative considered: Kafka + custom projection service ## Drawbacks - Learning curve for team - Increased complexity vs. CRUD - Need to design events carefully (immutable once stored) - Storage growth (events never deleted) ## Alternatives 1. **Audit tables**: Simpler but doesn't enable temporal queries 2. **CDC from existing DB**: Complex, doesn't change data model 3. **Hybrid**: Event source only for order state changes ## Unresolved Questions - [ ] Event schema versioning strategy - [ ] Retention policy for events - [ ] Snapshot frequency for performance ## Implementation Plan 1. Prototype with single order type (2 weeks) 2. Team training on event sourcing (1 week) 3. Full implementation and migration (4 weeks) 4. Monitoring and optimization (ongoing) ## References - [Event Sourcing by Martin Fowler](https://martinfowler.com/eaaDev/EventSourcing.html) - [EventStoreDB Documentation](https://www.eventstore.com/docs) ``` ## ADR Management ### Directory Structure ``` docs/ ā”œā”€ā”€ adr/ │ ā”œā”€ā”€ README.md # Index and guidelines │ ā”œā”€ā”€ template.md # Team's ADR template │ ā”œā”€ā”€ 0001-use-postgresql.md │ ā”œā”€ā”€ 0002-caching-strategy.md │ ā”œā”€ā”€ 0003-mongodb-user-profiles.md # [DEPRECATED] │ └── 0020-deprecate-mongodb.md # Supersedes 0003 ``` ### ADR Index (README.md) ```markdown # Architecture Decision Records This directory contains Architecture Decision Records (ADRs) for [Project Name]. ## Index | ADR | Title | Status | Date | | ------------------------------------- | ---------------------------------- | ---------- | ---------- | | [0001](0001-use-postgresql.md) | Use PostgreSQL as Primary Database | Accepted | 2024-01-10 | | [0002](0002-caching-strategy.md) | Caching Strategy with Redis | Accepted | 2024-01-12 | | [0003](0003-mongodb-user-profiles.md) | MongoDB for User Profiles | Deprecated | 2023-06-15 | | [0020](0020-deprecate-mongodb.md) | Deprecate MongoDB | Accepted | 2024-01-15 | ## Creating a New ADR 1. Copy `template.md` to `NNNN-title-with-dashes.md` 2. Fill in the template 3. Submit PR for review 4. Update this index after approval ## ADR Status - **Proposed**: Under discussion - **Accepted**: Decision made, implementing - **Deprecated**: No longer relevant - **Superseded**: Replaced by another ADR - **Rejected**: Considered but not adopted ``` ### Automation (adr-tools) ```bash # Install adr-tools brew install adr-tools # Initialize ADR directory adr init docs/adr # Create new ADR adr new "Use PostgreSQL as Primary Database" # Supersede an ADR adr new -s 3 "Deprecate MongoDB in Favor of PostgreSQL" # Generate table of contents adr generate toc > docs/adr/README.md # Link related ADRs adr link 2 "Complements" 1 "Is complemented by" ``` ## Review Process ```markdown ## ADR Review Checklist ### Before Submission - [ ] Context clearly explains the problem - [ ] All viable options considered - [ ] Pros/cons balanced and honest - [ ] Consequences (positive and negative) documented - [ ] Related ADRs linked ### During Review - [ ] At least 2 senior engineers reviewed - [ ] Affected teams consulted - [ ] Security implications considered - [ ] Cost implications documented - [ ] Reversibility assessed ### After Acceptance - [ ] ADR index updated - [ ] Team notified - [ ] Implementation tickets created - [ ] Related documentation updated ``` ## Best Practices ### Do's - **Write ADRs early** - Before implementation starts - **Keep them short** - 1-2 pages maximum - **Be honest about trade-offs** - Include real cons - **Link related decisions** - Build decision graph - **Update status** - Deprecate when superseded ### Don'ts - **Don't change accepted ADRs** - Write new ones to supersede - **Don't skip context** - Future readers need background - **Don't hide failures** - Rejected decisions are valuable - **Don't be vague** - Specific decisions, specific consequences - **Don't forget implementation** - ADR without action is waste ## Resources - [Documenting Architecture Decisions (Michael Nygard)](https://cognitect.com/blog/2011/11/15/documenting-architecture-decisions) - [MADR Template](https://adr.github.io/madr/) - [ADR GitHub Organization](https://adr.github.io/) - [adr-tools](https://github.com/npryce/adr-tools)
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

dependency-upgrade

Manage major dependency version upgrades with compatibility

coding
⭐1
# Dependency Upgrade Master major dependency version upgrades, compatibility analysis, staged upgrade strategies, and comprehensive testing approaches. ## When to Use This Skill - Upgrading major framework versions - Updating security-vulnerable dependencies - Modernizing legacy dependencies - Resolving dependency conflicts - Planning incremental upgrade paths - Testing compatibility matrices - Automating dependency updates ## Semantic Versioning Review ``` MAJOR.MINOR.PATCH (e.g., 2.3.1) MAJOR: Breaking changes MINOR: New features, backward compatible PATCH: Bug fixes, backward compatible ^2.3.1 = >=2.3.1 <3.0.0 (minor updates) ~2.3.1 = >=2.3.1 <2.4.0 (patch updates) 2.3.1 = exact version ``` ## Dependency Analysis ### Audit Dependencies ```bash # npm npm outdated npm audit npm audit fix # yarn yarn outdated yarn audit # Check for major updates npx npm-check-updates npx npm-check-updates -u # Update package.json ``` ### Analyze Dependency Tree ```bash # See why a package is installed npm ls package-name yarn why package-name # Find duplicate packages npm dedupe yarn dedupe # Visualize dependencies npx madge --image graph.png src/ ``` ## Compatibility Matrix ```javascript // compatibility-matrix.js const compatibilityMatrix = { react: { "16.x": { "react-dom": "^16.0.0", "react-router-dom": "^5.0.0", "@testing-library/react": "^11.0.0", }, "17.x": { "react-dom": "^17.0.0", "react-router-dom": "^5.0.0 || ^6.0.0", "@testing-library/react": "^12.0.0", }, "18.x": { "react-dom": "^18.0.0", "react-router-dom": "^6.0.0", "@testing-library/react": "^13.0.0", }, }, }; function checkCompatibility(packages) { // Validate package versions against matrix } ``` ## Staged Upgrade Strategy ### Phase 1: Planning ```bash # 1. Identify current versions npm list --depth=0 # 2. Check for breaking changes # Read CHANGELOG.md and MIGRATION.md # 3. Create upgrade plan echo "Upgrade order: 1. TypeScript 2. React 3. React Router 4. Testing libraries 5. Build tools" > UPGRADE_PLAN.md ``` ### Phase 2: Incremental Updates ```bash # Don't upgrade everything at once! # Step 1: Update TypeScript npm install typescript@latest # Test npm run test npm run build # Step 2: Update React (one major version at a time) npm install react@17 react-dom@17 # Test again npm run test # Step 3: Continue with other packages npm install react-router-dom@6 # And so on... ``` ### Phase 3: Validation ```javascript // tests/compatibility.test.js describe("Dependency Compatibility", () => { it("should have compatible React versions", () => { const reactVersion = require("react/package.json").version; const reactDomVersion = require("react-dom/package.json").version; expect(reactVersion).toBe(reactDomVersion); }); it("should not have peer dependency warnings", () => { // Run npm ls and check for warnings }); }); ``` ## Breaking Change Handling ### Identifying Breaking Changes ```bash # Check the changelog directly curl https://raw.githubusercontent.com/facebook/react/master/CHANGELOG.md ``` ### Codemod for Automated Fixes ```bash # Run jscodeshift with transform URL npx jscodeshift -t <transform-url> <path> # Example: Rename unsafe lifecycle methods npx jscodeshift -t https://raw.githubusercontent.com/reactjs/react-codemod/master/transforms/rename-unsafe-lifecycles.js src/ # For TypeScript files npx jscodeshift -t https://raw.githubusercontent.com/reactjs/react-codemod/master/transforms/rename-unsafe-lifecycles.js --parser=tsx src/ # Dry run to preview changes npx jscodeshift -t https://raw.githubusercontent.com/reactjs/react-codemod/master/transforms/rename-unsafe-lifecycles.js --dry src/ ``` ### Custom Migration Script ```javascript // migration-script.js const fs = require("fs"); const glob = require("glob"); glob("src/**/*.tsx", (err, files) => { files.forEach((file) => { let content = fs.readFileSync(file, "utf8"); // Replace old API with new API content = content.replace( /componentWillMount/g, "UNSAFE_componentWillMount", ); // Update imports content = content.replace( /import { Component } from 'react'/g, "import React, { Component } from 'react'", ); fs.writeFileSync(file, content); }); }); ``` ## Testing Strategy ### Unit Tests ```javascript // Ensure tests pass before and after upgrade npm run test // Update test utilities if needed npm install @testing-library/react@latest ``` ### Integration Tests ```javascript // tests/integration/app.test.js describe("App Integration", () => { it("should render without crashing", () => { render(<App />); }); it("should handle navigation", () => { const { getByText } = render(<App />); fireEvent.click(getByText("Navigate")); expect(screen.getByText("New Page")).toBeInTheDocument(); }); }); ``` ### Visual Regression Tests ```javascript // visual-regression.test.js describe("Visual Regression", () => { it("should match snapshot", () => { const { container } = render(<App />); expect(container.firstChild).toMatchSnapshot(); }); }); ``` ### E2E Tests ```javascript // cypress/e2e/app.cy.js describe("E2E Tests", () => { it("should complete user flow", () => { cy.visit("/"); cy.get('[data-testid="login"]').click(); cy.get('input[name="email"]').type("user@example.com"); cy.get('button[type="submit"]').click(); cy.url().should("include", "/dashboard"); }); }); ``` ## Automated Dependency Updates ### Renovate Configuration ```json // renovate.json { "extends": ["config:base"], "packageRules": [ { "matchUpdateTypes": ["minor", "patch"], "automerge": true }, { "matchUpdateTypes": ["major"], "automerge": false, "labels": ["major-update"] } ], "schedule": ["before 3am on Monday"], "timezone": "America/New_York" } ``` ### Dependabot Configuration ```yaml # .github/dependabot.yml version: 2 updates: - package-ecosystem: "npm" directory: "/" schedule: interval: "weekly" open-pull-requests-limit: 5 reviewers: - "team-leads" commit-message: prefix: "chore" include: "scope" ``` ## Rollback Plan ```javascript // rollback.sh #!/bin/bash # Save current state git stash git checkout -b upgrade-branch # Attempt upgrade npm install package@latest # Run tests if npm run test; then echo "Upgrade successful" git add package.json package-lock.json git commit -m "chore: upgrade package" else echo "Upgrade failed, rolling back" git checkout main git branch -D upgrade-branch npm install # Restore from package-lock.json fi ``` ## Common Upgrade Patterns ### Lock File Management ```bash # npm npm install --package-lock-only # Update lock file only npm ci # Clean install from lock file # yarn yarn install --frozen-lockfile # CI mode yarn upgrade-interactive # Interactive upgrades ``` ### Peer Dependency Resolution ```bash # npm 7+: strict peer dependencies npm install --legacy-peer-deps # Ignore peer deps # npm 8+: override peer dependencies npm install --force ``` ### Workspace Upgrades ```bash # Update all workspace packages npm install --workspaces # Update specific workspace npm install package@latest --workspace=packages/app ``` ## Resources - **references/semver.md**: Semantic versioning guide - **references/compatibility-matrix.md**: Common compatibility issues - **references/staged-upgrades.md**: Incremental upgrade strategies - **references/testing-strategy.md**: Comprehensive testing approaches - **assets/upgrade-checklist.md**: Step-by-step checklist - **assets/compatibility-matrix.csv**: Version compatibility table - **scripts/audit-dependencies.sh**: Dependency audit script ## Best Practices 1. **Read Changelogs**: Understand what changed 2. **Upgrade Incrementally**: One major version at a time 3. **Test Thoroughly**: Unit, integration, E2E tests 4. **Check Peer Dependencies**: Resolve conflicts early 5. **Use Lock Files**: Ensure reproducible installs 6. **Automate Updates**: Use Renovate or Dependabot 7. **Monitor**: Watch for runtime errors post-upgrade 8. **Document**: Keep upgrade notes ## Upgrade Checklist ```markdown Pre-Upgrade: - [ ] Review current dependency versions - [ ] Read changelogs for breaking changes - [ ] Create feature branch - [ ] Backup current state (git tag) - [ ] Run full test suite (baseline) During Upgrade: - [ ] Upgrade one dependency at a time - [ ] Update peer dependencies - [ ] Fix TypeScript errors - [ ] Update tests if needed - [ ] Run test suite after each upgrade - [ ] Check bundle size impact Post-Upgrade: - [ ] Full regression testing - [ ] Performance testing - [ ] Update documentation - [ ] Deploy to staging - [ ] Monitor for errors - [ ] Deploy to production ``` ## Common Pitfalls - Upgrading all dependencies at once - Not testing after each upgrade - Ignoring peer dependency warnings - Forgetting to update lock file - Not reading breaking change notes - Skipping major versions - Not having rollback plan
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

gdpr-data-handling

Implement GDPR-compliant data handling with consent management,

data
⭐1
# GDPR Data Handling Practical implementation guide for GDPR-compliant data processing, consent management, and privacy controls. ## When to Use This Skill - Building systems that process EU personal data - Implementing consent management - Handling data subject requests (DSRs) - Conducting GDPR compliance reviews - Designing privacy-first architectures - Creating data processing agreements ## Core Concepts ### 1. Personal Data Categories | Category | Examples | Protection Level | | ---------------------- | --------------------------- | ------------------ | | **Basic** | Name, email, phone | Standard | | **Sensitive (Art. 9)** | Health, religion, ethnicity | Explicit consent | | **Criminal (Art. 10)** | Convictions, offenses | Official authority | | **Children's** | Under 16 data | Parental consent | ### 2. Legal Bases for Processing ``` Article 6 - Lawful Bases: ā”œā”€ā”€ Consent: Freely given, specific, informed ā”œā”€ā”€ Contract: Necessary for contract performance ā”œā”€ā”€ Legal Obligation: Required by law ā”œā”€ā”€ Vital Interests: Protecting someone's life ā”œā”€ā”€ Public Interest: Official functions └── Legitimate Interest: Balanced against rights ``` ### 3. Data Subject Rights ``` Right to Access (Art. 15) ─┐ Right to Rectification (Art. 16) │ Right to Erasure (Art. 17) │ Must respond Right to Restrict (Art. 18) │ within 1 month Right to Portability (Art. 20) │ Right to Object (Art. 21) ā”€ā”˜ ``` ## Implementation Patterns ### Pattern 1: Consent Management ```javascript // Consent data model const consentSchema = { userId: String, consents: [ { purpose: String, // 'marketing', 'analytics', etc. granted: Boolean, timestamp: Date, source: String, // 'web_form', 'api', etc. version: String, // Privacy policy version ipAddress: String, // For proof userAgent: String, // For proof }, ], auditLog: [ { action: String, // 'granted', 'withdrawn', 'updated' purpose: String, timestamp: Date, source: String, }, ], }; // Consent service class ConsentManager { async recordConsent(userId, purpose, granted, metadata) { const consent = { purpose, granted, timestamp: new Date(), source: metadata.source, version: await this.getCurrentPolicyVersion(), ipAddress: metadata.ipAddress, userAgent: metadata.userAgent, }; // Store consent await this.db.consents.updateOne( { userId }, { $push: { consents: consent, auditLog: { action: granted ? "granted" : "withdrawn", purpose, timestamp: consent.timestamp, source: metadata.source, }, }, }, { upsert: true }, ); // Emit event for downstream systems await this.eventBus.emit("consent.changed", { userId, purpose, granted, timestamp: consent.timestamp, }); } async hasConsent(userId, purpose) { const record = await this.db.consents.findOne({ userId }); if (!record) return false; const latestConsent = record.consents .filter((c) => c.purpose === purpose) .sort((a, b) => b.timestamp - a.timestamp)[0]; return latestConsent?.granted === true; } async getConsentHistory(userId) { const record = await this.db.consents.findOne({ userId }); return record?.auditLog || []; } } ``` ```html <!-- GDPR-compliant consent UI --> <div class="consent-banner" role="dialog" aria-labelledby="consent-title"> <h2 id="consent-title">Cookie Preferences</h2> <p> We use cookies to improve your experience. Select your preferences below. </p> <form id="consent-form"> <!-- Necessary - always on, no consent needed --> <div class="consent-category"> <input type="checkbox" id="necessary" checked disabled /> <label for="necessary"> <strong>Necessary</strong> <span>Required for the website to function. Cannot be disabled.</span> </label> </div> <!-- Analytics - requires consent --> <div class="consent-category"> <input type="checkbox" id="analytics" name="analytics" /> <label for="analytics"> <strong>Analytics</strong> <span>Help us understand how you use our site.</span> </label> </div> <!-- Marketing - requires consent --> <div class="consent-category"> <input type="checkbox" id="marketing" name="marketing" /> <label for="marketing"> <strong>Marketing</strong> <span>Personalized ads based on your interests.</span> </label> </div> <div class="consent-actions"> <button type="button" id="accept-all">Accept All</button> <button type="button" id="reject-all">Reject All</button> <button type="submit">Save Preferences</button> </div> <p class="consent-links"> <a href="/privacy-policy">Privacy Policy</a> | <a href="/cookie-policy">Cookie Policy</a> </p> </form> </div> ``` ### Pattern 2: Data Subject Access Request (DSAR) ```python from datetime import datetime, timedelta from typing import Dict, List, Optional import json class DSARHandler: """Handle Data Subject Access Requests.""" RESPONSE_DEADLINE_DAYS = 30 EXTENSION_ALLOWED_DAYS = 60 # For complex requests def __init__(self, data_sources: List['DataSource']): self.data_sources = data_sources async def submit_request( self, request_type: str, # 'access', 'erasure', 'rectification', 'portability' user_id: str, verified: bool, details: Optional[Dict] = None ) -> str: """Submit a new DSAR.""" request = { 'id': self.generate_request_id(), 'type': request_type, 'user_id': user_id, 'status': 'pending_verification' if not verified else 'processing', 'submitted_at': datetime.utcnow(), 'deadline': datetime.utcnow() + timedelta(days=self.RESPONSE_DEADLINE_DAYS), 'details': details or {}, 'audit_log': [{ 'action': 'submitted', 'timestamp': datetime.utcnow(), 'details': 'Request received' }] } await self.db.dsar_requests.insert_one(request) await self.notify_dpo(request) return request['id'] async def process_access_request(self, request_id: str) -> Dict: """Process a data access request.""" request = await self.get_request(request_id) if request['type'] != 'access': raise ValueError("Not an access request") # Collect data from all sources user_data = {} for source in self.data_sources: try: data = await source.get_user_data(request['user_id']) user_data[source.name] = data except Exception as e: user_data[source.name] = {'error': str(e)} # Format response response = { 'request_id': request_id, 'generated_at': datetime.utcnow().isoformat(), 'data_categories': list(user_data.keys()), 'data': user_data, 'retention_info': await self.get_retention_info(), 'processing_purposes': await self.get_processing_purposes(), 'third_party_recipients': await self.get_recipients() } # Update request status await self.update_request(request_id, 'completed', response) return response async def process_erasure_request(self, request_id: str) -> Dict: """Process a right to erasure request.""" request = await self.get_request(request_id) if request['type'] != 'erasure': raise ValueError("Not an erasure request") results = {} exceptions = [] for source in self.data_sources: try: # Check for legal exceptions can_delete, reason = await source.can_delete(request['user_id']) if can_delete: await source.delete_user_data(request['user_id']) results[source.name] = 'deleted' else: exceptions.append({ 'source': source.name, 'reason': reason # e.g., 'legal retention requirement' }) results[source.name] = f'retained: {reason}' except Exception as e: results[source.name] = f'error: {str(e)}' response = { 'request_id': request_id, 'completed_at': datetime.utcnow().isoformat(), 'results': results, 'exceptions': exceptions } await self.update_request(request_id, 'completed', response) return response async def process_portability_request(self, request_id: str) -> bytes: """Generate portable data export.""" request = await self.get_request(request_id) user_data = await self.process_access_request(request_id) # Convert to machine-readable format (JSON) portable_data = { 'export_date': datetime.utcnow().isoformat(), 'format_version': '1.0', 'data': user_data['data'] } return json.dumps(portable_data, indent=2, default=str).encode() ``` ### Pattern 3: Data Retention ```python from datetime import datetime, timedelta from enum import Enum class RetentionBasis(Enum): CONSENT = "consent" CONTRACT = "contract" LEGAL_OBLIGATION = "legal_obligation" LEGITIMATE_INTEREST = "legitimate_interest" class DataRetentionPolicy: """Define and enforce data retention policies.""" POLICIES = { 'user_account': { 'retention_period_days': 365 * 3, # 3 years after last activity 'basis': RetentionBasis.CONTRACT, 'trigger': 'last_activity_date', 'archive_before_delete': True }, 'transaction_records': { 'retention_period_days': 365 * 7, # 7 years for tax 'basis': RetentionBasis.LEGAL_OBLIGATION, 'trigger': 'transaction_date', 'archive_before_delete': True, 'legal_reference': 'Tax regulations require 7 year retention' }, 'marketing_consent': { 'retention_period_days': 365 * 2, # 2 years 'basis': RetentionBasis.CONSENT, 'trigger': 'consent_date', 'archive_before_delete': False }, 'support_tickets': { 'retention_period_days': 365 * 2, 'basis': RetentionBasis.LEGITIMATE_INTEREST, 'trigger': 'ticket_closed_date', 'archive_before_delete': True }, 'analytics_data': { 'retention_period_days': 365, # 1 year 'basis': RetentionBasis.CONSENT, 'trigger': 'collection_date', 'archive_before_delete': False, 'anonymize_instead': True } } async def apply_retention_policies(self): """Run retention policy enforcement.""" for data_type, policy in self.POLICIES.items(): cutoff_date = datetime.utcnow() - timedelta( days=policy['retention_period_days'] ) if policy.get('anonymize_instead'): await self.anonymize_old_data(data_type, cutoff_date) else: if policy.get('archive_before_delete'): await self.archive_data(data_type, cutoff_date) await self.delete_old_data(data_type, cutoff_date) await self.log_retention_action(data_type, cutoff_date) async def anonymize_old_data(self, data_type: str, before_date: datetime): """Anonymize data instead of deleting.""" # Example: Replace identifying fields with hashes if data_type == 'analytics_data': await self.db.analytics.update_many( {'collection_date': {'$lt': before_date}}, {'$set': { 'user_id': None, 'ip_address': None, 'device_id': None, 'anonymized': True, 'anonymized_date': datetime.utcnow() }} ) ``` ### Pattern 4: Privacy by Design ```python class PrivacyFirstDataModel: """Example of privacy-by-design data model.""" # Separate PII from behavioral data user_profile_schema = { 'user_id': str, # UUID, not sequential 'email_hash': str, # Hashed for lookups 'created_at': datetime, # Minimal data collection 'preferences': { 'language': str, 'timezone': str } } # Encrypted at rest user_pii_schema = { 'user_id': str, 'email': str, # Encrypted 'name': str, # Encrypted 'phone': str, # Encrypted (optional) 'address': dict, # Encrypted (optional) 'encryption_key_id': str } # Pseudonymized behavioral data analytics_schema = { 'session_id': str, # Not linked to user_id 'pseudonym_id': str, # Rotating pseudonym 'events': list, 'device_category': str, # Generalized, not specific 'country': str, # Not city-level } class DataMinimization: """Implement data minimization principles.""" @staticmethod def collect_only_needed(form_data: dict, purpose: str) -> dict: """Filter form data to only fields needed for purpose.""" REQUIRED_FIELDS = { 'account_creation': ['email', 'password'], 'newsletter': ['email'], 'purchase': ['email', 'name', 'address', 'payment'], 'support': ['email', 'message'] } allowed = REQUIRED_FIELDS.get(purpose, []) return {k: v for k, v in form_data.items() if k in allowed} @staticmethod def generalize_location(ip_address: str) -> str: """Generalize IP to country level only.""" import geoip2.database reader = geoip2.database.Reader('GeoLite2-Country.mmdb') try: response = reader.country(ip_address) return response.country.iso_code except: return 'UNKNOWN' ``` ### Pattern 5: Breach Notification ```python from datetime import datetime from enum import Enum class BreachSeverity(Enum): LOW = "low" MEDIUM = "medium" HIGH = "high" CRITICAL = "critical" class BreachNotificationHandler: """Handle GDPR breach notification requirements.""" AUTHORITY_NOTIFICATION_HOURS = 72 AFFECTED_NOTIFICATION_REQUIRED_SEVERITY = BreachSeverity.HIGH async def report_breach( self, description: str, data_types: List[str], affected_count: int, severity: BreachSeverity ) -> dict: """Report and handle a data breach.""" breach = { 'id': self.generate_breach_id(), 'reported_at': datetime.utcnow(), 'description': description, 'data_types_affected': data_types, 'affected_individuals_count': affected_count, 'severity': severity.value, 'status': 'investigating', 'timeline': [{ 'event': 'breach_reported', 'timestamp': datetime.utcnow(), 'details': description }] } await self.db.breaches.insert_one(breach) # Immediate notifications await self.notify_dpo(breach) await self.notify_security_team(breach) # Authority notification required within 72 hours if self.requires_authority_notification(severity, data_types): breach['authority_notification_deadline'] = ( datetime.utcnow() + timedelta(hours=self.AUTHORITY_NOTIFICATION_HOURS) ) await self.schedule_authority_notification(breach) # Affected individuals notification if severity.value in [BreachSeverity.HIGH.value, BreachSeverity.CRITICAL.value]: await self.schedule_individual_notifications(breach) return breach def requires_authority_notification( self, severity: BreachSeverity, data_types: List[str] ) -> bool: """Determine if supervisory authority must be notified.""" # Always notify for sensitive data sensitive_types = ['health', 'financial', 'credentials', 'biometric'] if any(t in sensitive_types for t in data_types): return True # Notify for medium+ severity return severity in [BreachSeverity.MEDIUM, BreachSeverity.HIGH, BreachSeverity.CRITICAL] async def generate_authority_report(self, breach_id: str) -> dict: """Generate report for supervisory authority.""" breach = await self.get_breach(breach_id) return { 'organization': { 'name': self.config.org_name, 'contact': self.config.dpo_contact, 'registration': self.config.registration_number }, 'breach': { 'nature': breach['description'], 'categories_affected': breach['data_types_affected'], 'approximate_number_affected': breach['affected_individuals_count'], 'likely_consequences': self.assess_consequences(breach), 'measures_taken': await self.get_remediation_measures(breach_id), 'measures_proposed': await self.get_proposed_measures(breach_id) }, 'timeline': breach['timeline'], 'submitted_at': datetime.utcnow().isoformat() } ``` ## Compliance Checklist ```markdown ## GDPR Implementation Checklist ### Legal Basis - [ ] Documented legal basis for each processing activity - [ ] Consent mechanisms meet GDPR requirements - [ ] Legitimate interest assessments completed ### Transparency - [ ] Privacy policy is clear and accessible - [ ] Processing purposes clearly stated - [ ] Data retention periods documented ### Data Subject Rights - [ ] Access request process implemented - [ ] Erasure request process implemented - [ ] Portability export available - [ ] Rectification process available - [ ] Response within 30-day deadline ### Security - [ ] Encryption at rest implemented - [ ] Encryption in transit (TLS) - [ ] Access controls in place - [ ] Audit logging enabled ### Breach Response - [ ] Breach detection mechanisms - [ ] 72-hour notification process - [ ] Breach documentation system ### Documentation - [ ] Records of processing activities (Art. 30) - [ ] Data protection impact assessments - [ ] Data processing agreements with vendors ``` ## Best Practices ### Do's - **Minimize data collection** - Only collect what's needed - **Document everything** - Processing activities, legal bases - **Encrypt PII** - At rest and in transit - **Implement access controls** - Need-to-know basis - **Regular audits** - Verify compliance continuously ### Don'ts - **Don't pre-check consent boxes** - Must be opt-in - **Don't bundle consent** - Separate purposes separately - **Don't retain indefinitely** - Defi
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

incident-runbook-templates

Create structured incident response runbooks with step-by-step

coding
⭐1
# Incident Runbook Templates Production-ready templates for incident response runbooks covering detection, triage, mitigation, resolution, and communication. ## When to Use This Skill - Creating incident response procedures - Building service-specific runbooks - Establishing escalation paths - Documenting recovery procedures - Responding to active incidents - Onboarding on-call engineers ## Core Concepts ### 1. Incident Severity Levels | Severity | Impact | Response Time | Example | | -------- | -------------------------- | ----------------- | ----------------------- | | **SEV1** | Complete outage, data loss | 15 min | Production down | | **SEV2** | Major degradation | 30 min | Critical feature broken | | **SEV3** | Minor impact | 2 hours | Non-critical bug | | **SEV4** | Minimal impact | Next business day | Cosmetic issue | ### 2. Runbook Structure ``` 1. Overview & Impact 2. Detection & Alerts 3. Initial Triage 4. Mitigation Steps 5. Root Cause Investigation 6. Resolution Procedures 7. Verification & Rollback 8. Communication Templates 9. Escalation Matrix ``` ## Runbook Templates ### Template 1: Service Outage Runbook ````markdown # [Service Name] Outage Runbook ## Overview **Service**: Payment Processing Service **Owner**: Platform Team **Slack**: #payments-incidents **PagerDuty**: payments-oncall ## Impact Assessment - [ ] Which customers are affected? - [ ] What percentage of traffic is impacted? - [ ] Are there financial implications? - [ ] What's the blast radius? ## Detection ### Alerts - `payment_error_rate > 5%` (PagerDuty) - `payment_latency_p99 > 2s` (Slack) - `payment_success_rate < 95%` (PagerDuty) ### Dashboards - [Payment Service Dashboard](https://grafana/d/payments) - [Error Tracking](https://sentry.io/payments) - [Dependency Status](https://status.stripe.com) ## Initial Triage (First 5 Minutes) ### 1. Assess Scope ```bash # Check service health kubectl get pods -n payments -l app=payment-service # Check recent deployments kubectl rollout history deployment/payment-service -n payments # Check error rates curl -s "http://prometheus:9090/api/v1/query?query=sum(rate(http_requests_total{status=~'5..'}[5m]))" ``` ```` ### 2. Quick Health Checks - [ ] Can you reach the service? `curl -I https://api.company.com/payments/health` - [ ] Database connectivity? Check connection pool metrics - [ ] External dependencies? Check Stripe, bank API status - [ ] Recent changes? Check deploy history ### 3. Initial Classification | Symptom | Likely Cause | Go To Section | | -------------------- | ------------------- | ------------- | | All requests failing | Service down | Section 4.1 | | High latency | Database/dependency | Section 4.2 | | Partial failures | Code bug | Section 4.3 | | Spike in errors | Traffic surge | Section 4.4 | ## Mitigation Procedures ### 4.1 Service Completely Down ```bash # Step 1: Check pod status kubectl get pods -n payments # Step 2: If pods are crash-looping, check logs kubectl logs -n payments -l app=payment-service --tail=100 # Step 3: Check recent deployments kubectl rollout history deployment/payment-service -n payments # Step 4: ROLLBACK if recent deploy is suspect kubectl rollout undo deployment/payment-service -n payments # Step 5: Scale up if resource constrained kubectl scale deployment/payment-service -n payments --replicas=10 # Step 6: Verify recovery kubectl rollout status deployment/payment-service -n payments ``` ### 4.2 High Latency ```bash # Step 1: Check database connections kubectl exec -n payments deploy/payment-service -- \ curl localhost:8080/metrics | grep db_pool # Step 2: Check slow queries (if DB issue) psql -h $DB_HOST -U $DB_USER -c " SELECT pid, now() - query_start AS duration, query FROM pg_stat_activity WHERE state = 'active' AND duration > interval '5 seconds' ORDER BY duration DESC;" # Step 3: Kill long-running queries if needed psql -h $DB_HOST -U $DB_USER -c "SELECT pg_terminate_backend(pid);" # Step 4: Check external dependency latency curl -w "@curl-format.txt" -o /dev/null -s https://api.stripe.com/v1/health # Step 5: Enable circuit breaker if dependency is slow kubectl set env deployment/payment-service \ STRIPE_CIRCUIT_BREAKER_ENABLED=true -n payments ``` ### 4.3 Partial Failures (Specific Errors) ```bash # Step 1: Identify error pattern kubectl logs -n payments -l app=payment-service --tail=500 | \ grep -i error | sort | uniq -c | sort -rn | head -20 # Step 2: Check error tracking # Go to Sentry: https://sentry.io/payments # Step 3: If specific endpoint, enable feature flag to disable curl -X POST https://api.company.com/internal/feature-flags \ -d '{"flag": "DISABLE_PROBLEMATIC_FEATURE", "enabled": true}' # Step 4: If data issue, check recent data changes psql -h $DB_HOST -c " SELECT * FROM audit_log WHERE table_name = 'payment_methods' AND created_at > now() - interval '1 hour';" ``` ### 4.4 Traffic Surge ```bash # Step 1: Check current request rate kubectl top pods -n payments # Step 2: Scale horizontally kubectl scale deployment/payment-service -n payments --replicas=20 # Step 3: Enable rate limiting kubectl set env deployment/payment-service \ RATE_LIMIT_ENABLED=true \ RATE_LIMIT_RPS=1000 -n payments # Step 4: If attack, block suspicious IPs kubectl apply -f - <<EOF apiVersion: networking.k8s.io/v1 kind: NetworkPolicy metadata: name: block-suspicious namespace: payments spec: podSelector: matchLabels: app: payment-service ingress: - from: - ipBlock: cidr: 0.0.0.0/0 except: - 192.168.1.0/24 # Suspicious range EOF ``` ## Verification Steps ```bash # Verify service is healthy curl -s https://api.company.com/payments/health | jq # Verify error rate is back to normal curl -s "http://prometheus:9090/api/v1/query?query=sum(rate(http_requests_total{status=~'5..'}[5m]))" | jq '.data.result[0].value[1]' # Verify latency is acceptable curl -s "http://prometheus:9090/api/v1/query?query=histogram_quantile(0.99,sum(rate(http_request_duration_seconds_bucket[5m]))by(le))" | jq # Smoke test critical flows ./scripts/smoke-test-payments.sh ``` ## Rollback Procedures ```bash # Rollback Kubernetes deployment kubectl rollout undo deployment/payment-service -n payments # Rollback database migration (if applicable) ./scripts/db-rollback.sh $MIGRATION_VERSION # Rollback feature flag curl -X POST https://api.company.com/internal/feature-flags \ -d '{"flag": "NEW_PAYMENT_FLOW", "enabled": false}' ``` ## Escalation Matrix | Condition | Escalate To | Contact | | ----------------------------- | ------------------- | ------------------- | | > 15 min unresolved SEV1 | Engineering Manager | @manager (Slack) | | Data breach suspected | Security Team | #security-incidents | | Financial impact > $10k | Finance + Legal | @finance-oncall | | Customer communication needed | Support Lead | @support-lead | ## Communication Templates ### Initial Notification (Internal) ``` 🚨 INCIDENT: Payment Service Degradation Severity: SEV2 Status: Investigating Impact: ~20% of payment requests failing Start Time: [TIME] Incident Commander: [NAME] Current Actions: - Investigating root cause - Scaling up service - Monitoring dashboards Updates in #payments-incidents ``` ### Status Update ``` šŸ“Š UPDATE: Payment Service Incident Status: Mitigating Impact: Reduced to ~5% failure rate Duration: 25 minutes Actions Taken: - Rolled back deployment v2.3.4 → v2.3.3 - Scaled service from 5 → 10 replicas Next Steps: - Continuing to monitor - Root cause analysis in progress ETA to Resolution: ~15 minutes ``` ### Resolution Notification ``` āœ… RESOLVED: Payment Service Incident Duration: 45 minutes Impact: ~5,000 affected transactions Root Cause: Memory leak in v2.3.4 Resolution: - Rolled back to v2.3.3 - Transactions auto-retried successfully Follow-up: - Postmortem scheduled for [DATE] - Bug fix in progress ``` ```` ### Template 2: Database Incident Runbook ```markdown # Database Incident Runbook ## Quick Reference | Issue | Command | |-------|---------| | Check connections | `SELECT count(*) FROM pg_stat_activity;` | | Kill query | `SELECT pg_terminate_backend(pid);` | | Check replication lag | `SELECT extract(epoch from (now() - pg_last_xact_replay_timestamp()));` | | Check locks | `SELECT * FROM pg_locks WHERE NOT granted;` | ## Connection Pool Exhaustion ```sql -- Check current connections SELECT datname, usename, state, count(*) FROM pg_stat_activity GROUP BY datname, usename, state ORDER BY count(*) DESC; -- Identify long-running connections SELECT pid, usename, datname, state, query_start, query FROM pg_stat_activity WHERE state != 'idle' ORDER BY query_start; -- Terminate idle connections SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE state = 'idle' AND query_start < now() - interval '10 minutes'; ```` ## Replication Lag ```sql -- Check lag on replica SELECT CASE WHEN pg_last_wal_receive_lsn() = pg_last_wal_replay_lsn() THEN 0 ELSE extract(epoch from now() - pg_last_xact_replay_timestamp()) END AS lag_seconds; -- If lag > 60s, consider: -- 1. Check network between primary/replica -- 2. Check replica disk I/O -- 3. Consider failover if unrecoverable ``` ## Disk Space Critical ```bash # Check disk usage df -h /var/lib/postgresql/data # Find large tables psql -c "SELECT relname, pg_size_pretty(pg_total_relation_size(relid)) FROM pg_catalog.pg_statio_user_tables ORDER BY pg_total_relation_size(relid) DESC LIMIT 10;" # VACUUM to reclaim space psql -c "VACUUM FULL large_table;" # If emergency, delete old data or expand disk ``` ``` ## Best Practices ### Do's - **Keep runbooks updated** - Review after every incident - **Test runbooks regularly** - Game days, chaos engineering - **Include rollback steps** - Always have an escape hatch - **Document assumptions** - What must be true for steps to work - **Link to dashboards** - Quick access during stress ### Don'ts - **Don't assume knowledge** - Write for 3 AM brain - **Don't skip verification** - Confirm each step worked - **Don't forget communication** - Keep stakeholders informed - **Don't work alone** - Escalate early - **Don't skip postmortems** - Learn from every incident ## Resources - [Google SRE Book - Incident Management](https://sre.google/sre-book/managing-incidents/) - [PagerDuty Incident Response](https://response.pagerduty.com/) - [Atlassian Incident Management](https://www.atlassian.com/incident-management) ```
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

task-coordination-strategies

Decompose complex tasks, design dependency graphs, and coordinate

coding
⭐1
# Task Coordination Strategies Strategies for decomposing complex tasks into parallelizable units, designing dependency graphs, writing effective task descriptions, and monitoring workload across agent teams. ## When to Use This Skill - Breaking down a complex task for parallel execution - Designing task dependency relationships (blockedBy/blocks) - Writing task descriptions with clear acceptance criteria - Monitoring and rebalancing workload across teammates - Identifying the critical path in a multi-task workflow ## Task Decomposition Strategies ### By Layer Split work by architectural layer: - Frontend components - Backend API endpoints - Database migrations/models - Test suites **Best for**: Full-stack features, vertical slices ### By Component Split work by functional component: - Authentication module - User profile module - Notification module **Best for**: Microservices, modular architectures ### By Concern Split work by cross-cutting concern: - Security review - Performance review - Architecture review **Best for**: Code reviews, audits ### By File Ownership Split work by file/directory boundaries: - `src/components/` — Implementer 1 - `src/api/` — Implementer 2 - `src/utils/` — Implementer 3 **Best for**: Parallel implementation, conflict avoidance ## Dependency Graph Design ### Principles 1. **Minimize chain depth** — Prefer wide, shallow graphs over deep chains 2. **Identify the critical path** — The longest chain determines minimum completion time 3. **Use blockedBy sparingly** — Only add dependencies that are truly required 4. **Avoid circular dependencies** — Task A blocks B blocks A is a deadlock ### Patterns **Independent (Best parallelism)**: ``` Task A ─┐ Task B ─┼─→ Integration Task C ā”€ā”˜ ``` **Sequential (Necessary dependencies)**: ``` Task A → Task B → Task C ``` **Diamond (Mixed)**: ``` ā”Œā†’ Task B ─┐ Task A ─┤ ā”œā†’ Task D └→ Task C ā”€ā”˜ ``` ### Using blockedBy/blocks ``` TaskCreate: { subject: "Build API endpoints" } → Task #1 TaskCreate: { subject: "Build frontend components" } → Task #2 TaskCreate: { subject: "Integration testing" } → Task #3 TaskUpdate: { taskId: "3", addBlockedBy: ["1", "2"] } → #3 waits for #1 and #2 ``` ## Task Description Best Practices Every task should include: 1. **Objective** — What needs to be accomplished (1-2 sentences) 2. **Owned Files** — Explicit list of files/directories this teammate may modify 3. **Requirements** — Specific deliverables or behaviors expected 4. **Interface Contracts** — How this work connects to other teammates' work 5. **Acceptance Criteria** — How to verify the task is done correctly 6. **Scope Boundaries** — What is explicitly out of scope ### Template ``` ## Objective Build the user authentication API endpoints. ## Owned Files - src/api/auth.ts - src/api/middleware/auth-middleware.ts - src/types/auth.ts (shared — read only, do not modify) ## Requirements - POST /api/login — accepts email/password, returns JWT - POST /api/register — creates new user, returns JWT - GET /api/me — returns current user profile (requires auth) ## Interface Contract - Import User type from src/types/auth.ts (owned by implementer-1) - Export AuthResponse type for frontend consumption ## Acceptance Criteria - All endpoints return proper HTTP status codes - JWT tokens expire after 24 hours - Passwords are hashed with bcrypt ## Out of Scope - OAuth/social login - Password reset flow - Rate limiting ``` ## Workload Monitoring ### Indicators of Imbalance | Signal | Meaning | Action | | -------------------------- | ------------------- | --------------------------- | | Teammate idle, others busy | Uneven distribution | Reassign pending tasks | | Teammate stuck on one task | Possible blocker | Check in, offer help | | All tasks blocked | Dependency issue | Resolve critical path first | | One teammate has 3x others | Overloaded | Split tasks or reassign | ### Rebalancing Steps 1. Call `TaskList` to assess current state 2. Identify idle or overloaded teammates 3. Use `TaskUpdate` to reassign tasks 4. Use `SendMessage` to notify affected teammates 5. Monitor for improved throughput
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

billing-automation

Build automated billing systems for recurring payments, invoicing,

business
⭐1
# Billing Automation Master automated billing systems including recurring billing, invoice generation, dunning management, proration, and tax calculation. ## When to Use This Skill - Implementing SaaS subscription billing - Automating invoice generation and delivery - Managing failed payment recovery (dunning) - Calculating prorated charges for plan changes - Handling sales tax, VAT, and GST - Processing usage-based billing - Managing billing cycles and renewals ## Core Concepts ### 1. Billing Cycles **Common Intervals:** - Monthly (most common for SaaS) - Annual (discounted long-term) - Quarterly - Weekly - Custom (usage-based, per-seat) ### 2. Subscription States ``` trial → active → past_due → canceled → paused → resumed ``` ### 3. Dunning Management Automated process to recover failed payments through: - Retry schedules - Customer notifications - Grace periods - Account restrictions ### 4. Proration Adjusting charges when: - Upgrading/downgrading mid-cycle - Adding/removing seats - Changing billing frequency ## Quick Start ```python from billing import BillingEngine, Subscription # Initialize billing engine billing = BillingEngine() # Create subscription subscription = billing.create_subscription( customer_id="cus_123", plan_id="plan_pro_monthly", billing_cycle_anchor=datetime.now(), trial_days=14 ) # Process billing cycle billing.process_billing_cycle(subscription.id) ``` ## Subscription Lifecycle Management ```python from datetime import datetime, timedelta from enum import Enum class SubscriptionStatus(Enum): TRIAL = "trial" ACTIVE = "active" PAST_DUE = "past_due" CANCELED = "canceled" PAUSED = "paused" class Subscription: def __init__(self, customer_id, plan, billing_cycle_day=None): self.id = generate_id() self.customer_id = customer_id self.plan = plan self.status = SubscriptionStatus.TRIAL self.current_period_start = datetime.now() self.current_period_end = self.current_period_start + timedelta(days=plan.trial_days or 30) self.billing_cycle_day = billing_cycle_day or self.current_period_start.day self.trial_end = datetime.now() + timedelta(days=plan.trial_days) if plan.trial_days else None def start_trial(self, trial_days): """Start trial period.""" self.status = SubscriptionStatus.TRIAL self.trial_end = datetime.now() + timedelta(days=trial_days) self.current_period_end = self.trial_end def activate(self): """Activate subscription after trial or immediately.""" self.status = SubscriptionStatus.ACTIVE self.current_period_start = datetime.now() self.current_period_end = self.calculate_next_billing_date() def mark_past_due(self): """Mark subscription as past due after failed payment.""" self.status = SubscriptionStatus.PAST_DUE # Trigger dunning workflow def cancel(self, at_period_end=True): """Cancel subscription.""" if at_period_end: self.cancel_at_period_end = True # Will cancel when current period ends else: self.status = SubscriptionStatus.CANCELED self.canceled_at = datetime.now() def calculate_next_billing_date(self): """Calculate next billing date based on interval.""" if self.plan.interval == 'month': return self.current_period_start + timedelta(days=30) elif self.plan.interval == 'year': return self.current_period_start + timedelta(days=365) elif self.plan.interval == 'week': return self.current_period_start + timedelta(days=7) ``` ## Billing Cycle Processing ```python class BillingEngine: def process_billing_cycle(self, subscription_id): """Process billing for a subscription.""" subscription = self.get_subscription(subscription_id) # Check if billing is due if datetime.now() < subscription.current_period_end: return # Generate invoice invoice = self.generate_invoice(subscription) # Attempt payment payment_result = self.charge_customer( subscription.customer_id, invoice.total ) if payment_result.success: # Payment successful invoice.mark_paid() subscription.advance_billing_period() self.send_invoice(invoice) else: # Payment failed subscription.mark_past_due() self.start_dunning_process(subscription, invoice) def generate_invoice(self, subscription): """Generate invoice for billing period.""" invoice = Invoice( customer_id=subscription.customer_id, subscription_id=subscription.id, period_start=subscription.current_period_start, period_end=subscription.current_period_end ) # Add subscription line item invoice.add_line_item( description=subscription.plan.name, amount=subscription.plan.amount, quantity=subscription.quantity or 1 ) # Add usage-based charges if applicable if subscription.has_usage_billing: usage_charges = self.calculate_usage_charges(subscription) invoice.add_line_item( description="Usage charges", amount=usage_charges ) # Calculate tax tax = self.calculate_tax(invoice.subtotal, subscription.customer) invoice.tax = tax invoice.finalize() return invoice def charge_customer(self, customer_id, amount): """Charge customer using saved payment method.""" customer = self.get_customer(customer_id) try: # Charge using payment processor charge = stripe.Charge.create( customer=customer.stripe_id, amount=int(amount * 100), # Convert to cents currency='usd' ) return PaymentResult(success=True, transaction_id=charge.id) except stripe.error.CardError as e: return PaymentResult(success=False, error=str(e)) ``` ## Dunning Management ```python class DunningManager: """Manage failed payment recovery.""" def __init__(self): self.retry_schedule = [ {'days': 3, 'email_template': 'payment_failed_first'}, {'days': 7, 'email_template': 'payment_failed_reminder'}, {'days': 14, 'email_template': 'payment_failed_final'} ] def start_dunning_process(self, subscription, invoice): """Start dunning process for failed payment.""" dunning_attempt = DunningAttempt( subscription_id=subscription.id, invoice_id=invoice.id, attempt_number=1, next_retry=datetime.now() + timedelta(days=3) ) # Send initial failure notification self.send_dunning_email(subscription, 'payment_failed_first') # Schedule retries self.schedule_retries(dunning_attempt) def retry_payment(self, dunning_attempt): """Retry failed payment.""" subscription = self.get_subscription(dunning_attempt.subscription_id) invoice = self.get_invoice(dunning_attempt.invoice_id) # Attempt payment again result = self.charge_customer(subscription.customer_id, invoice.total) if result.success: # Payment succeeded invoice.mark_paid() subscription.status = SubscriptionStatus.ACTIVE self.send_dunning_email(subscription, 'payment_recovered') dunning_attempt.mark_resolved() else: # Still failing dunning_attempt.attempt_number += 1 if dunning_attempt.attempt_number < len(self.retry_schedule): # Schedule next retry next_retry_config = self.retry_schedule[dunning_attempt.attempt_number] dunning_attempt.next_retry = datetime.now() + timedelta(days=next_retry_config['days']) self.send_dunning_email(subscription, next_retry_config['email_template']) else: # Exhausted retries, cancel subscription subscription.cancel(at_period_end=False) self.send_dunning_email(subscription, 'subscription_canceled') def send_dunning_email(self, subscription, template): """Send dunning notification to customer.""" customer = self.get_customer(subscription.customer_id) email_content = self.render_template(template, { 'customer_name': customer.name, 'amount_due': subscription.plan.amount, 'update_payment_url': f"https://app.example.com/billing" }) send_email( to=customer.email, subject=email_content['subject'], body=email_content['body'] ) ``` ## Proration ```python class ProrationCalculator: """Calculate prorated charges for plan changes.""" @staticmethod def calculate_proration(old_plan, new_plan, period_start, period_end, change_date): """Calculate proration for plan change.""" # Days in current period total_days = (period_end - period_start).days # Days used on old plan days_used = (change_date - period_start).days # Days remaining on new plan days_remaining = (period_end - change_date).days # Calculate prorated amounts unused_amount = (old_plan.amount / total_days) * days_remaining new_plan_amount = (new_plan.amount / total_days) * days_remaining # Net charge/credit proration = new_plan_amount - unused_amount return { 'old_plan_credit': -unused_amount, 'new_plan_charge': new_plan_amount, 'net_proration': proration, 'days_used': days_used, 'days_remaining': days_remaining } @staticmethod def calculate_seat_proration(current_seats, new_seats, price_per_seat, period_start, period_end, change_date): """Calculate proration for seat changes.""" total_days = (period_end - period_start).days days_remaining = (period_end - change_date).days # Additional seats charge additional_seats = new_seats - current_seats prorated_amount = (additional_seats * price_per_seat / total_days) * days_remaining return { 'additional_seats': additional_seats, 'prorated_charge': max(0, prorated_amount), # No refund for removing seats mid-cycle 'effective_date': change_date } ``` ## Tax Calculation ```python class TaxCalculator: """Calculate sales tax, VAT, GST.""" def __init__(self): # Tax rates by region self.tax_rates = { 'US_CA': 0.0725, # California sales tax 'US_NY': 0.04, # New York sales tax 'GB': 0.20, # UK VAT 'DE': 0.19, # Germany VAT 'FR': 0.20, # France VAT 'AU': 0.10, # Australia GST } def calculate_tax(self, amount, customer): """Calculate applicable tax.""" # Determine tax jurisdiction jurisdiction = self.get_tax_jurisdiction(customer) if not jurisdiction: return 0 # Get tax rate tax_rate = self.tax_rates.get(jurisdiction, 0) # Calculate tax tax = amount * tax_rate return { 'tax_amount': tax, 'tax_rate': tax_rate, 'jurisdiction': jurisdiction, 'tax_type': self.get_tax_type(jurisdiction) } def get_tax_jurisdiction(self, customer): """Determine tax jurisdiction based on customer location.""" if customer.country == 'US': # US: Tax based on customer state return f"US_{customer.state}" elif customer.country in ['GB', 'DE', 'FR']: # EU: VAT return customer.country elif customer.country == 'AU': # Australia: GST return 'AU' else: return None def get_tax_type(self, jurisdiction): """Get type of tax for jurisdiction.""" if jurisdiction.startswith('US_'): return 'Sales Tax' elif jurisdiction in ['GB', 'DE', 'FR']: return 'VAT' elif jurisdiction == 'AU': return 'GST' return 'Tax' def validate_vat_number(self, vat_number, country): """Validate EU VAT number.""" # Use VIES API for validation # Returns True if valid, False otherwise pass ``` ## Invoice Generation ```python class Invoice: def __init__(self, customer_id, subscription_id=None): self.id = generate_invoice_number() self.customer_id = customer_id self.subscription_id = subscription_id self.status = 'draft' self.line_items = [] self.subtotal = 0 self.tax = 0 self.total = 0 self.created_at = datetime.now() def add_line_item(self, description, amount, quantity=1): """Add line item to invoice.""" line_item = { 'description': description, 'unit_amount': amount, 'quantity': quantity, 'total': amount * quantity } self.line_items.append(line_item) self.subtotal += line_item['total'] def finalize(self): """Finalize invoice and calculate total.""" self.total = self.subtotal + self.tax self.status = 'open' self.finalized_at = datetime.now() def mark_paid(self): """Mark invoice as paid.""" self.status = 'paid' self.paid_at = datetime.now() def to_pdf(self): """Generate PDF invoice.""" from reportlab.pdfgen import canvas # Generate PDF # Include: company info, customer info, line items, tax, total pass def to_html(self): """Generate HTML invoice.""" template = """ <!DOCTYPE html> <html> <head><title>Invoice #{invoice_number}</title></head> <body> <h1>Invoice #{invoice_number}</h1> <p>Date: {date}</p> <h2>Bill To:</h2> <p>{customer_name}<br>{customer_address}</p> <table> <tr><th>Description</th><th>Quantity</th><th>Amount</th></tr> {line_items} </table> <p>Subtotal: ${subtotal}</p> <p>Tax: ${tax}</p> <h3>Total: ${total}</h3> </body> </html> """ return template.format( invoice_number=self.id, date=self.created_at.strftime('%Y-%m-%d'), customer_name=self.customer.name, customer_address=self.customer.address, line_items=self.render_line_items(), subtotal=self.subtotal, tax=self.tax, total=self.total ) ``` ## Usage-Based Billing ```python class UsageBillingEngine: """Track and bill for usage.""" def track_usage(self, customer_id, metric, quantity): """Track usage event.""" UsageRecord.create( customer_id=customer_id, metric=metric, quantity=quantity, timestamp=datetime.now() ) def calculate_usage_charges(self, subscription, period_start, period_end): """Calculate charges for usage in billing period.""" usage_records = UsageRecord.get_for_period( subscription.customer_id, period_start, period_end ) total_usage = sum(record.quantity for record in usage_records) # Tiered pricing if subscription.plan.pricing_model == 'tiered': charge = self.calculate_tiered_pricing(total_usage, subscription.plan.tiers) # Per-unit pricing elif subscription.plan.pricing_model == 'per_unit': charge = total_usage * subscription.plan.unit_price # Volume pricing elif subscription.plan.pricing_model == 'volume': charge = self.calculate_volume_pricing(total_usage, subscription.plan.tiers) return charge def calculate_tiered_pricing(self, total_usage, tiers): """Calculate cost using tiered pricing.""" charge = 0 remaining = total_usage for tier in sorted(tiers, key=lambda x: x['up_to']): tier_usage = min(remaining, tier['up_to'] - tier['from']) charge += tier_usage * tier['unit_price'] remaining -= tier_usage if remaining <= 0: break return charge ``` ## Resources - **references/billing-cycles.md**: Billing cycle management - **references/dunning-management.md**: Failed payment recovery - **references/proration.md**: Prorated charge calculations - **references/tax-calculation.md**: Tax/VAT/GST handling - **references/invoice-lifecycle.md**: Invoice state management - **assets/billing-state-machine.yaml**: Billing workflow - **assets/invoice-template.html**: Invoice templates - **assets/dunning-policy.yaml**: Dunning configuration ## Best Practices 1. **Automate Everything**: Minimize manual intervention 2. **Clear Communication**: Notify customers of billing events 3. **Flexible Retry Logic**: Balance recovery with customer experience 4. **Accurate Proration**: Fair calculation for plan changes 5. **Tax Compliance**: Calculate correct tax for jurisdiction 6. **Audit Trail**: Log all billing events 7. **Graceful Degradation**: Handle edge cases without breaking ## Common Pitfalls - **Incorrect Proration**: Not accounting for partial periods - **Missing Tax**: Forgetting to add tax to invoices - **Aggressive Dunning**: Canceling too quickly - **No Notifications**: Not informing customers of failures - **Hardcoded Cycles**: Not supporting custom billing dates
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

pci-compliance

Implement PCI DSS compliance requirements for secure handling of

business
⭐1
# 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.
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

python-resource-management

Python resource management with context managers, cleanup patterns,

coding
⭐1
# Python Resource Management Manage resources deterministically using context managers. Resources like database connections, file handles, and network sockets should be released reliably, even when exceptions occur. ## When to Use This Skill - Managing database connections and connection pools - Working with file handles and I/O - Implementing custom context managers - Building streaming responses with state - Handling nested resource cleanup - Creating async context managers ## Core Concepts ### 1. Context Managers The `with` statement ensures resources are released automatically, even on exceptions. ### 2. Protocol Methods `__enter__`/`__exit__` for sync, `__aenter__`/`__aexit__` for async resource management. ### 3. Unconditional Cleanup `__exit__` always runs, regardless of whether an exception occurred. ### 4. Exception Handling Return `True` from `__exit__` to suppress exceptions, `False` to propagate them. ## Quick Start ```python from contextlib import contextmanager @contextmanager def managed_resource(): resource = acquire_resource() try: yield resource finally: resource.cleanup() with managed_resource() as r: r.do_work() ``` ## Fundamental Patterns ### Pattern 1: Class-Based Context Manager Implement the context manager protocol for complex resources. ```python class DatabaseConnection: """Database connection with automatic cleanup.""" def __init__(self, dsn: str) -> None: self._dsn = dsn self._conn: Connection | None = None def connect(self) -> None: """Establish database connection.""" self._conn = psycopg.connect(self._dsn) def close(self) -> None: """Close connection if open.""" if self._conn is not None: self._conn.close() self._conn = None def __enter__(self) -> "DatabaseConnection": """Enter context: connect and return self.""" self.connect() return self def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) -> None: """Exit context: always close connection.""" self.close() # Usage with context manager (preferred) with DatabaseConnection(dsn) as db: result = db.execute(query) # Manual management when needed db = DatabaseConnection(dsn) db.connect() try: result = db.execute(query) finally: db.close() ``` ### Pattern 2: Async Context Manager For async resources, implement the async protocol. ```python class AsyncDatabasePool: """Async database connection pool.""" def __init__(self, dsn: str, min_size: int = 1, max_size: int = 10) -> None: self._dsn = dsn self._min_size = min_size self._max_size = max_size self._pool: asyncpg.Pool | None = None async def __aenter__(self) -> "AsyncDatabasePool": """Create connection pool.""" self._pool = await asyncpg.create_pool( self._dsn, min_size=self._min_size, max_size=self._max_size, ) return self async def __aexit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) -> None: """Close all connections in pool.""" if self._pool is not None: await self._pool.close() async def execute(self, query: str, *args) -> list[dict]: """Execute query using pooled connection.""" async with self._pool.acquire() as conn: return await conn.fetch(query, *args) # Usage async with AsyncDatabasePool(dsn) as pool: users = await pool.execute("SELECT * FROM users WHERE active = $1", True) ``` ### Pattern 3: Using @contextmanager Decorator Simplify context managers with the decorator for straightforward cases. ```python from contextlib import contextmanager, asynccontextmanager import time import structlog logger = structlog.get_logger() @contextmanager def timed_block(name: str): """Time a block of code.""" start = time.perf_counter() try: yield finally: elapsed = time.perf_counter() - start logger.info(f"{name} completed", duration_seconds=round(elapsed, 3)) # Usage with timed_block("data_processing"): process_large_dataset() @asynccontextmanager async def database_transaction(conn: AsyncConnection): """Manage database transaction.""" await conn.execute("BEGIN") try: yield conn await conn.execute("COMMIT") except Exception: await conn.execute("ROLLBACK") raise # Usage async with database_transaction(conn) as tx: await tx.execute("INSERT INTO users ...") await tx.execute("INSERT INTO audit_log ...") ``` ### Pattern 4: Unconditional Resource Release Always clean up resources in `__exit__`, regardless of exceptions. ```python class FileProcessor: """Process file with guaranteed cleanup.""" def __init__(self, path: str) -> None: self._path = path self._file: IO | None = None self._temp_files: list[Path] = [] def __enter__(self) -> "FileProcessor": self._file = open(self._path, "r") return self def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) -> None: """Clean up all resources unconditionally.""" # Close main file if self._file is not None: self._file.close() # Clean up any temporary files for temp_file in self._temp_files: try: temp_file.unlink() except OSError: pass # Best effort cleanup # Return None/False to propagate any exception ``` ## Advanced Patterns ### Pattern 5: Selective Exception Suppression Only suppress specific, documented exceptions. ```python class StreamWriter: """Writer that handles broken pipe gracefully.""" def __init__(self, stream) -> None: self._stream = stream def __enter__(self) -> "StreamWriter": return self def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) -> bool: """Clean up, suppressing BrokenPipeError on shutdown.""" self._stream.close() # Suppress BrokenPipeError (client disconnected) # This is expected behavior, not an error if exc_type is BrokenPipeError: return True # Exception suppressed return False # Propagate all other exceptions ``` ### Pattern 6: Streaming with Accumulated State Maintain both incremental chunks and accumulated state during streaming. ```python from collections.abc import Generator from dataclasses import dataclass, field @dataclass class StreamingResult: """Accumulated streaming result.""" chunks: list[str] = field(default_factory=list) _finalized: bool = False @property def content(self) -> str: """Get accumulated content.""" return "".join(self.chunks) def add_chunk(self, chunk: str) -> None: """Add chunk to accumulator.""" if self._finalized: raise RuntimeError("Cannot add to finalized result") self.chunks.append(chunk) def finalize(self) -> str: """Mark stream complete and return content.""" self._finalized = True return self.content def stream_with_accumulation( response: StreamingResponse, ) -> Generator[tuple[str, str], None, str]: """Stream response while accumulating content. Yields: Tuple of (accumulated_content, new_chunk) for each chunk. Returns: Final accumulated content. """ result = StreamingResult() for chunk in response.iter_content(): result.add_chunk(chunk) yield result.content, chunk return result.finalize() ``` ### Pattern 7: Efficient String Accumulation Avoid O(n²) string concatenation when accumulating. ```python def accumulate_stream(stream) -> str: """Efficiently accumulate stream content.""" # BAD: O(n²) due to string immutability # content = "" # for chunk in stream: # content += chunk # Creates new string each time # GOOD: O(n) with list and join chunks: list[str] = [] for chunk in stream: chunks.append(chunk) return "".join(chunks) # Single allocation ``` ### Pattern 8: Tracking Stream Metrics Measure time-to-first-byte and total streaming time. ```python import time from collections.abc import Generator def stream_with_metrics( response: StreamingResponse, ) -> Generator[str, None, dict]: """Stream response while collecting metrics. Yields: Content chunks. Returns: Metrics dictionary. """ start = time.perf_counter() first_chunk_time: float | None = None chunk_count = 0 total_bytes = 0 for chunk in response.iter_content(): if first_chunk_time is None: first_chunk_time = time.perf_counter() - start chunk_count += 1 total_bytes += len(chunk.encode()) yield chunk total_time = time.perf_counter() - start return { "time_to_first_byte_ms": round((first_chunk_time or 0) * 1000, 2), "total_time_ms": round(total_time * 1000, 2), "chunk_count": chunk_count, "total_bytes": total_bytes, } ``` ### Pattern 9: Managing Multiple Resources with ExitStack Handle a dynamic number of resources cleanly. ```python from contextlib import ExitStack, AsyncExitStack from pathlib import Path def process_files(paths: list[Path]) -> list[str]: """Process multiple files with automatic cleanup.""" results = [] with ExitStack() as stack: # Open all files - they'll all be closed when block exits files = [stack.enter_context(open(p)) for p in paths] for f in files: results.append(f.read()) return results async def process_connections(hosts: list[str]) -> list[dict]: """Process multiple async connections.""" results = [] async with AsyncExitStack() as stack: connections = [ await stack.enter_async_context(connect_to_host(host)) for host in hosts ] for conn in connections: results.append(await conn.fetch_data()) return results ``` ## Best Practices Summary 1. **Always use context managers** - For any resource that needs cleanup 2. **Clean up unconditionally** - `__exit__` runs even on exception 3. **Don't suppress unexpectedly** - Return `False` unless suppression is intentional 4. **Use @contextmanager** - For simple resource patterns 5. **Implement both protocols** - Support `with` and manual management 6. **Use ExitStack** - For dynamic numbers of resources 7. **Accumulate efficiently** - List + join, not string concatenation 8. **Track metrics** - Time-to-first-byte matters for streaming 9. **Document behavior** - Especially exception suppression 10. **Test cleanup paths** - Verify resources are released on errors
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

security-requirement-extraction

Derive security requirements from threat models and business

security
⭐1
# 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
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

stride-analysis-patterns

Apply STRIDE methodology to systematically identify threats. Use

security
⭐1
# STRIDE Analysis Patterns Systematic threat identification using the STRIDE methodology. ## When to Use This Skill - Starting new threat modeling sessions - Analyzing existing system architecture - Reviewing security design decisions - Creating threat documentation - Training teams on threat identification - Compliance and audit preparation ## Core Concepts ### 1. STRIDE Categories ``` S - Spoofing → Authentication threats T - Tampering → Integrity threats R - Repudiation → Non-repudiation threats I - Information → Confidentiality threats Disclosure D - Denial of → Availability threats Service E - Elevation of → Authorization threats Privilege ``` ### 2. Threat Analysis Matrix | Category | Question | Control Family | | ------------------- | ----------------------------------------- | -------------- | | **Spoofing** | Can attacker pretend to be someone else? | Authentication | | **Tampering** | Can attacker modify data in transit/rest? | Integrity | | **Repudiation** | Can attacker deny actions? | Logging/Audit | | **Info Disclosure** | Can attacker access unauthorized data? | Encryption | | **DoS** | Can attacker disrupt availability? | Rate limiting | | **Elevation** | Can attacker gain higher privileges? | Authorization | ## Templates ### Template 1: STRIDE Threat Model Document ```markdown # Threat Model: [System Name] ## 1. System Overview ### 1.1 Description [Brief description of the system and its purpose] ### 1.2 Data Flow Diagram ``` [User] --> [Web App] --> [API Gateway] --> [Backend Services] | v [Database] ``` ### 1.3 Trust Boundaries - **External Boundary**: Internet to DMZ - **Internal Boundary**: DMZ to Internal Network - **Data Boundary**: Application to Database ## 2. Assets | Asset | Sensitivity | Description | |-------|-------------|-------------| | User Credentials | High | Authentication tokens, passwords | | Personal Data | High | PII, financial information | | Session Data | Medium | Active user sessions | | Application Logs | Medium | System activity records | | Configuration | High | System settings, secrets | ## 3. STRIDE Analysis ### 3.1 Spoofing Threats | ID | Threat | Target | Impact | Likelihood | |----|--------|--------|--------|------------| | S1 | Session hijacking | User sessions | High | Medium | | S2 | Token forgery | JWT tokens | High | Low | | S3 | Credential stuffing | Login endpoint | High | High | **Mitigations:** - [ ] Implement MFA - [ ] Use secure session management - [ ] Implement account lockout policies ### 3.2 Tampering Threats | ID | Threat | Target | Impact | Likelihood | |----|--------|--------|--------|------------| | T1 | SQL injection | Database queries | Critical | Medium | | T2 | Parameter manipulation | API requests | High | High | | T3 | File upload abuse | File storage | High | Medium | **Mitigations:** - [ ] Input validation on all endpoints - [ ] Parameterized queries - [ ] File type validation ### 3.3 Repudiation Threats | ID | Threat | Target | Impact | Likelihood | |----|--------|--------|--------|------------| | R1 | Transaction denial | Financial ops | High | Medium | | R2 | Access log tampering | Audit logs | Medium | Low | | R3 | Action attribution | User actions | Medium | Medium | **Mitigations:** - [ ] Comprehensive audit logging - [ ] Log integrity protection - [ ] Digital signatures for critical actions ### 3.4 Information Disclosure Threats | ID | Threat | Target | Impact | Likelihood | |----|--------|--------|--------|------------| | I1 | Data breach | User PII | Critical | Medium | | I2 | Error message leakage | System info | Low | High | | I3 | Insecure transmission | Network traffic | High | Medium | **Mitigations:** - [ ] Encryption at rest and in transit - [ ] Sanitize error messages - [ ] Implement TLS 1.3 ### 3.5 Denial of Service Threats | ID | Threat | Target | Impact | Likelihood | |----|--------|--------|--------|------------| | D1 | Resource exhaustion | API servers | High | High | | D2 | Database overload | Database | Critical | Medium | | D3 | Bandwidth saturation | Network | High | Medium | **Mitigations:** - [ ] Rate limiting - [ ] Auto-scaling - [ ] DDoS protection ### 3.6 Elevation of Privilege Threats | ID | Threat | Target | Impact | Likelihood | |----|--------|--------|--------|------------| | E1 | IDOR vulnerabilities | User resources | High | High | | E2 | Role manipulation | Admin access | Critical | Low | | E3 | JWT claim tampering | Authorization | High | Medium | **Mitigations:** - [ ] Proper authorization checks - [ ] Principle of least privilege - [ ] Server-side role validation ## 4. Risk Assessment ### 4.1 Risk Matrix ``` IMPACT Low Med High Crit Low 1 2 3 4 L Med 2 4 6 8 I High 3 6 9 12 K Crit 4 8 12 16 ``` ### 4.2 Prioritized Risks | Rank | Threat | Risk Score | Priority | |------|--------|------------|----------| | 1 | SQL Injection (T1) | 12 | Critical | | 2 | IDOR (E1) | 9 | High | | 3 | Credential Stuffing (S3) | 9 | High | | 4 | Data Breach (I1) | 8 | High | ## 5. Recommendations ### Immediate Actions 1. Implement input validation framework 2. Add rate limiting to authentication endpoints 3. Enable comprehensive audit logging ### Short-term (30 days) 1. Deploy WAF with OWASP ruleset 2. Implement MFA for sensitive operations 3. Encrypt all PII at rest ### Long-term (90 days) 1. Security awareness training 2. Penetration testing 3. Bug bounty program ``` ### Template 2: STRIDE Analysis Code ```python from dataclasses import dataclass, field from enum import Enum from typing import List, Dict, Optional import json class StrideCategory(Enum): SPOOFING = "S" TAMPERING = "T" REPUDIATION = "R" INFORMATION_DISCLOSURE = "I" DENIAL_OF_SERVICE = "D" ELEVATION_OF_PRIVILEGE = "E" class Impact(Enum): LOW = 1 MEDIUM = 2 HIGH = 3 CRITICAL = 4 class Likelihood(Enum): LOW = 1 MEDIUM = 2 HIGH = 3 CRITICAL = 4 @dataclass class Threat: id: str category: StrideCategory title: str description: str target: str impact: Impact likelihood: Likelihood mitigations: List[str] = field(default_factory=list) status: str = "open" @property def risk_score(self) -> int: return self.impact.value * self.likelihood.value @property def risk_level(self) -> str: score = self.risk_score if score >= 12: return "Critical" elif score >= 6: return "High" elif score >= 3: return "Medium" return "Low" @dataclass class Asset: name: str sensitivity: str description: str data_classification: str @dataclass class TrustBoundary: name: str description: str from_zone: str to_zone: str @dataclass class ThreatModel: name: str version: str description: str assets: List[Asset] = field(default_factory=list) boundaries: List[TrustBoundary] = field(default_factory=list) threats: List[Threat] = field(default_factory=list) def add_threat(self, threat: Threat) -> None: self.threats.append(threat) def get_threats_by_category(self, category: StrideCategory) -> List[Threat]: return [t for t in self.threats if t.category == category] def get_critical_threats(self) -> List[Threat]: return [t for t in self.threats if t.risk_level in ("Critical", "High")] def generate_report(self) -> Dict: """Generate threat model report.""" return { "summary": { "name": self.name, "version": self.version, "total_threats": len(self.threats), "critical_threats": len([t for t in self.threats if t.risk_level == "Critical"]), "high_threats": len([t for t in self.threats if t.risk_level == "High"]), }, "by_category": { cat.name: len(self.get_threats_by_category(cat)) for cat in StrideCategory }, "top_risks": [ { "id": t.id, "title": t.title, "risk_score": t.risk_score, "risk_level": t.risk_level } for t in sorted(self.threats, key=lambda x: x.risk_score, reverse=True)[:10] ] } class StrideAnalyzer: """Automated STRIDE analysis helper.""" STRIDE_QUESTIONS = { StrideCategory.SPOOFING: [ "Can an attacker impersonate a legitimate user?", "Are authentication tokens properly validated?", "Can session identifiers be predicted or stolen?", "Is multi-factor authentication available?", ], StrideCategory.TAMPERING: [ "Can data be modified in transit?", "Can data be modified at rest?", "Are input validation controls sufficient?", "Can an attacker manipulate application logic?", ], StrideCategory.REPUDIATION: [ "Are all security-relevant actions logged?", "Can logs be tampered with?", "Is there sufficient attribution for actions?", "Are timestamps reliable and synchronized?", ], StrideCategory.INFORMATION_DISCLOSURE: [ "Is sensitive data encrypted at rest?", "Is sensitive data encrypted in transit?", "Can error messages reveal sensitive information?", "Are access controls properly enforced?", ], StrideCategory.DENIAL_OF_SERVICE: [ "Are rate limits implemented?", "Can resources be exhausted by malicious input?", "Is there protection against amplification attacks?", "Are there single points of failure?", ], StrideCategory.ELEVATION_OF_PRIVILEGE: [ "Are authorization checks performed consistently?", "Can users access other users' resources?", "Can privilege escalation occur through parameter manipulation?", "Is the principle of least privilege followed?", ], } def generate_questionnaire(self, component: str) -> List[Dict]: """Generate STRIDE questionnaire for a component.""" questionnaire = [] for category, questions in self.STRIDE_QUESTIONS.items(): for q in questions: questionnaire.append({ "component": component, "category": category.name, "question": q, "answer": None, "notes": "" }) return questionnaire def suggest_mitigations(self, category: StrideCategory) -> List[str]: """Suggest common mitigations for a STRIDE category.""" mitigations = { StrideCategory.SPOOFING: [ "Implement multi-factor authentication", "Use secure session management", "Implement account lockout policies", "Use cryptographically secure tokens", "Validate authentication at every request", ], StrideCategory.TAMPERING: [ "Implement input validation", "Use parameterized queries", "Apply integrity checks (HMAC, signatures)", "Implement Content Security Policy", "Use immutable infrastructure", ], StrideCategory.REPUDIATION: [ "Enable comprehensive audit logging", "Protect log integrity", "Implement digital signatures", "Use centralized, tamper-evident logging", "Maintain accurate timestamps", ], StrideCategory.INFORMATION_DISCLOSURE: [ "Encrypt data at rest and in transit", "Implement proper access controls", "Sanitize error messages", "Use secure defaults", "Implement data classification", ], StrideCategory.DENIAL_OF_SERVICE: [ "Implement rate limiting", "Use auto-scaling", "Deploy DDoS protection", "Implement circuit breakers", "Set resource quotas", ], StrideCategory.ELEVATION_OF_PRIVILEGE: [ "Implement proper authorization", "Follow principle of least privilege", "Validate permissions server-side", "Use role-based access control", "Implement security boundaries", ], } return mitigations.get(category, []) ``` ### Template 3: Data Flow Diagram Analysis ```python from dataclasses import dataclass from typing import List, Set, Tuple from enum import Enum class ElementType(Enum): EXTERNAL_ENTITY = "external" PROCESS = "process" DATA_STORE = "datastore" DATA_FLOW = "dataflow" @dataclass class DFDElement: id: str name: str type: ElementType trust_level: int # 0 = untrusted, higher = more trusted description: str = "" @dataclass class DataFlow: id: str name: str source: str destination: str data_type: str protocol: str encrypted: bool = False class DFDAnalyzer: """Analyze Data Flow Diagrams for STRIDE threats.""" def __init__(self): self.elements: Dict[str, DFDElement] = {} self.flows: List[DataFlow] = [] def add_element(self, element: DFDElement) -> None: self.elements[element.id] = element def add_flow(self, flow: DataFlow) -> None: self.flows.append(flow) def find_trust_boundary_crossings(self) -> List[Tuple[DataFlow, int]]: """Find data flows that cross trust boundaries.""" crossings = [] for flow in self.flows: source = self.elements.get(flow.source) dest = self.elements.get(flow.destination) if source and dest and source.trust_level != dest.trust_level: trust_diff = abs(source.trust_level - dest.trust_level) crossings.append((flow, trust_diff)) return sorted(crossings, key=lambda x: x[1], reverse=True) def identify_threats_per_element(self) -> Dict[str, List[StrideCategory]]: """Map applicable STRIDE categories to element types.""" threat_mapping = { ElementType.EXTERNAL_ENTITY: [ StrideCategory.SPOOFING, StrideCategory.REPUDIATION, ], ElementType.PROCESS: [ StrideCategory.SPOOFING, StrideCategory.TAMPERING, StrideCategory.REPUDIATION, StrideCategory.INFORMATION_DISCLOSURE, StrideCategory.DENIAL_OF_SERVICE, StrideCategory.ELEVATION_OF_PRIVILEGE, ], ElementType.DATA_STORE: [ StrideCategory.TAMPERING, StrideCategory.REPUDIATION, StrideCategory.INFORMATION_DISCLOSURE, StrideCategory.DENIAL_OF_SERVICE, ], ElementType.DATA_FLOW: [ StrideCategory.TAMPERING, StrideCategory.INFORMATION_DISCLOSURE, StrideCategory.DENIAL_OF_SERVICE, ], } result = {} for elem_id, elem in self.elements.items(): result[elem_id] = threat_mapping.get(elem.type, []) return result def analyze_unencrypted_flows(self) -> List[DataFlow]: """Find unencrypted data flows crossing trust boundaries.""" risky_flows = [] for flow in self.flows: if not flow.encrypted: source = self.elements.get(flow.source) dest = self.elements.get(flow.destination) if source and dest and source.trust_level != dest.trust_level: risky_flows.append(flow) return risky_flows def generate_threat_enumeration(self) -> List[Dict]: """Generate comprehensive threat enumeration.""" threats = [] element_threats = self.identify_threats_per_element() for elem_id, categories in element_threats.items(): elem = self.elements[elem_id] for category in categories: threats.append({ "element_id": elem_id, "element_name": elem.name, "element_type": elem.type.value, "stride_category": category.name, "description": f"{category.name} threat against {elem.name}", "trust_level": elem.trust_level }) return threats ``` ### Template 4: STRIDE per Interaction ```python from typing import List, Dict, Optional from dataclasses import dataclass @dataclass class Interaction: """Represents an interaction between two components.""" id: str source: str target: str action: str data: str protocol: str class StridePerInteraction: """Apply STRIDE to each interaction in the system.""" INTERACTION_THREATS = { # Source type -> Target type -> Applicable threats ("external", "process"): { "S": "External entity spoofing identity to process", "T": "Tampering with data sent to process", "R": "External entity denying sending data", "I": "Data exposure during transmission", "D": "Flooding process with requests", "E": "Exploiting process to gain privileges", }, ("process", "datastore"): { "T": "Process tampering with stored data", "R": "Process denying data modifications", "I": "Unauthorized data access by process", "D": "Process exhausting storage resources", }, ("process", "process"): { "S": "Process spoofing another process", "T": "Tampering with inter-process data", "I": "Data leakage between processes", "D": "One process overwhelming another", "E": "Process gaining elevated access", }, } def analyze_interaction( self, interaction: Interaction, source_type: str, target_type: str ) -> List[Dict]: """Analyze a single interaction for STRIDE threats.""" threats = [] key = (source_type, target_type) applicable_threats = self.INTERACTION_THREATS.get(key, {}) for stride_code, description in applicable_threats.items(): threats.append({ "interaction_id": interaction.id, "source": interaction.source, "target": interaction.target, "stride_category": stride_code, "threat_description": description, "context": f"{interaction.action} - {interaction.data}", }) return threats def generate_threat_matrix( self, interactions: List[Interaction], element_types: Dict[str, str] ) -> List[Dict]:
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered