Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

28 NODES
πŸ€–system promptβ€’6 months ago

Semantic Session Checkpointing Pattern

Record lightweight semantic checkpoints at planning, risk, failure, and handoff boundaries while keeping heartbeat outside the model loop.

productivity
⭐1
# Semantic Session Checkpointing Pattern Imported from curated first-party documentation sources. ## What this covers Use this pattern when session continuity matters and you need checkpoints that survive model drift or hard crashes. ## Use this when - Capturing handoff context during long sessions - Checkpointing before risky refactors - Separating liveness tracking from agent behavior ## Expected outcomes - Checkpoint timing becomes explicit and repeatable - Session heartbeat stays reliable even if the model stalls - Handoffs capture task, next action, files touched, and risk ## Source synthesis - REVOKORE/docs/MCP-Integration.md ## Dedupe notes Uses the compact REVOKORE integration guide as a distinct session-state pattern that is not already covered by EVOKORE-MCP or AGENT33 docs. ## Source excerpts ### REVOKORE/docs/MCP-Integration.md ## Recommended MCP Pattern Expose a tool that writes a short checkpoint into the active session: - current task - next action - files touched - blocker or risk The simplest implementation is to call: ```powershell REVOKORE/scripts/Write-AiCliCheckpoint.ps1 -Kind note -Message "working on auth bug" ``` From an MCP server, you can call the same script directly or append JSON lines to `REVOKORE_CHECKPOINT_PATH`.
πŸ‘0
πŸ‘οΈ14
docs
πŸ“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

PR Merge Boundary Validation Runbook

Validate chained pull requests at each merge boundary with explicit checks, stale-approval handling, and rollback awareness.

devops
⭐1
# PR Merge Boundary Validation Runbook Imported from curated first-party documentation sources. ## What this covers Use this runbook when a merge sequence spans dependent pull requests and each boundary needs fresh validation. ## Use this when - Managing dependency-chained pull requests - Refreshing stale approvals near merge time - Reducing merge surprises through checkpoint validation ## Expected outcomes - Each merge boundary has a concrete verification step - Review freshness stays visible as branches evolve - Rollback planning is captured before the merge happens ## Source synthesis - EVOKORE-MCP/docs/PR_MERGE_RUNBOOK.md (https://github.com/mattmre/EVOKORE-MCP/blob/main/docs/PR_MERGE_RUNBOOK.md) ## Dedupe notes Keeps the merge-boundary validation concept separate from more general workflow or orchestration docs. ## Source excerpts ### EVOKORE-MCP/docs/PR_MERGE_RUNBOOK.md Operator runbook for reliable merges and context-rot prevention. ## Pre-merge Checklist - [ ] PR scope matches approved plan - [ ] PR description is filled using `.github/PULL_REQUEST_TEMPLATE.md` - [ ] PR metadata automation check (`scripts/validate-pr-metadata.js`) is passing for pull_request CI runs - [ ] Required tests pass locally/CI - [ ] Docs updated for user-facing behavior changes - [ ] Release-impacting changes called out - [ ] Follow-up issues captured (if any) ## Required Checks by Change Type Use this as the minimum check set before approval and merge: | Change type | Required checks | | --- | --- | | Docs-only changes | `node test-docs-canonical-links.js` | | Ops/docs process changes (`docs/PR_MERGE_RUNBOOK.md`, `next-session.md`, orchestration docs) | `node test-ops-docs-validation.js` and `node test-docs-canonical-links.js` | | Source/tooling/config changes (`src/`, `scripts/`, workflow/config files) | Relevant targeted tests for touched area plus CI-required suite | | Release-flow changes | `node test-npm-release-flow-validation.js` plus docs/link checks | If a PR spans multiple change types, run the union of required checks. ## Reviewer Responsibilities - Confirm PR scope and dependency assumptions are explicit in description. - Confirm PR metadata fields from `.github/PULL_REQUEST_TEMPLATE.md` are complete. - Verify required checks for each change type are attached in PR evidence. - Block approval if dependency base PR is not merged or branch is stale. - Approve only the current chain head; do not pre-approve non-head dependent PRs. - Confirm all review conversations are resolved before final approval. - Ensure merge strategy and rollback notes are documented for risky changes. ## Merge Steps 1. Rebase or update branch with latest target branch. 2. Re-run required validations. 3. Confirm reviewer approvals and resolved conversations. 4. Merge via approved strategy. 5. Record merge commit/PR number in tracker. ## Merge-boundary Checkpoints At every dependency merge boundary (`base -> dependent`): 1. Rebase dependent PR branch on latest `main` immediately after parent merge. 2. Re-run required checks and attach updated evidence in PR metadata. 3. Revalidate approvals for the new head state (stale approvals must be refreshed). 4. Confirm merge-boundary revalidation notes are updated before merge. ## Merge-order Controls (Dependency Chain) 1. Define merge order explicitly in PR descriptions (`base -> dependent`). 2. Merge only the current chain head; hold dependents until parent merge is complete. 3. After each merge, rebase ...
πŸ‘0
πŸ‘οΈ0
docs
πŸ€–system promptβ€’6 months ago

Hook Observability and Session Replay

Instrument hooks with replayable logs, task-state visibility, and non-blocking observability around agent sessions.

devops
⭐1
# Hook Observability and Session Replay Imported from curated first-party documentation sources. ## What this covers Use this skill when you need to understand what hooks fired, what they emitted, and how a session can be replayed after the fact. ## Use this when - Debugging hook-driven automation - Replaying session events after failures - Adding observability without blocking interactive work ## Expected outcomes - Hook activity becomes inspectable and replayable - Operational state survives beyond a single terminal session - Observability stays useful without overwhelming operators ## Source synthesis - EVOKORE-MCP/docs/VOICE_AND_HOOKS.md (https://github.com/mattmre/EVOKORE-MCP/blob/main/docs/VOICE_AND_HOOKS.md) - EVOKORE-MCP/docs/USE_CASES_AND_WALKTHROUGHS.md (https://github.com/mattmre/EVOKORE-MCP/blob/main/docs/USE_CASES_AND_WALKTHROUGHS.md) ## Dedupe notes Focuses on replay and observability instead of importing the broader voice-sidecar guide verbatim. ## Source excerpts ### EVOKORE-MCP/docs/VOICE_AND_HOOKS.md EVOKORE currently has three separate voice-related systems plus a set of hook and observability utilities. They overlap in operator workflows, but they are not the same runtime. ## The three voice-related systems ### 1. ElevenLabs MCP proxy This is the optional `elevenlabs` child server configured in `mcp.config.json`. What it is: - proxied through the EVOKORE router - exposed as prefixed MCP tools - available to any EVOKORE-connected MCP client when configured successfully What it is for: - text-to-speech and other ElevenLabs MCP operations as tools - routing voice-related actions through the standard EVOKORE proxy/security stack Requirements: - `uvx` available on PATH - `ELEVENLABS_API_KEY` set ### 2. VoiceMode VoiceMode is a separate voice-conversation system for Claude Code. What it is: - registered separately from EVOKORE - not routed through EVOKOREÒ€ℒs stdio server - used for bidirectional voice conversation in Claude Code What it is for: - speaking to Claude and hearing spoken responses - using `OPENAI_API_KEY` and VoiceModeÒ€ℒs own runtime Windows note: - VoiceMode relies on `uvx` being directly available - set `OPENAI_API_KEY` in the shell that launches Claude Code ### 3. VoiceSidecar VoiceSidecar is a standalone WebSocket server implemented in `src/Voice ... ### EVOKORE-MCP/docs/USE_CASES_AND_WALKTHROUGHS.md This guide turns the runtime contracts into practical operator flows. ## Walkthrough 1: Adopt a workflow from the skill library Use this when you want EVOKORE to retrieve process guidance before the model starts acting. ### Goal Find and adopt an existing workflow such as `session-wrap`. ### Steps 1. Ask the client to search skills: ```text Search the MCP for a workflow about session wrap-up and continuity. ``` 2. EVOKORE uses `search_skills` and returns matching skills. 3. Ask for a specific skill: ```text Show me help for the session-wrap skill. ``` 4. EVOKORE uses `get_skill_help` and returns the skillÒ€ℒs internal instructions. 5. For broader task matching, ask: ```text Resolve a workflow for wrapping this session, documenting open risks, and preparing the next handoff. ``` 6. EVOKORE uses `resolve_workflow` and injects the top 1-3 relevant workflows directly into the tool response. ### Why this matters - keeps the model grounded in repo-specific process - reduces prompt drift - makes handoff and governance behavior repeatable ## Walkthrough 2: Use a proxied tool that requires HITL approval Use this when the tool is configured as `require_approval` in `permissions.yml`. ### Goal Allow a protected proxied tool call such as `fs_write_f ...
πŸ‘0
πŸ‘οΈ0
docs
πŸ€–system promptβ€’6 months ago

Human-in-the-Loop Approval Token Workflow

Gate risky tool execution behind approval tokens so agents can retry safely after a human reviewer signs off.

security
⭐1
# Human-in-the-Loop Approval Token Workflow Imported from curated first-party documentation sources. ## What this covers Use this workflow when autonomous execution needs a durable handoff from human review back into the agent loop. ## Use this when - Retrying a blocked tool call after approval - Adding a human checkpoint to sensitive actions - Reducing insecure workarounds around approval flows ## Expected outcomes - Approval becomes a reusable tokenized workflow - Agents resume work without losing execution context - Security controls stay explicit instead of being implied ## Source synthesis - EVOKORE-MCP/docs/V2_MULTI_AGENT_WORKFLOWS.md (https://github.com/mattmre/EVOKORE-MCP/blob/main/docs/V2_MULTI_AGENT_WORKFLOWS.md) - EVOKORE-MCP/docs/AGENT33_IMPROVEMENT_INSTRUCTIONS.md (https://github.com/mattmre/EVOKORE-MCP/blob/main/docs/AGENT33_IMPROVEMENT_INSTRUCTIONS.md) ## Dedupe notes Synthesizes the approval-token concept from the main workflow doc and the Agent33 improvement transfer notes. ## Source excerpts ### EVOKORE-MCP/docs/V2_MULTI_AGENT_WORKFLOWS.md With EVOKORE-MCP v2.0 fully operational, we can now leverage 40+ proxied GitHub and Filesystem tools seamlessly within complex multi-agent workflows. The core architectural advancements include: ## 1. Dynamic Tool Prefixing & Indexing Instead of loading 40+ tools into an LLM's context window statically (which causes massive bloat), EVOKORE's `ProxyManager` boots child servers (like `@modelcontextprotocol/server-github` and `@modelcontextprotocol/server-filesystem`) and dynamically prefixes their tools (`github_create_issue`, `fs_write_file`). This prevents namespace collisions while keeping the tools accessible to native skills. ## 2. Human-in-the-Loop (HITL) Security Interceptor Automated multi-agent workflows involving sensitive endpoints (like GitHub write access or file deletion) are governed by EVOKORE's stateless `_evokore_approval_token` architecture. - When an agent attempts a restricted action, the tool call is intercepted and blocked. - The server returns an error explicitly commanding the agent to prompt the human for approval. - Upon approval, the agent retries the exact tool call with the injected token, securely fulfilling the workflow without severing the conversational context. ## 3. Active Skill Orchestration (Native Harnessing) Unlike v1.0 where skills merel ... ### EVOKORE-MCP/docs/AGENT33_IMPROVEMENT_INSTRUCTIONS.md > **Purpose**: Feed this file into Claude Code CLI when working on the Agent33 repo. It contains patterns, architectures, and capabilities proven in EVOKORE-MCP that Agent33 should adopt. --- ## 1. Multi-Server MCP Aggregation Pattern **What Agent33 lacks**: Agent33's MCP server (Phase 43) is a single-endpoint bridge. It doesn't aggregate multiple child MCP servers behind a unified namespace. **What to build**: A proxy layer that spawns and manages multiple child MCP servers from a single config file, presenting them as one unified tool surface. ### Implementation spec: ``` mcp.config.json { "servers": { "github": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-github"], "env": { "GITHUB_TOKEN": "${GITHUB_TOKEN}" } }, "fs": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "./"] }, "elevenlabs": { "command": "uvx", "args": ["elevenlabs-mcp"], "env": { "ELEVENLABS_API_KEY": "${ELEVENLABS_API_KEY}" } } } } ``` **Key patterns from EVOKORE**: - **Tool name prefixing**: Every proxied tool gets renamed `{serverId}_{originalName}` to prevent namespace collisions (e.g., `github_create_issue`, `fs_read_file`). First-registration-wins for duplicates. - **Environment interpolation**: `${VAR}` syntax in `env` blocks resolved ...
πŸ‘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

Autonomy Budget Lifecycle

Define, approve, enforce, and escalate autonomous execution budgets before agents act on files, commands, or network requests.

productivity
⭐1
# Autonomy Budget Lifecycle Imported from curated first-party documentation sources. ## What this covers Use this lifecycle when you need bounded autonomous execution with approvals, preflight checks, and runtime enforcement. ## Use this when - Running agents with scoped permissions - Escalating risky actions instead of silently proceeding - Adding governance to autonomous sessions ## Expected outcomes - Autonomy state moves through draft, approval, and active phases - Preflight checks happen before execution begins - Escalations are tracked instead of disappearing into logs ## Source synthesis - AGENT33/docs/functionality-and-workflows.md (https://github.com/mattmre/AGENT33/blob/main/docs/functionality-and-workflows.md) - AGENT33/docs/use-cases.md (https://github.com/mattmre/AGENT33/blob/main/docs/use-cases.md) ## Dedupe notes Collapses autonomy lifecycle and use-case material into a single execution-governance entry. ## Source excerpts ### AGENT33/docs/functionality-and-workflows.md ### 4.4 Autonomy Budget Lifecycle States: - `draft -> pending_approval -> active -> suspended|expired|completed` Flow: 1. Create budget 2. Activate or transition 3. Run preflight checks 4. Create enforcer 5. Evaluate command/file/network requests 6. Track escalations ### AGENT33/docs/use-cases.md ## 3. Autonomous Execution Budgeting Goal: - Enforce hard runtime limits for file, command, and network activity. Use these modules: - `api/routes/autonomy.py` - `autonomy/service.py` - `autonomy/enforcement.py` - `autonomy/preflight.py` Typical flow: 1. Create and activate budget. 2. Run preflight checks. 3. Attach runtime enforcer. 4. Gate each file/command/network action through enforcement APIs. 5. Trigger and resolve escalations. Best fit: - High-control automation in regulated or sensitive environments.
πŸ‘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

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

dotnet-backend-patterns

Master C#/.NET backend development patterns for building robust

coding
⭐1
# .NET Backend Development Patterns Master C#/.NET patterns for building production-grade APIs, MCP servers, and enterprise backends with modern best practices (2024/2025). ## When to Use This Skill - Developing new .NET Web APIs or MCP servers - Reviewing C# code for quality and performance - Designing service architectures with dependency injection - Implementing caching strategies with Redis - Writing unit and integration tests - Optimizing database access with EF Core or Dapper - Configuring applications with IOptions pattern - Handling errors and implementing resilience patterns ## Core Concepts ### 1. Project Structure (Clean Architecture) ``` src/ β”œβ”€β”€ Domain/ # Core business logic (no dependencies) β”‚ β”œβ”€β”€ Entities/ β”‚ β”œβ”€β”€ Interfaces/ β”‚ β”œβ”€β”€ Exceptions/ β”‚ └── ValueObjects/ β”œβ”€β”€ Application/ # Use cases, DTOs, validation β”‚ β”œβ”€β”€ Services/ β”‚ β”œβ”€β”€ DTOs/ β”‚ β”œβ”€β”€ Validators/ β”‚ └── Interfaces/ β”œβ”€β”€ Infrastructure/ # External implementations β”‚ β”œβ”€β”€ Data/ # EF Core, Dapper repositories β”‚ β”œβ”€β”€ Caching/ # Redis, Memory cache β”‚ β”œβ”€β”€ External/ # HTTP clients, third-party APIs β”‚ └── DependencyInjection/ # Service registration └── Api/ # Entry point β”œβ”€β”€ Controllers/ # Or MinimalAPI endpoints β”œβ”€β”€ Middleware/ β”œβ”€β”€ Filters/ └── Program.cs ``` ### 2. Dependency Injection Patterns ```csharp // Service registration by lifetime public static class ServiceCollectionExtensions { public static IServiceCollection AddApplicationServices( this IServiceCollection services, IConfiguration configuration) { // Scoped: One instance per HTTP request services.AddScoped<IProductService, ProductService>(); services.AddScoped<IOrderService, OrderService>(); // Singleton: One instance for app lifetime services.AddSingleton<ICacheService, RedisCacheService>(); services.AddSingleton<IConnectionMultiplexer>(_ => ConnectionMultiplexer.Connect(configuration["Redis:Connection"]!)); // Transient: New instance every time services.AddTransient<IValidator<CreateOrderRequest>, CreateOrderValidator>(); // Options pattern for configuration services.Configure<CatalogOptions>(configuration.GetSection("Catalog")); services.Configure<RedisOptions>(configuration.GetSection("Redis")); // Factory pattern for conditional creation services.AddScoped<IPriceCalculator>(sp => { var options = sp.GetRequiredService<IOptions<PricingOptions>>().Value; return options.UseNewEngine ? sp.GetRequiredService<NewPriceCalculator>() : sp.GetRequiredService<LegacyPriceCalculator>(); }); // Keyed services (.NET 8+) services.AddKeyedScoped<IPaymentProcessor, StripeProcessor>("stripe"); services.AddKeyedScoped<IPaymentProcessor, PayPalProcessor>("paypal"); return services; } } // Usage with keyed services public class CheckoutService { public CheckoutService( [FromKeyedServices("stripe")] IPaymentProcessor stripeProcessor) { _processor = stripeProcessor; } } ``` ### 3. Async/Await Patterns ```csharp // βœ… CORRECT: Async all the way down public async Task<Product> GetProductAsync(string id, CancellationToken ct = default) { return await _repository.GetByIdAsync(id, ct); } // βœ… CORRECT: Parallel execution with WhenAll public async Task<(Stock, Price)> GetStockAndPriceAsync( string productId, CancellationToken ct = default) { var stockTask = _stockService.GetAsync(productId, ct); var priceTask = _priceService.GetAsync(productId, ct); await Task.WhenAll(stockTask, priceTask); return (await stockTask, await priceTask); } // βœ… CORRECT: ConfigureAwait in libraries public async Task<T> LibraryMethodAsync<T>(CancellationToken ct = default) { var result = await _httpClient.GetAsync(url, ct).ConfigureAwait(false); return await result.Content.ReadFromJsonAsync<T>(ct).ConfigureAwait(false); } // βœ… CORRECT: ValueTask for hot paths with caching public ValueTask<Product?> GetCachedProductAsync(string id) { if (_cache.TryGetValue(id, out Product? product)) return ValueTask.FromResult(product); return new ValueTask<Product?>(GetFromDatabaseAsync(id)); } // ❌ WRONG: Blocking on async (deadlock risk) var result = GetProductAsync(id).Result; // NEVER do this var result2 = GetProductAsync(id).GetAwaiter().GetResult(); // Also bad // ❌ WRONG: async void (except event handlers) public async void ProcessOrder() { } // Exceptions are lost // ❌ WRONG: Unnecessary Task.Run for already async code await Task.Run(async () => await GetDataAsync()); // Wastes thread ``` ### 4. Configuration with IOptions ```csharp // Configuration classes public class CatalogOptions { public const string SectionName = "Catalog"; public int DefaultPageSize { get; set; } = 50; public int MaxPageSize { get; set; } = 200; public TimeSpan CacheDuration { get; set; } = TimeSpan.FromMinutes(15); public bool EnableEnrichment { get; set; } = true; } public class RedisOptions { public const string SectionName = "Redis"; public string Connection { get; set; } = "localhost:6379"; public string KeyPrefix { get; set; } = "mcp:"; public int Database { get; set; } = 0; } // appsettings.json { "Catalog": { "DefaultPageSize": 50, "MaxPageSize": 200, "CacheDuration": "00:15:00", "EnableEnrichment": true }, "Redis": { "Connection": "localhost:6379", "KeyPrefix": "mcp:", "Database": 0 } } // Registration services.Configure<CatalogOptions>(configuration.GetSection(CatalogOptions.SectionName)); services.Configure<RedisOptions>(configuration.GetSection(RedisOptions.SectionName)); // Usage with IOptions (singleton, read once at startup) public class CatalogService { private readonly CatalogOptions _options; public CatalogService(IOptions<CatalogOptions> options) { _options = options.Value; } } // Usage with IOptionsSnapshot (scoped, re-reads on each request) public class DynamicService { private readonly CatalogOptions _options; public DynamicService(IOptionsSnapshot<CatalogOptions> options) { _options = options.Value; // Fresh value per request } } // Usage with IOptionsMonitor (singleton, notified on changes) public class MonitoredService { private CatalogOptions _options; public MonitoredService(IOptionsMonitor<CatalogOptions> monitor) { _options = monitor.CurrentValue; monitor.OnChange(newOptions => _options = newOptions); } } ``` ### 5. Result Pattern (Avoiding Exceptions for Flow Control) ```csharp // Generic Result type public class Result<T> { public bool IsSuccess { get; } public T? Value { get; } public string? Error { get; } public string? ErrorCode { get; } private Result(bool isSuccess, T? value, string? error, string? errorCode) { IsSuccess = isSuccess; Value = value; Error = error; ErrorCode = errorCode; } public static Result<T> Success(T value) => new(true, value, null, null); public static Result<T> Failure(string error, string? code = null) => new(false, default, error, code); public Result<TNew> Map<TNew>(Func<T, TNew> mapper) => IsSuccess ? Result<TNew>.Success(mapper(Value!)) : Result<TNew>.Failure(Error!, ErrorCode); public async Task<Result<TNew>> MapAsync<TNew>(Func<T, Task<TNew>> mapper) => IsSuccess ? Result<TNew>.Success(await mapper(Value!)) : Result<TNew>.Failure(Error!, ErrorCode); } // Usage in service public async Task<Result<Order>> CreateOrderAsync(CreateOrderRequest request, CancellationToken ct) { // Validation var validation = await _validator.ValidateAsync(request, ct); if (!validation.IsValid) return Result<Order>.Failure( validation.Errors.First().ErrorMessage, "VALIDATION_ERROR"); // Business rule check var stock = await _stockService.CheckAsync(request.ProductId, request.Quantity, ct); if (!stock.IsAvailable) return Result<Order>.Failure( $"Insufficient stock: {stock.Available} available, {request.Quantity} requested", "INSUFFICIENT_STOCK"); // Create order var order = await _repository.CreateAsync(request.ToEntity(), ct); return Result<Order>.Success(order); } // Usage in controller/endpoint app.MapPost("/orders", async ( CreateOrderRequest request, IOrderService orderService, CancellationToken ct) => { var result = await orderService.CreateOrderAsync(request, ct); return result.IsSuccess ? Results.Created($"/orders/{result.Value!.Id}", result.Value) : Results.BadRequest(new { error = result.Error, code = result.ErrorCode }); }); ``` ## Data Access Patterns ### Entity Framework Core ```csharp // DbContext configuration public class AppDbContext : DbContext { public DbSet<Product> Products => Set<Product>(); public DbSet<Order> Orders => Set<Order>(); protected override void OnModelCreating(ModelBuilder modelBuilder) { // Apply all configurations from assembly modelBuilder.ApplyConfigurationsFromAssembly(typeof(AppDbContext).Assembly); // Global query filters modelBuilder.Entity<Product>().HasQueryFilter(p => !p.IsDeleted); } } // Entity configuration public class ProductConfiguration : IEntityTypeConfiguration<Product> { public void Configure(EntityTypeBuilder<Product> builder) { builder.ToTable("Products"); builder.HasKey(p => p.Id); builder.Property(p => p.Id).HasMaxLength(40); builder.Property(p => p.Name).HasMaxLength(200).IsRequired(); builder.Property(p => p.Price).HasPrecision(18, 2); builder.HasIndex(p => p.Sku).IsUnique(); builder.HasIndex(p => new { p.CategoryId, p.Name }); builder.HasMany(p => p.OrderItems) .WithOne(oi => oi.Product) .HasForeignKey(oi => oi.ProductId); } } // Repository with EF Core public class ProductRepository : IProductRepository { private readonly AppDbContext _context; public async Task<Product?> GetByIdAsync(string id, CancellationToken ct = default) { return await _context.Products .AsNoTracking() .FirstOrDefaultAsync(p => p.Id == id, ct); } public async Task<IReadOnlyList<Product>> SearchAsync( ProductSearchCriteria criteria, CancellationToken ct = default) { var query = _context.Products.AsNoTracking(); if (!string.IsNullOrWhiteSpace(criteria.SearchTerm)) query = query.Where(p => EF.Functions.Like(p.Name, $"%{criteria.SearchTerm}%")); if (criteria.CategoryId.HasValue) query = query.Where(p => p.CategoryId == criteria.CategoryId); if (criteria.MinPrice.HasValue) query = query.Where(p => p.Price >= criteria.MinPrice); if (criteria.MaxPrice.HasValue) query = query.Where(p => p.Price <= criteria.MaxPrice); return await query .OrderBy(p => p.Name) .Skip((criteria.Page - 1) * criteria.PageSize) .Take(criteria.PageSize) .ToListAsync(ct); } } ``` ### Dapper for Performance ```csharp public class DapperProductRepository : IProductRepository { private readonly IDbConnection _connection; public async Task<Product?> GetByIdAsync(string id, CancellationToken ct = default) { const string sql = """ SELECT Id, Name, Sku, Price, CategoryId, Stock, CreatedAt FROM Products WHERE Id = @Id AND IsDeleted = 0 """; return await _connection.QueryFirstOrDefaultAsync<Product>( new CommandDefinition(sql, new { Id = id }, cancellationToken: ct)); } public async Task<IReadOnlyList<Product>> SearchAsync( ProductSearchCriteria criteria, CancellationToken ct = default) { var sql = new StringBuilder(""" SELECT Id, Name, Sku, Price, CategoryId, Stock, CreatedAt FROM Products WHERE IsDeleted = 0 """); var parameters = new DynamicParameters(); if (!string.IsNullOrWhiteSpace(criteria.SearchTerm)) { sql.Append(" AND Name LIKE @SearchTerm"); parameters.Add("SearchTerm", $"%{criteria.SearchTerm}%"); } if (criteria.CategoryId.HasValue) { sql.Append(" AND CategoryId = @CategoryId"); parameters.Add("CategoryId", criteria.CategoryId); } if (criteria.MinPrice.HasValue) { sql.Append(" AND Price >= @MinPrice"); parameters.Add("MinPrice", criteria.MinPrice); } if (criteria.MaxPrice.HasValue) { sql.Append(" AND Price <= @MaxPrice"); parameters.Add("MaxPrice", criteria.MaxPrice); } sql.Append(" ORDER BY Name OFFSET @Offset ROWS FETCH NEXT @PageSize ROWS ONLY"); parameters.Add("Offset", (criteria.Page - 1) * criteria.PageSize); parameters.Add("PageSize", criteria.PageSize); var results = await _connection.QueryAsync<Product>( new CommandDefinition(sql.ToString(), parameters, cancellationToken: ct)); return results.ToList(); } // Multi-mapping for related data public async Task<Order?> GetOrderWithItemsAsync(int orderId, CancellationToken ct = default) { const string sql = """ SELECT o.*, oi.*, p.* FROM Orders o LEFT JOIN OrderItems oi ON o.Id = oi.OrderId LEFT JOIN Products p ON oi.ProductId = p.Id WHERE o.Id = @OrderId """; var orderDictionary = new Dictionary<int, Order>(); await _connection.QueryAsync<Order, OrderItem, Product, Order>( new CommandDefinition(sql, new { OrderId = orderId }, cancellationToken: ct), (order, item, product) => { if (!orderDictionary.TryGetValue(order.Id, out var existingOrder)) { existingOrder = order; existingOrder.Items = new List<OrderItem>(); orderDictionary.Add(order.Id, existingOrder); } if (item != null) { item.Product = product; existingOrder.Items.Add(item); } return existingOrder; }, splitOn: "Id,Id"); return orderDictionary.Values.FirstOrDefault(); } } ``` ## Caching Patterns ### Multi-Level Cache with Redis ```csharp public class CachedProductService : IProductService { private readonly IProductRepository _repository; private readonly IMemoryCache _memoryCache; private readonly IDistributedCache _distributedCache; private readonly ILogger<CachedProductService> _logger; private static readonly TimeSpan MemoryCacheDuration = TimeSpan.FromMinutes(1); private static readonly TimeSpan DistributedCacheDuration = TimeSpan.FromMinutes(15); public async Task<Product?> GetByIdAsync(string id, CancellationToken ct = default) { var cacheKey = $"product:{id}"; // L1: Memory cache (in-process, fastest) if (_memoryCache.TryGetValue(cacheKey, out Product? cached)) { _logger.LogDebug("L1 cache hit for {CacheKey}", cacheKey); return cached; } // L2: Distributed cache (Redis) var distributed = await _distributedCache.GetStringAsync(cacheKey, ct); if (distributed != null) { _logger.LogDebug("L2 cache hit for {CacheKey}", cacheKey); var product = JsonSerializer.Deserialize<Product>(distributed); // Populate L1 _memoryCache.Set(cacheKey, product, MemoryCacheDuration); return product; } // L3: Database _logger.LogDebug("Cache miss for {CacheKey}, fetching from database", cacheKey); var fromDb = await _repository.GetByIdAsync(id, ct); if (fromDb != null) { var serialized = JsonSerializer.Serialize(fromDb); // Populate both caches await _distributedCache.SetStringAsync( cacheKey, serialized, new DistributedCacheEntryOptions { AbsoluteExpirationRelativeToNow = DistributedCacheDuration }, ct); _memoryCache.Set(cacheKey, fromDb, MemoryCacheDuration); } return fromDb; } public async Task InvalidateAsync(string id, CancellationToken ct = default) { var cacheKey = $"product:{id}"; _memoryCache.Remove(cacheKey); await _distributedCache.RemoveAsync(cacheKey, ct); _logger.LogInformation("Invalidated cache for {CacheKey}", cacheKey); } } // Stale-while-revalidate pattern public class StaleWhileRevalidateCache<T> { private readonly IDistributedCache _cache; private readonly TimeSpan _freshDuration; private readonly TimeSpan _staleDuration; public async Task<T?> GetOrCreateAsync( string key, Func<CancellationToken, Task<T>> factory, CancellationToken ct = default) { var cached = await _cache.GetStringAsync(key, ct); if (cached != null) { var entry = JsonSerializer.Deserialize<CacheEntry<T>>(cached)!; if (entry.IsStale && !entry.IsExpired) { // Return stale data immediately, refresh in background _ = Task.Run(async () => { var fresh = await factory(CancellationToken.None); await SetAsync(key, fresh, CancellationToken.None); }); } if (!entry.IsExpired) return entry.Value; } // Cache miss or expired var value = await factory(ct); await SetAsync(key, value, ct); return value; } private record CacheEntry<TValue>(TValue Value, DateTime CreatedAt) { public bool IsStale => DateTime.UtcNow - CreatedAt > _freshDuration; public bool IsExpired => DateTime.UtcNow - CreatedAt > _staleDuration; } } ``` ## Testing Patterns ### Unit Tests with xUnit and Moq ```csharp public class OrderServiceTests { private readonly Mock<IOrderRepository> _mockRepository; private readonly Mock<IStockService> _mockStockService; private readonly Mock<IValidator<CreateOrderRequest>> _mockValidator; private readonly OrderService _sut; // System Under Test public OrderServiceTests() { _mockRepository = new Mock<IOrderRepository>(); _mockStockService = new Mock<IStockService>(); _mockValidator = new Mock<IValidator<CreateOrderRequest>>(); // Default: validation passes _mockValidator .Setup(v => v.ValidateAsync(It.IsAny<CreateOrderRequest>(), It.IsAny<
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

slo-implementation

Define and implement Service Level Indicators (SLIs) and Service

coding
⭐1
# SLO Implementation Framework for defining and implementing Service Level Indicators (SLIs), Service Level Objectives (SLOs), and error budgets. ## Purpose Implement measurable reliability targets using SLIs, SLOs, and error budgets to balance reliability with innovation velocity. ## When to Use - Define service reliability targets - Measure user-perceived reliability - Implement error budgets - Create SLO-based alerts - Track reliability goals ## SLI/SLO/SLA Hierarchy ``` SLA (Service Level Agreement) ↓ Contract with customers SLO (Service Level Objective) ↓ Internal reliability target SLI (Service Level Indicator) ↓ Actual measurement ``` ## Defining SLIs ### Common SLI Types #### 1. Availability SLI ```promql # Successful requests / Total requests sum(rate(http_requests_total{status!~"5.."}[28d])) / sum(rate(http_requests_total[28d])) ``` #### 2. Latency SLI ```promql # Requests below latency threshold / Total requests sum(rate(http_request_duration_seconds_bucket{le="0.5"}[28d])) / sum(rate(http_request_duration_seconds_count[28d])) ``` #### 3. Durability SLI ``` # Successful writes / Total writes sum(storage_writes_successful_total) / sum(storage_writes_total) ``` **Reference:** See `references/slo-definitions.md` ## Setting SLO Targets ### Availability SLO Examples | SLO % | Downtime/Month | Downtime/Year | | ------ | -------------- | ------------- | | 99% | 7.2 hours | 3.65 days | | 99.9% | 43.2 minutes | 8.76 hours | | 99.95% | 21.6 minutes | 4.38 hours | | 99.99% | 4.32 minutes | 52.56 minutes | ### Choose Appropriate SLOs **Consider:** - User expectations - Business requirements - Current performance - Cost of reliability - Competitor benchmarks **Example SLOs:** ```yaml slos: - name: api_availability target: 99.9 window: 28d sli: | sum(rate(http_requests_total{status!~"5.."}[28d])) / sum(rate(http_requests_total[28d])) - name: api_latency_p95 target: 99 window: 28d sli: | sum(rate(http_request_duration_seconds_bucket{le="0.5"}[28d])) / sum(rate(http_request_duration_seconds_count[28d])) ``` ## Error Budget Calculation ### Error Budget Formula ``` Error Budget = 1 - SLO Target ``` **Example:** - SLO: 99.9% availability - Error Budget: 0.1% = 43.2 minutes/month - Current Error: 0.05% = 21.6 minutes/month - Remaining Budget: 50% ### Error Budget Policy ```yaml error_budget_policy: - remaining_budget: 100% action: Normal development velocity - remaining_budget: 50% action: Consider postponing risky changes - remaining_budget: 10% action: Freeze non-critical changes - remaining_budget: 0% action: Feature freeze, focus on reliability ``` **Reference:** See `references/error-budget.md` ## SLO Implementation ### Prometheus Recording Rules ```yaml # SLI Recording Rules groups: - name: sli_rules interval: 30s rules: # Availability SLI - record: sli:http_availability:ratio expr: | sum(rate(http_requests_total{status!~"5.."}[28d])) / sum(rate(http_requests_total[28d])) # Latency SLI (requests < 500ms) - record: sli:http_latency:ratio expr: | sum(rate(http_request_duration_seconds_bucket{le="0.5"}[28d])) / sum(rate(http_request_duration_seconds_count[28d])) - name: slo_rules interval: 5m rules: # SLO compliance (1 = meeting SLO, 0 = violating) - record: slo:http_availability:compliance expr: sli:http_availability:ratio >= bool 0.999 - record: slo:http_latency:compliance expr: sli:http_latency:ratio >= bool 0.99 # Error budget remaining (percentage) - record: slo:http_availability:error_budget_remaining expr: | (sli:http_availability:ratio - 0.999) / (1 - 0.999) * 100 # Error budget burn rate - record: slo:http_availability:burn_rate_5m expr: | (1 - ( sum(rate(http_requests_total{status!~"5.."}[5m])) / sum(rate(http_requests_total[5m])) )) / (1 - 0.999) ``` ### SLO Alerting Rules ```yaml groups: - name: slo_alerts interval: 1m rules: # Fast burn: 14.4x rate, 1 hour window # Consumes 2% error budget in 1 hour - alert: SLOErrorBudgetBurnFast expr: | slo:http_availability:burn_rate_1h > 14.4 and slo:http_availability:burn_rate_5m > 14.4 for: 2m labels: severity: critical annotations: summary: "Fast error budget burn detected" description: "Error budget burning at {{ $value }}x rate" # Slow burn: 6x rate, 6 hour window # Consumes 5% error budget in 6 hours - alert: SLOErrorBudgetBurnSlow expr: | slo:http_availability:burn_rate_6h > 6 and slo:http_availability:burn_rate_30m > 6 for: 15m labels: severity: warning annotations: summary: "Slow error budget burn detected" description: "Error budget burning at {{ $value }}x rate" # Error budget exhausted - alert: SLOErrorBudgetExhausted expr: slo:http_availability:error_budget_remaining < 0 for: 5m labels: severity: critical annotations: summary: "SLO error budget exhausted" description: "Error budget remaining: {{ $value }}%" ``` ## SLO Dashboard **Grafana Dashboard Structure:** ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ SLO Compliance (Current) β”‚ β”‚ βœ“ 99.95% (Target: 99.9%) β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ Error Budget Remaining: 65% β”‚ β”‚ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–‘β–‘ 65% β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ SLI Trend (28 days) β”‚ β”‚ [Time series graph] β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ Burn Rate Analysis β”‚ β”‚ [Burn rate by time window] β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` **Example Queries:** ```promql # Current SLO compliance sli:http_availability:ratio * 100 # Error budget remaining slo:http_availability:error_budget_remaining # Days until error budget exhausted (at current burn rate) (slo:http_availability:error_budget_remaining / 100) * 28 / (1 - sli:http_availability:ratio) * (1 - 0.999) ``` ## Multi-Window Burn Rate Alerts ```yaml # Combination of short and long windows reduces false positives rules: - alert: SLOBurnRateHigh expr: | ( slo:http_availability:burn_rate_1h > 14.4 and slo:http_availability:burn_rate_5m > 14.4 ) or ( slo:http_availability:burn_rate_6h > 6 and slo:http_availability:burn_rate_30m > 6 ) labels: severity: critical ``` ## SLO Review Process ### Weekly Review - Current SLO compliance - Error budget status - Trend analysis - Incident impact ### Monthly Review - SLO achievement - Error budget usage - Incident postmortems - SLO adjustments ### Quarterly Review - SLO relevance - Target adjustments - Process improvements - Tooling enhancements ## Best Practices 1. **Start with user-facing services** 2. **Use multiple SLIs** (availability, latency, etc.) 3. **Set achievable SLOs** (don't aim for 100%) 4. **Implement multi-window alerts** to reduce noise 5. **Track error budget** consistently 6. **Review SLOs regularly** 7. **Document SLO decisions** 8. **Align with business goals** 9. **Automate SLO reporting** 10. **Use SLOs for prioritization** ## Reference Files - `assets/slo-template.md` - SLO definition template - `references/slo-definitions.md` - SLO definition patterns - `references/error-budget.md` - Error budget calculations ## Related Skills - `prometheus-configuration` - For metric collection - `grafana-dashboards` - For SLO visualization
πŸ‘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

async-python-patterns

Master Python asyncio, concurrent programming, and async/await

coding
⭐1
# Async Python Patterns Comprehensive guidance for implementing asynchronous Python applications using asyncio, concurrent programming patterns, and async/await for building high-performance, non-blocking systems. ## When to Use This Skill - Building async web APIs (FastAPI, aiohttp, Sanic) - Implementing concurrent I/O operations (database, file, network) - Creating web scrapers with concurrent requests - Developing real-time applications (WebSocket servers, chat systems) - Processing multiple independent tasks simultaneously - Building microservices with async communication - Optimizing I/O-bound workloads - Implementing async background tasks and queues ## Sync vs Async Decision Guide Before adopting async, consider whether it's the right choice for your use case. | Use Case | Recommended Approach | |----------|---------------------| | Many concurrent network/DB calls | `asyncio` | | CPU-bound computation | `multiprocessing` or thread pool | | Mixed I/O + CPU | Offload CPU work with `asyncio.to_thread()` | | Simple scripts, few connections | Sync (simpler, easier to debug) | | Web APIs with high concurrency | Async frameworks (FastAPI, aiohttp) | **Key Rule:** Stay fully sync or fully async within a call path. Mixing creates hidden blocking and complexity. ## Core Concepts ### 1. Event Loop The event loop is the heart of asyncio, managing and scheduling asynchronous tasks. **Key characteristics:** - Single-threaded cooperative multitasking - Schedules coroutines for execution - Handles I/O operations without blocking - Manages callbacks and futures ### 2. Coroutines Functions defined with `async def` that can be paused and resumed. **Syntax:** ```python async def my_coroutine(): result = await some_async_operation() return result ``` ### 3. Tasks Scheduled coroutines that run concurrently on the event loop. ### 4. Futures Low-level objects representing eventual results of async operations. ### 5. Async Context Managers Resources that support `async with` for proper cleanup. ### 6. Async Iterators Objects that support `async for` for iterating over async data sources. ## Quick Start ```python import asyncio async def main(): print("Hello") await asyncio.sleep(1) print("World") # Python 3.7+ asyncio.run(main()) ``` ## Fundamental Patterns ### Pattern 1: Basic Async/Await ```python import asyncio async def fetch_data(url: str) -> dict: """Fetch data from URL asynchronously.""" await asyncio.sleep(1) # Simulate I/O return {"url": url, "data": "result"} async def main(): result = await fetch_data("https://api.example.com") print(result) asyncio.run(main()) ``` ### Pattern 2: Concurrent Execution with gather() ```python import asyncio from typing import List async def fetch_user(user_id: int) -> dict: """Fetch user data.""" await asyncio.sleep(0.5) return {"id": user_id, "name": f"User {user_id}"} async def fetch_all_users(user_ids: List[int]) -> List[dict]: """Fetch multiple users concurrently.""" tasks = [fetch_user(uid) for uid in user_ids] results = await asyncio.gather(*tasks) return results async def main(): user_ids = [1, 2, 3, 4, 5] users = await fetch_all_users(user_ids) print(f"Fetched {len(users)} users") asyncio.run(main()) ``` ### Pattern 3: Task Creation and Management ```python import asyncio async def background_task(name: str, delay: int): """Long-running background task.""" print(f"{name} started") await asyncio.sleep(delay) print(f"{name} completed") return f"Result from {name}" async def main(): # Create tasks task1 = asyncio.create_task(background_task("Task 1", 2)) task2 = asyncio.create_task(background_task("Task 2", 1)) # Do other work print("Main: doing other work") await asyncio.sleep(0.5) # Wait for tasks result1 = await task1 result2 = await task2 print(f"Results: {result1}, {result2}") asyncio.run(main()) ``` ### Pattern 4: Error Handling in Async Code ```python import asyncio from typing import List, Optional async def risky_operation(item_id: int) -> dict: """Operation that might fail.""" await asyncio.sleep(0.1) if item_id % 3 == 0: raise ValueError(f"Item {item_id} failed") return {"id": item_id, "status": "success"} async def safe_operation(item_id: int) -> Optional[dict]: """Wrapper with error handling.""" try: return await risky_operation(item_id) except ValueError as e: print(f"Error: {e}") return None async def process_items(item_ids: List[int]): """Process multiple items with error handling.""" tasks = [safe_operation(iid) for iid in item_ids] results = await asyncio.gather(*tasks, return_exceptions=True) # Filter out failures successful = [r for r in results if r is not None and not isinstance(r, Exception)] failed = [r for r in results if isinstance(r, Exception)] print(f"Success: {len(successful)}, Failed: {len(failed)}") return successful asyncio.run(process_items([1, 2, 3, 4, 5, 6])) ``` ### Pattern 5: Timeout Handling ```python import asyncio async def slow_operation(delay: int) -> str: """Operation that takes time.""" await asyncio.sleep(delay) return f"Completed after {delay}s" async def with_timeout(): """Execute operation with timeout.""" try: result = await asyncio.wait_for(slow_operation(5), timeout=2.0) print(result) except asyncio.TimeoutError: print("Operation timed out") asyncio.run(with_timeout()) ``` ## Advanced Patterns ### Pattern 6: Async Context Managers ```python import asyncio from typing import Optional class AsyncDatabaseConnection: """Async database connection context manager.""" def __init__(self, dsn: str): self.dsn = dsn self.connection: Optional[object] = None async def __aenter__(self): print("Opening connection") await asyncio.sleep(0.1) # Simulate connection self.connection = {"dsn": self.dsn, "connected": True} return self.connection async def __aexit__(self, exc_type, exc_val, exc_tb): print("Closing connection") await asyncio.sleep(0.1) # Simulate cleanup self.connection = None async def query_database(): """Use async context manager.""" async with AsyncDatabaseConnection("postgresql://localhost") as conn: print(f"Using connection: {conn}") await asyncio.sleep(0.2) # Simulate query return {"rows": 10} asyncio.run(query_database()) ``` ### Pattern 7: Async Iterators and Generators ```python import asyncio from typing import AsyncIterator async def async_range(start: int, end: int, delay: float = 0.1) -> AsyncIterator[int]: """Async generator that yields numbers with delay.""" for i in range(start, end): await asyncio.sleep(delay) yield i async def fetch_pages(url: str, max_pages: int) -> AsyncIterator[dict]: """Fetch paginated data asynchronously.""" for page in range(1, max_pages + 1): await asyncio.sleep(0.2) # Simulate API call yield { "page": page, "url": f"{url}?page={page}", "data": [f"item_{page}_{i}" for i in range(5)] } async def consume_async_iterator(): """Consume async iterator.""" async for number in async_range(1, 5): print(f"Number: {number}") print("\nFetching pages:") async for page_data in fetch_pages("https://api.example.com/items", 3): print(f"Page {page_data['page']}: {len(page_data['data'])} items") asyncio.run(consume_async_iterator()) ``` ### Pattern 8: Producer-Consumer Pattern ```python import asyncio from asyncio import Queue from typing import Optional async def producer(queue: Queue, producer_id: int, num_items: int): """Produce items and put them in queue.""" for i in range(num_items): item = f"Item-{producer_id}-{i}" await queue.put(item) print(f"Producer {producer_id} produced: {item}") await asyncio.sleep(0.1) await queue.put(None) # Signal completion async def consumer(queue: Queue, consumer_id: int): """Consume items from queue.""" while True: item = await queue.get() if item is None: queue.task_done() break print(f"Consumer {consumer_id} processing: {item}") await asyncio.sleep(0.2) # Simulate work queue.task_done() async def producer_consumer_example(): """Run producer-consumer pattern.""" queue = Queue(maxsize=10) # Create tasks producers = [ asyncio.create_task(producer(queue, i, 5)) for i in range(2) ] consumers = [ asyncio.create_task(consumer(queue, i)) for i in range(3) ] # Wait for producers await asyncio.gather(*producers) # Wait for queue to be empty await queue.join() # Cancel consumers for c in consumers: c.cancel() asyncio.run(producer_consumer_example()) ``` ### Pattern 9: Semaphore for Rate Limiting ```python import asyncio from typing import List async def api_call(url: str, semaphore: asyncio.Semaphore) -> dict: """Make API call with rate limiting.""" async with semaphore: print(f"Calling {url}") await asyncio.sleep(0.5) # Simulate API call return {"url": url, "status": 200} async def rate_limited_requests(urls: List[str], max_concurrent: int = 5): """Make multiple requests with rate limiting.""" semaphore = asyncio.Semaphore(max_concurrent) tasks = [api_call(url, semaphore) for url in urls] results = await asyncio.gather(*tasks) return results async def main(): urls = [f"https://api.example.com/item/{i}" for i in range(20)] results = await rate_limited_requests(urls, max_concurrent=3) print(f"Completed {len(results)} requests") asyncio.run(main()) ``` ### Pattern 10: Async Locks and Synchronization ```python import asyncio class AsyncCounter: """Thread-safe async counter.""" def __init__(self): self.value = 0 self.lock = asyncio.Lock() async def increment(self): """Safely increment counter.""" async with self.lock: current = self.value await asyncio.sleep(0.01) # Simulate work self.value = current + 1 async def get_value(self) -> int: """Get current value.""" async with self.lock: return self.value async def worker(counter: AsyncCounter, worker_id: int): """Worker that increments counter.""" for _ in range(10): await counter.increment() print(f"Worker {worker_id} incremented") async def test_counter(): """Test concurrent counter.""" counter = AsyncCounter() workers = [asyncio.create_task(worker(counter, i)) for i in range(5)] await asyncio.gather(*workers) final_value = await counter.get_value() print(f"Final counter value: {final_value}") asyncio.run(test_counter()) ``` ## Real-World Applications ### Web Scraping with aiohttp ```python import asyncio import aiohttp from typing import List, Dict async def fetch_url(session: aiohttp.ClientSession, url: str) -> Dict: """Fetch single URL.""" try: async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response: text = await response.text() return { "url": url, "status": response.status, "length": len(text) } except Exception as e: return {"url": url, "error": str(e)} async def scrape_urls(urls: List[str]) -> List[Dict]: """Scrape multiple URLs concurrently.""" async with aiohttp.ClientSession() as session: tasks = [fetch_url(session, url) for url in urls] results = await asyncio.gather(*tasks) return results async def main(): urls = [ "https://httpbin.org/delay/1", "https://httpbin.org/delay/2", "https://httpbin.org/status/404", ] results = await scrape_urls(urls) for result in results: print(result) asyncio.run(main()) ``` ### Async Database Operations ```python import asyncio from typing import List, Optional # Simulated async database client class AsyncDB: """Simulated async database.""" async def execute(self, query: str) -> List[dict]: """Execute query.""" await asyncio.sleep(0.1) return [{"id": 1, "name": "Example"}] async def fetch_one(self, query: str) -> Optional[dict]: """Fetch single row.""" await asyncio.sleep(0.1) return {"id": 1, "name": "Example"} async def get_user_data(db: AsyncDB, user_id: int) -> dict: """Fetch user and related data concurrently.""" user_task = db.fetch_one(f"SELECT * FROM users WHERE id = {user_id}") orders_task = db.execute(f"SELECT * FROM orders WHERE user_id = {user_id}") profile_task = db.fetch_one(f"SELECT * FROM profiles WHERE user_id = {user_id}") user, orders, profile = await asyncio.gather(user_task, orders_task, profile_task) return { "user": user, "orders": orders, "profile": profile } async def main(): db = AsyncDB() user_data = await get_user_data(db, 1) print(user_data) asyncio.run(main()) ``` ### WebSocket Server ```python import asyncio from typing import Set # Simulated WebSocket connection class WebSocket: """Simulated WebSocket.""" def __init__(self, client_id: str): self.client_id = client_id async def send(self, message: str): """Send message.""" print(f"Sending to {self.client_id}: {message}") await asyncio.sleep(0.01) async def recv(self) -> str: """Receive message.""" await asyncio.sleep(1) return f"Message from {self.client_id}" class WebSocketServer: """Simple WebSocket server.""" def __init__(self): self.clients: Set[WebSocket] = set() async def register(self, websocket: WebSocket): """Register new client.""" self.clients.add(websocket) print(f"Client {websocket.client_id} connected") async def unregister(self, websocket: WebSocket): """Unregister client.""" self.clients.remove(websocket) print(f"Client {websocket.client_id} disconnected") async def broadcast(self, message: str): """Broadcast message to all clients.""" if self.clients: tasks = [client.send(message) for client in self.clients] await asyncio.gather(*tasks) async def handle_client(self, websocket: WebSocket): """Handle individual client connection.""" await self.register(websocket) try: async for message in self.message_iterator(websocket): await self.broadcast(f"{websocket.client_id}: {message}") finally: await self.unregister(websocket) async def message_iterator(self, websocket: WebSocket): """Iterate over messages from client.""" for _ in range(3): # Simulate 3 messages yield await websocket.recv() ``` ## Performance Best Practices ### 1. Use Connection Pools ```python import asyncio import aiohttp async def with_connection_pool(): """Use connection pool for efficiency.""" connector = aiohttp.TCPConnector(limit=100, limit_per_host=10) async with aiohttp.ClientSession(connector=connector) as session: tasks = [session.get(f"https://api.example.com/item/{i}") for i in range(50)] responses = await asyncio.gather(*tasks) return responses ``` ### 2. Batch Operations ```python async def batch_process(items: List[str], batch_size: int = 10): """Process items in batches.""" for i in range(0, len(items), batch_size): batch = items[i:i + batch_size] tasks = [process_item(item) for item in batch] await asyncio.gather(*tasks) print(f"Processed batch {i // batch_size + 1}") async def process_item(item: str): """Process single item.""" await asyncio.sleep(0.1) return f"Processed: {item}" ``` ### 3. Avoid Blocking Operations Never block the event loop with synchronous operations. A single blocking call stalls all concurrent tasks. ```python # BAD - blocks the entire event loop async def fetch_data_bad(): import time import requests time.sleep(1) # Blocks! response = requests.get(url) # Also blocks! # GOOD - use async-native libraries (e.g., httpx for async HTTP) import httpx async def fetch_data_good(url: str): await asyncio.sleep(1) async with httpx.AsyncClient() as client: response = await client.get(url) ``` **Wrapping Blocking Code with `asyncio.to_thread()` (Python 3.9+):** When you must use synchronous libraries, offload to a thread pool: ```python import asyncio from pathlib import Path async def read_file_async(path: str) -> str: """Read file without blocking event loop.""" # asyncio.to_thread() runs sync code in a thread pool return await asyncio.to_thread(Path(path).read_text) async def call_sync_library(data: dict) -> dict: """Wrap a synchronous library call.""" # Useful for sync database drivers, file I/O, CPU work return await asyncio.to_thread(sync_library.process, data) ``` **Lower-level approach with `run_in_executor()`:** ```python import asyncio import concurrent.futures from typing import Any def blocking_operation(data: Any) -> Any: """CPU-intensive blocking operation.""" import time time.sleep(1) return data * 2 async def run_in_executor(data: Any) -> Any: """Run blocking operation in thread pool.""" loop = asyncio.get_running_loop() with concurrent.futures.ThreadPoolExecutor() as pool: result = await loop.run_in_executor(pool, blocking_operation, data) return result async def main(): results = await asyncio.gather(*[run_in_executor(i) for i in range(5)]) print(results) asyncio.run(main()) ``` ## Common Pitfalls ### 1. Forgetting await ```python # Wrong - returns coroutine object, doesn't execute result = async_function() # Correct result = await async_function() ``` ### 2. Blocking the Event Loop ```python # Wrong - blocks event loop import time async def bad(): time.sleep(1) # Blocks! # Correct async def good(): await asyncio.sleep(1) # Non-blocking ``` ### 3. Not Handling Cancellation ```python async def cancelable_task(): """Task that handles cancellation.""" try: while True: await asyncio.sleep(1) print("Working...") except asyncio.CancelledError: print("Task cancelled, cleaning up...") # Perform cleanup raise # Re-raise to propagate cancellation ``` ### 4. Mixing Sync and Async Code ```python # Wrong - can't call async from sync directly def sync_function(): result = await async_function() # SyntaxError! # Correct def sync_function(): result = asyncio.run(async_function()) ``` ## Testing Async Code ```python import asyncio import pytest # Using pytest-asyncio @pytest.mark.asyncio async def test_async_function(): """Test async function.""" result = await fetch_data("https
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

backtesting-frameworks

Build robust backtesting systems for trading strategies with proper

coding
⭐1
# Backtesting Frameworks Build robust, production-grade backtesting systems that avoid common pitfalls and produce reliable strategy performance estimates. ## When to Use This Skill - Developing trading strategy backtests - Building backtesting infrastructure - Validating strategy performance - Avoiding common backtesting biases - Implementing walk-forward analysis - Comparing strategy alternatives ## Core Concepts ### 1. Backtesting Biases | Bias | Description | Mitigation | | ---------------- | ------------------------- | ----------------------- | | **Look-ahead** | Using future information | Point-in-time data | | **Survivorship** | Only testing on survivors | Use delisted securities | | **Overfitting** | Curve-fitting to history | Out-of-sample testing | | **Selection** | Cherry-picking strategies | Pre-registration | | **Transaction** | Ignoring trading costs | Realistic cost models | ### 2. Proper Backtest Structure ``` Historical Data β”‚ β–Ό β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Training Set β”‚ β”‚ (Strategy Development & Optimization) β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β–Ό β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Validation Set β”‚ β”‚ (Parameter Selection, No Peeking) β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β–Ό β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Test Set β”‚ β”‚ (Final Performance Evaluation) β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` ### 3. Walk-Forward Analysis ``` Window 1: [Train──────][Test] Window 2: [Train──────][Test] Window 3: [Train──────][Test] Window 4: [Train──────][Test] ─────▢ Time ``` ## Implementation Patterns ### Pattern 1: Event-Driven Backtester ```python from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime from decimal import Decimal from enum import Enum from typing import Dict, List, Optional import pandas as pd import numpy as np class OrderSide(Enum): BUY = "buy" SELL = "sell" class OrderType(Enum): MARKET = "market" LIMIT = "limit" STOP = "stop" @dataclass class Order: symbol: str side: OrderSide quantity: Decimal order_type: OrderType limit_price: Optional[Decimal] = None stop_price: Optional[Decimal] = None timestamp: Optional[datetime] = None @dataclass class Fill: order: Order fill_price: Decimal fill_quantity: Decimal commission: Decimal slippage: Decimal timestamp: datetime @dataclass class Position: symbol: str quantity: Decimal = Decimal("0") avg_cost: Decimal = Decimal("0") realized_pnl: Decimal = Decimal("0") def update(self, fill: Fill) -> None: if fill.order.side == OrderSide.BUY: new_quantity = self.quantity + fill.fill_quantity if new_quantity != 0: self.avg_cost = ( (self.quantity * self.avg_cost + fill.fill_quantity * fill.fill_price) / new_quantity ) self.quantity = new_quantity else: self.realized_pnl += fill.fill_quantity * (fill.fill_price - self.avg_cost) self.quantity -= fill.fill_quantity @dataclass class Portfolio: cash: Decimal positions: Dict[str, Position] = field(default_factory=dict) def get_position(self, symbol: str) -> Position: if symbol not in self.positions: self.positions[symbol] = Position(symbol=symbol) return self.positions[symbol] def process_fill(self, fill: Fill) -> None: position = self.get_position(fill.order.symbol) position.update(fill) if fill.order.side == OrderSide.BUY: self.cash -= fill.fill_price * fill.fill_quantity + fill.commission else: self.cash += fill.fill_price * fill.fill_quantity - fill.commission def get_equity(self, prices: Dict[str, Decimal]) -> Decimal: equity = self.cash for symbol, position in self.positions.items(): if position.quantity != 0 and symbol in prices: equity += position.quantity * prices[symbol] return equity class Strategy(ABC): @abstractmethod def on_bar(self, timestamp: datetime, data: pd.DataFrame) -> List[Order]: pass @abstractmethod def on_fill(self, fill: Fill) -> None: pass class ExecutionModel(ABC): @abstractmethod def execute(self, order: Order, bar: pd.Series) -> Optional[Fill]: pass class SimpleExecutionModel(ExecutionModel): def __init__(self, slippage_bps: float = 10, commission_per_share: float = 0.01): self.slippage_bps = slippage_bps self.commission_per_share = commission_per_share def execute(self, order: Order, bar: pd.Series) -> Optional[Fill]: if order.order_type == OrderType.MARKET: base_price = Decimal(str(bar["open"])) # Apply slippage slippage_mult = 1 + (self.slippage_bps / 10000) if order.side == OrderSide.BUY: fill_price = base_price * Decimal(str(slippage_mult)) else: fill_price = base_price / Decimal(str(slippage_mult)) commission = order.quantity * Decimal(str(self.commission_per_share)) slippage = abs(fill_price - base_price) * order.quantity return Fill( order=order, fill_price=fill_price, fill_quantity=order.quantity, commission=commission, slippage=slippage, timestamp=bar.name ) return None class Backtester: def __init__( self, strategy: Strategy, execution_model: ExecutionModel, initial_capital: Decimal = Decimal("100000") ): self.strategy = strategy self.execution_model = execution_model self.portfolio = Portfolio(cash=initial_capital) self.equity_curve: List[tuple] = [] self.trades: List[Fill] = [] def run(self, data: pd.DataFrame) -> pd.DataFrame: """Run backtest on OHLCV data with DatetimeIndex.""" pending_orders: List[Order] = [] for timestamp, bar in data.iterrows(): # Execute pending orders at today's prices for order in pending_orders: fill = self.execution_model.execute(order, bar) if fill: self.portfolio.process_fill(fill) self.strategy.on_fill(fill) self.trades.append(fill) pending_orders.clear() # Get current prices for equity calculation prices = {data.index.name or "default": Decimal(str(bar["close"]))} equity = self.portfolio.get_equity(prices) self.equity_curve.append((timestamp, float(equity))) # Generate new orders for next bar new_orders = self.strategy.on_bar(timestamp, data.loc[:timestamp]) pending_orders.extend(new_orders) return self._create_results() def _create_results(self) -> pd.DataFrame: equity_df = pd.DataFrame(self.equity_curve, columns=["timestamp", "equity"]) equity_df.set_index("timestamp", inplace=True) equity_df["returns"] = equity_df["equity"].pct_change() return equity_df ``` ### Pattern 2: Vectorized Backtester (Fast) ```python import pandas as pd import numpy as np from typing import Callable, Dict, Any class VectorizedBacktester: """Fast vectorized backtester for simple strategies.""" def __init__( self, initial_capital: float = 100000, commission: float = 0.001, # 0.1% slippage: float = 0.0005 # 0.05% ): self.initial_capital = initial_capital self.commission = commission self.slippage = slippage def run( self, prices: pd.DataFrame, signal_func: Callable[[pd.DataFrame], pd.Series] ) -> Dict[str, Any]: """ Run backtest with signal function. Args: prices: DataFrame with 'close' column signal_func: Function that returns position signals (-1, 0, 1) Returns: Dictionary with results """ # Generate signals (shifted to avoid look-ahead) signals = signal_func(prices).shift(1).fillna(0) # Calculate returns returns = prices["close"].pct_change() # Calculate strategy returns with costs position_changes = signals.diff().abs() trading_costs = position_changes * (self.commission + self.slippage) strategy_returns = signals * returns - trading_costs # Build equity curve equity = (1 + strategy_returns).cumprod() * self.initial_capital # Calculate metrics results = { "equity": equity, "returns": strategy_returns, "signals": signals, "metrics": self._calculate_metrics(strategy_returns, equity) } return results def _calculate_metrics( self, returns: pd.Series, equity: pd.Series ) -> Dict[str, float]: """Calculate performance metrics.""" total_return = (equity.iloc[-1] / self.initial_capital) - 1 annual_return = (1 + total_return) ** (252 / len(returns)) - 1 annual_vol = returns.std() * np.sqrt(252) sharpe = annual_return / annual_vol if annual_vol > 0 else 0 # Drawdown rolling_max = equity.cummax() drawdown = (equity - rolling_max) / rolling_max max_drawdown = drawdown.min() # Win rate winning_days = (returns > 0).sum() total_days = (returns != 0).sum() win_rate = winning_days / total_days if total_days > 0 else 0 return { "total_return": total_return, "annual_return": annual_return, "annual_volatility": annual_vol, "sharpe_ratio": sharpe, "max_drawdown": max_drawdown, "win_rate": win_rate, "num_trades": int((returns != 0).sum()) } # Example usage def momentum_signal(prices: pd.DataFrame, lookback: int = 20) -> pd.Series: """Simple momentum strategy: long when price > SMA, else flat.""" sma = prices["close"].rolling(lookback).mean() return (prices["close"] > sma).astype(int) # Run backtest # backtester = VectorizedBacktester() # results = backtester.run(price_data, lambda p: momentum_signal(p, 50)) ``` ### Pattern 3: Walk-Forward Optimization ```python from typing import Callable, Dict, List, Tuple, Any import pandas as pd import numpy as np from itertools import product class WalkForwardOptimizer: """Walk-forward analysis with anchored or rolling windows.""" def __init__( self, train_period: int, test_period: int, anchored: bool = False, n_splits: int = None ): """ Args: train_period: Number of bars in training window test_period: Number of bars in test window anchored: If True, training always starts from beginning n_splits: Number of train/test splits (auto-calculated if None) """ self.train_period = train_period self.test_period = test_period self.anchored = anchored self.n_splits = n_splits def generate_splits( self, data: pd.DataFrame ) -> List[Tuple[pd.DataFrame, pd.DataFrame]]: """Generate train/test splits.""" splits = [] n = len(data) if self.n_splits: step = (n - self.train_period) // self.n_splits else: step = self.test_period start = 0 while start + self.train_period + self.test_period <= n: if self.anchored: train_start = 0 else: train_start = start train_end = start + self.train_period test_end = min(train_end + self.test_period, n) train_data = data.iloc[train_start:train_end] test_data = data.iloc[train_end:test_end] splits.append((train_data, test_data)) start += step return splits def optimize( self, data: pd.DataFrame, strategy_func: Callable, param_grid: Dict[str, List], metric: str = "sharpe_ratio" ) -> Dict[str, Any]: """ Run walk-forward optimization. Args: data: Full dataset strategy_func: Function(data, **params) -> results dict param_grid: Parameter combinations to test metric: Metric to optimize Returns: Combined results from all test periods """ splits = self.generate_splits(data) all_results = [] optimal_params_history = [] for i, (train_data, test_data) in enumerate(splits): # Optimize on training data best_params, best_metric = self._grid_search( train_data, strategy_func, param_grid, metric ) optimal_params_history.append(best_params) # Test with optimal params test_results = strategy_func(test_data, **best_params) test_results["split"] = i test_results["params"] = best_params all_results.append(test_results) print(f"Split {i+1}/{len(splits)}: " f"Best {metric}={best_metric:.4f}, params={best_params}") return { "split_results": all_results, "param_history": optimal_params_history, "combined_equity": self._combine_equity_curves(all_results) } def _grid_search( self, data: pd.DataFrame, strategy_func: Callable, param_grid: Dict[str, List], metric: str ) -> Tuple[Dict, float]: """Grid search for best parameters.""" best_params = None best_metric = -np.inf # Generate all parameter combinations param_names = list(param_grid.keys()) param_values = list(param_grid.values()) for values in product(*param_values): params = dict(zip(param_names, values)) results = strategy_func(data, **params) if results["metrics"][metric] > best_metric: best_metric = results["metrics"][metric] best_params = params return best_params, best_metric def _combine_equity_curves( self, results: List[Dict] ) -> pd.Series: """Combine equity curves from all test periods.""" combined = pd.concat([r["equity"] for r in results]) return combined ``` ### Pattern 4: Monte Carlo Analysis ```python import numpy as np import pandas as pd from typing import Dict, List class MonteCarloAnalyzer: """Monte Carlo simulation for strategy robustness.""" def __init__(self, n_simulations: int = 1000, confidence: float = 0.95): self.n_simulations = n_simulations self.confidence = confidence def bootstrap_returns( self, returns: pd.Series, n_periods: int = None ) -> np.ndarray: """ Bootstrap simulation by resampling returns. Args: returns: Historical returns series n_periods: Length of each simulation (default: same as input) Returns: Array of shape (n_simulations, n_periods) """ if n_periods is None: n_periods = len(returns) simulations = np.zeros((self.n_simulations, n_periods)) for i in range(self.n_simulations): # Resample with replacement simulated_returns = np.random.choice( returns.values, size=n_periods, replace=True ) simulations[i] = simulated_returns return simulations def analyze_drawdowns( self, returns: pd.Series ) -> Dict[str, float]: """Analyze drawdown distribution via simulation.""" simulations = self.bootstrap_returns(returns) max_drawdowns = [] for sim_returns in simulations: equity = (1 + sim_returns).cumprod() rolling_max = np.maximum.accumulate(equity) drawdowns = (equity - rolling_max) / rolling_max max_drawdowns.append(drawdowns.min()) max_drawdowns = np.array(max_drawdowns) return { "expected_max_dd": np.mean(max_drawdowns), "median_max_dd": np.median(max_drawdowns), f"worst_{int(self.confidence*100)}pct": np.percentile( max_drawdowns, (1 - self.confidence) * 100 ), "worst_case": max_drawdowns.min() } def probability_of_loss( self, returns: pd.Series, holding_periods: List[int] = [21, 63, 126, 252] ) -> Dict[int, float]: """Calculate probability of loss over various holding periods.""" results = {} for period in holding_periods: if period > len(returns): continue simulations = self.bootstrap_returns(returns, period) total_returns = (1 + simulations).prod(axis=1) - 1 prob_loss = (total_returns < 0).mean() results[period] = prob_loss return results def confidence_interval( self, returns: pd.Series, periods: int = 252 ) -> Dict[str, float]: """Calculate confidence interval for future returns.""" simulations = self.bootstrap_returns(returns, periods) total_returns = (1 + simulations).prod(axis=1) - 1 lower = (1 - self.confidence) / 2 upper = 1 - lower return { "expected": total_returns.mean(), "lower_bound": np.percentile(total_returns, lower * 100), "upper_bound": np.percentile(total_returns, upper * 100), "std": total_returns.std() } ``` ## Performance Metrics ```python def calculate_metrics(returns: pd.Series, rf_rate: float = 0.02) -> Dict[str, float]: """Calculate comprehensive performance metrics.""" # Annualization factor (assuming daily returns) ann_factor = 252 # Basic metrics total_return = (1 + returns).prod() - 1 annual_return = (1 + total_return) ** (ann_factor / len(returns)) - 1 annual_vol = returns.std() * np.sqrt(ann_factor) # Risk-adjusted returns sharpe = (annual_return - rf_rate) / annual_vol if annual_vol > 0 else 0 # Sortino (downside deviation) downside_returns = returns[returns < 0] downside_vol = downside_returns.std() * np.sqrt(ann_factor) sortino = (annual_return - rf_rate) / downside_vol if downside_vol > 0 else 0 # Calmar ratio equity = (1 + returns).cumprod() rolling_max = equity.cummax() drawdowns = (equity - rolling_max) / rolling_max max_drawdown = drawdowns.min() calmar = annual_return / abs(max_drawdown) if max_drawdown != 0 else 0 # Win rate and profit factor wins = returns[returns > 0] losses = return
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

solidity-security

Master smart contract security best practices to prevent common

security
⭐1
# Solidity Security Master smart contract security best practices, vulnerability prevention, and secure Solidity development patterns. ## When to Use This Skill - Writing secure smart contracts - Auditing existing contracts for vulnerabilities - Implementing secure DeFi protocols - Preventing reentrancy, overflow, and access control issues - Optimizing gas usage while maintaining security - Preparing contracts for professional audits - Understanding common attack vectors ## Critical Vulnerabilities ### 1. Reentrancy Attacker calls back into your contract before state is updated. **Vulnerable Code:** ```solidity // VULNERABLE TO REENTRANCY contract VulnerableBank { mapping(address => uint256) public balances; function withdraw() public { uint256 amount = balances[msg.sender]; // DANGER: External call before state update (bool success, ) = msg.sender.call{value: amount}(""); require(success); balances[msg.sender] = 0; // Too late! } } ``` **Secure Pattern (Checks-Effects-Interactions):** ```solidity contract SecureBank { mapping(address => uint256) public balances; function withdraw() public { uint256 amount = balances[msg.sender]; require(amount > 0, "Insufficient balance"); // EFFECTS: Update state BEFORE external call balances[msg.sender] = 0; // INTERACTIONS: External call last (bool success, ) = msg.sender.call{value: amount}(""); require(success, "Transfer failed"); } } ``` **Alternative: ReentrancyGuard** ```solidity import "@openzeppelin/contracts/security/ReentrancyGuard.sol"; contract SecureBank is ReentrancyGuard { mapping(address => uint256) public balances; function withdraw() public nonReentrant { uint256 amount = balances[msg.sender]; require(amount > 0, "Insufficient balance"); balances[msg.sender] = 0; (bool success, ) = msg.sender.call{value: amount}(""); require(success, "Transfer failed"); } } ``` ### 2. Integer Overflow/Underflow **Vulnerable Code (Solidity < 0.8.0):** ```solidity // VULNERABLE contract VulnerableToken { mapping(address => uint256) public balances; function transfer(address to, uint256 amount) public { // No overflow check - can wrap around balances[msg.sender] -= amount; // Can underflow! balances[to] += amount; // Can overflow! } } ``` **Secure Pattern (Solidity >= 0.8.0):** ```solidity // Solidity 0.8+ has built-in overflow/underflow checks contract SecureToken { mapping(address => uint256) public balances; function transfer(address to, uint256 amount) public { // Automatically reverts on overflow/underflow balances[msg.sender] -= amount; balances[to] += amount; } } ``` **For Solidity < 0.8.0, use SafeMath:** ```solidity import "@openzeppelin/contracts/utils/math/SafeMath.sol"; contract SecureToken { using SafeMath for uint256; mapping(address => uint256) public balances; function transfer(address to, uint256 amount) public { balances[msg.sender] = balances[msg.sender].sub(amount); balances[to] = balances[to].add(amount); } } ``` ### 3. Access Control **Vulnerable Code:** ```solidity // VULNERABLE: Anyone can call critical functions contract VulnerableContract { address public owner; function withdraw(uint256 amount) public { // No access control! payable(msg.sender).transfer(amount); } } ``` **Secure Pattern:** ```solidity import "@openzeppelin/contracts/access/Ownable.sol"; contract SecureContract is Ownable { function withdraw(uint256 amount) public onlyOwner { payable(owner()).transfer(amount); } } // Or implement custom role-based access contract RoleBasedContract { mapping(address => bool) public admins; modifier onlyAdmin() { require(admins[msg.sender], "Not an admin"); _; } function criticalFunction() public onlyAdmin { // Protected function } } ``` ### 4. Front-Running **Vulnerable:** ```solidity // VULNERABLE TO FRONT-RUNNING contract VulnerableDEX { function swap(uint256 amount, uint256 minOutput) public { // Attacker sees this in mempool and front-runs uint256 output = calculateOutput(amount); require(output >= minOutput, "Slippage too high"); // Perform swap } } ``` **Mitigation:** ```solidity contract SecureDEX { mapping(bytes32 => bool) public usedCommitments; // Step 1: Commit to trade function commitTrade(bytes32 commitment) public { usedCommitments[commitment] = true; } // Step 2: Reveal trade (next block) function revealTrade( uint256 amount, uint256 minOutput, bytes32 secret ) public { bytes32 commitment = keccak256(abi.encodePacked( msg.sender, amount, minOutput, secret )); require(usedCommitments[commitment], "Invalid commitment"); // Perform swap } } ``` ## Security Best Practices ### Checks-Effects-Interactions Pattern ```solidity contract SecurePattern { mapping(address => uint256) public balances; function withdraw(uint256 amount) public { // 1. CHECKS: Validate conditions require(amount <= balances[msg.sender], "Insufficient balance"); require(amount > 0, "Amount must be positive"); // 2. EFFECTS: Update state balances[msg.sender] -= amount; // 3. INTERACTIONS: External calls last (bool success, ) = msg.sender.call{value: amount}(""); require(success, "Transfer failed"); } } ``` ### Pull Over Push Pattern ```solidity // Prefer this (pull) contract SecurePayment { mapping(address => uint256) public pendingWithdrawals; function recordPayment(address recipient, uint256 amount) internal { pendingWithdrawals[recipient] += amount; } function withdraw() public { uint256 amount = pendingWithdrawals[msg.sender]; require(amount > 0, "Nothing to withdraw"); pendingWithdrawals[msg.sender] = 0; payable(msg.sender).transfer(amount); } } // Over this (push) contract RiskyPayment { function distributePayments(address[] memory recipients, uint256[] memory amounts) public { for (uint i = 0; i < recipients.length; i++) { // If any transfer fails, entire batch fails payable(recipients[i]).transfer(amounts[i]); } } } ``` ### Input Validation ```solidity contract SecureContract { function transfer(address to, uint256 amount) public { // Validate inputs require(to != address(0), "Invalid recipient"); require(to != address(this), "Cannot send to contract"); require(amount > 0, "Amount must be positive"); require(amount <= balances[msg.sender], "Insufficient balance"); // Proceed with transfer balances[msg.sender] -= amount; balances[to] += amount; } } ``` ### Emergency Stop (Circuit Breaker) ```solidity import "@openzeppelin/contracts/security/Pausable.sol"; contract EmergencyStop is Pausable, Ownable { function criticalFunction() public whenNotPaused { // Function logic } function emergencyStop() public onlyOwner { _pause(); } function resume() public onlyOwner { _unpause(); } } ``` ## Gas Optimization ### Use `uint256` Instead of Smaller Types ```solidity // More gas efficient contract GasEfficient { uint256 public value; // Optimal function set(uint256 _value) public { value = _value; } } // Less efficient contract GasInefficient { uint8 public value; // Still uses 256-bit slot function set(uint8 _value) public { value = _value; // Extra gas for type conversion } } ``` ### Pack Storage Variables ```solidity // Gas efficient (3 variables in 1 slot) contract PackedStorage { uint128 public a; // Slot 0 uint64 public b; // Slot 0 uint64 public c; // Slot 0 uint256 public d; // Slot 1 } // Gas inefficient (each variable in separate slot) contract UnpackedStorage { uint256 public a; // Slot 0 uint256 public b; // Slot 1 uint256 public c; // Slot 2 uint256 public d; // Slot 3 } ``` ### Use `calldata` Instead of `memory` for Function Arguments ```solidity contract GasOptimized { // More gas efficient function processData(uint256[] calldata data) public pure returns (uint256) { return data[0]; } // Less efficient function processDataMemory(uint256[] memory data) public pure returns (uint256) { return data[0]; } } ``` ### Use Events for Data Storage (When Appropriate) ```solidity contract EventStorage { // Emitting events is cheaper than storage event DataStored(address indexed user, uint256 indexed id, bytes data); function storeData(uint256 id, bytes calldata data) public { emit DataStored(msg.sender, id, data); // Don't store in contract storage unless needed } } ``` ## Common Vulnerabilities Checklist ```solidity // Security Checklist Contract contract SecurityChecklist { /** * [ ] Reentrancy protection (ReentrancyGuard or CEI pattern) * [ ] Integer overflow/underflow (Solidity 0.8+ or SafeMath) * [ ] Access control (Ownable, roles, modifiers) * [ ] Input validation (require statements) * [ ] Front-running mitigation (commit-reveal if applicable) * [ ] Gas optimization (packed storage, calldata) * [ ] Emergency stop mechanism (Pausable) * [ ] Pull over push pattern for payments * [ ] No delegatecall to untrusted contracts * [ ] No tx.origin for authentication (use msg.sender) * [ ] Proper event emission * [ ] External calls at end of function * [ ] Check return values of external calls * [ ] No hardcoded addresses * [ ] Upgrade mechanism (if proxy pattern) */ } ``` ## Testing for Security ```javascript // Hardhat test example const { expect } = require("chai"); const { ethers } = require("hardhat"); describe("Security Tests", function () { it("Should prevent reentrancy attack", async function () { const [attacker] = await ethers.getSigners(); const VictimBank = await ethers.getContractFactory("SecureBank"); const bank = await VictimBank.deploy(); const Attacker = await ethers.getContractFactory("ReentrancyAttacker"); const attackerContract = await Attacker.deploy(bank.address); // Deposit funds await bank.deposit({ value: ethers.utils.parseEther("10") }); // Attempt reentrancy attack await expect( attackerContract.attack({ value: ethers.utils.parseEther("1") }), ).to.be.revertedWith("ReentrancyGuard: reentrant call"); }); it("Should prevent integer overflow", async function () { const Token = await ethers.getContractFactory("SecureToken"); const token = await Token.deploy(); // Attempt overflow await expect(token.transfer(attacker.address, ethers.constants.MaxUint256)) .to.be.reverted; }); it("Should enforce access control", async function () { const [owner, attacker] = await ethers.getSigners(); const Contract = await ethers.getContractFactory("SecureContract"); const contract = await Contract.deploy(); // Attempt unauthorized withdrawal await expect(contract.connect(attacker).withdraw(100)).to.be.revertedWith( "Ownable: caller is not the owner", ); }); }); ``` ## Audit Preparation ```solidity contract WellDocumentedContract { /** * @title Well Documented Contract * @dev Example of proper documentation for audits * @notice This contract handles user deposits and withdrawals */ /// @notice Mapping of user balances mapping(address => uint256) public balances; /** * @dev Deposits ETH into the contract * @notice Anyone can deposit funds */ function deposit() public payable { require(msg.value > 0, "Must send ETH"); balances[msg.sender] += msg.value; } /** * @dev Withdraws user's balance * @notice Follows CEI pattern to prevent reentrancy * @param amount Amount to withdraw in wei */ function withdraw(uint256 amount) public { // CHECKS require(amount <= balances[msg.sender], "Insufficient balance"); // EFFECTS balances[msg.sender] -= amount; // INTERACTIONS (bool success, ) = msg.sender.call{value: amount}(""); require(success, "Transfer failed"); } } ``` ## Resources - **references/reentrancy.md**: Comprehensive reentrancy prevention - **references/access-control.md**: Role-based access patterns - **references/overflow-underflow.md**: SafeMath and integer safety - **references/gas-optimization.md**: Gas saving techniques - **references/vulnerability-patterns.md**: Common vulnerability catalog - **assets/solidity-contracts-templates.sol**: Secure contract templates - **assets/security-checklist.md**: Pre-audit checklist - **scripts/analyze-contract.sh**: Static analysis tools ## Tools for Security Analysis - **Slither**: Static analysis tool - **Mythril**: Security analysis tool - **Echidna**: Fuzzing tool - **Manticore**: Symbolic execution - **Securify**: Automated security scanner ## Common Pitfalls 1. **Using `tx.origin` for Authentication**: Use `msg.sender` instead 2. **Unchecked External Calls**: Always check return values 3. **Delegatecall to Untrusted Contracts**: Can hijack your contract 4. **Floating Pragma**: Pin to specific Solidity version 5. **Missing Events**: Emit events for state changes 6. **Excessive Gas in Loops**: Can hit block gas limit 7. **No Upgrade Path**: Consider proxy patterns if upgrades needed
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

attack-tree-construction

Build comprehensive attack trees to visualize threat paths. Use

security
⭐1
# Attack Tree Construction Systematic attack path visualization and analysis. ## When to Use This Skill - Visualizing complex attack scenarios - Identifying defense gaps and priorities - Communicating risks to stakeholders - Planning defensive investments - Penetration test planning - Security architecture review ## Core Concepts ### 1. Attack Tree Structure ``` [Root Goal] | β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ β”‚ [Sub-goal 1] [Sub-goal 2] (OR node) (AND node) β”‚ β”‚ β”Œβ”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β” β”‚ β”‚ β”‚ β”‚ [Attack] [Attack] [Attack] [Attack] (leaf) (leaf) (leaf) (leaf) ``` ### 2. Node Types | Type | Symbol | Description | | -------- | --------- | ----------------------- | | **OR** | Oval | Any child achieves goal | | **AND** | Rectangle | All children required | | **Leaf** | Box | Atomic attack step | ### 3. Attack Attributes | Attribute | Description | Values | | ------------- | ----------------------- | ------------------ | | **Cost** | Resources needed | $, $$, $$$ | | **Time** | Duration to execute | Hours, Days, Weeks | | **Skill** | Expertise required | Low, Medium, High | | **Detection** | Likelihood of detection | Low, Medium, High | ## Templates ### Template 1: Attack Tree Data Model ```python from dataclasses import dataclass, field from enum import Enum from typing import List, Dict, Optional, Union import json class NodeType(Enum): OR = "or" AND = "and" LEAF = "leaf" class Difficulty(Enum): TRIVIAL = 1 LOW = 2 MEDIUM = 3 HIGH = 4 EXPERT = 5 class Cost(Enum): FREE = 0 LOW = 1 MEDIUM = 2 HIGH = 3 VERY_HIGH = 4 class DetectionRisk(Enum): NONE = 0 LOW = 1 MEDIUM = 2 HIGH = 3 CERTAIN = 4 @dataclass class AttackAttributes: difficulty: Difficulty = Difficulty.MEDIUM cost: Cost = Cost.MEDIUM detection_risk: DetectionRisk = DetectionRisk.MEDIUM time_hours: float = 8.0 requires_insider: bool = False requires_physical: bool = False @dataclass class AttackNode: id: str name: str description: str node_type: NodeType attributes: AttackAttributes = field(default_factory=AttackAttributes) children: List['AttackNode'] = field(default_factory=list) mitigations: List[str] = field(default_factory=list) cve_refs: List[str] = field(default_factory=list) def add_child(self, child: 'AttackNode') -> None: self.children.append(child) def calculate_path_difficulty(self) -> float: """Calculate aggregate difficulty for this path.""" if self.node_type == NodeType.LEAF: return self.attributes.difficulty.value if not self.children: return 0 child_difficulties = [c.calculate_path_difficulty() for c in self.children] if self.node_type == NodeType.OR: return min(child_difficulties) else: # AND return max(child_difficulties) def calculate_path_cost(self) -> float: """Calculate aggregate cost for this path.""" if self.node_type == NodeType.LEAF: return self.attributes.cost.value if not self.children: return 0 child_costs = [c.calculate_path_cost() for c in self.children] if self.node_type == NodeType.OR: return min(child_costs) else: # AND return sum(child_costs) def to_dict(self) -> Dict: """Convert to dictionary for serialization.""" return { "id": self.id, "name": self.name, "description": self.description, "type": self.node_type.value, "attributes": { "difficulty": self.attributes.difficulty.name, "cost": self.attributes.cost.name, "detection_risk": self.attributes.detection_risk.name, "time_hours": self.attributes.time_hours, }, "mitigations": self.mitigations, "children": [c.to_dict() for c in self.children] } @dataclass class AttackTree: name: str description: str root: AttackNode version: str = "1.0" def find_easiest_path(self) -> List[AttackNode]: """Find the path with lowest difficulty.""" return self._find_path(self.root, minimize="difficulty") def find_cheapest_path(self) -> List[AttackNode]: """Find the path with lowest cost.""" return self._find_path(self.root, minimize="cost") def find_stealthiest_path(self) -> List[AttackNode]: """Find the path with lowest detection risk.""" return self._find_path(self.root, minimize="detection") def _find_path( self, node: AttackNode, minimize: str ) -> List[AttackNode]: """Recursive path finding.""" if node.node_type == NodeType.LEAF: return [node] if not node.children: return [node] if node.node_type == NodeType.OR: # Pick the best child path best_path = None best_score = float('inf') for child in node.children: child_path = self._find_path(child, minimize) score = self._path_score(child_path, minimize) if score < best_score: best_score = score best_path = child_path return [node] + (best_path or []) else: # AND # Must traverse all children path = [node] for child in node.children: path.extend(self._find_path(child, minimize)) return path def _path_score(self, path: List[AttackNode], metric: str) -> float: """Calculate score for a path.""" if metric == "difficulty": return sum(n.attributes.difficulty.value for n in path if n.node_type == NodeType.LEAF) elif metric == "cost": return sum(n.attributes.cost.value for n in path if n.node_type == NodeType.LEAF) elif metric == "detection": return sum(n.attributes.detection_risk.value for n in path if n.node_type == NodeType.LEAF) return 0 def get_all_leaf_attacks(self) -> List[AttackNode]: """Get all leaf attack nodes.""" leaves = [] self._collect_leaves(self.root, leaves) return leaves def _collect_leaves(self, node: AttackNode, leaves: List[AttackNode]) -> None: if node.node_type == NodeType.LEAF: leaves.append(node) for child in node.children: self._collect_leaves(child, leaves) def get_unmitigated_attacks(self) -> List[AttackNode]: """Find attacks without mitigations.""" return [n for n in self.get_all_leaf_attacks() if not n.mitigations] def export_json(self) -> str: """Export tree to JSON.""" return json.dumps({ "name": self.name, "description": self.description, "version": self.version, "root": self.root.to_dict() }, indent=2) ``` ### Template 2: Attack Tree Builder ```python class AttackTreeBuilder: """Fluent builder for attack trees.""" def __init__(self, name: str, description: str): self.name = name self.description = description self._node_stack: List[AttackNode] = [] self._root: Optional[AttackNode] = None def goal(self, id: str, name: str, description: str = "") -> 'AttackTreeBuilder': """Set the root goal (OR node by default).""" self._root = AttackNode( id=id, name=name, description=description, node_type=NodeType.OR ) self._node_stack = [self._root] return self def or_node(self, id: str, name: str, description: str = "") -> 'AttackTreeBuilder': """Add an OR sub-goal.""" node = AttackNode( id=id, name=name, description=description, node_type=NodeType.OR ) self._current().add_child(node) self._node_stack.append(node) return self def and_node(self, id: str, name: str, description: str = "") -> 'AttackTreeBuilder': """Add an AND sub-goal (all children required).""" node = AttackNode( id=id, name=name, description=description, node_type=NodeType.AND ) self._current().add_child(node) self._node_stack.append(node) return self def attack( self, id: str, name: str, description: str = "", difficulty: Difficulty = Difficulty.MEDIUM, cost: Cost = Cost.MEDIUM, detection: DetectionRisk = DetectionRisk.MEDIUM, time_hours: float = 8.0, mitigations: List[str] = None ) -> 'AttackTreeBuilder': """Add a leaf attack node.""" node = AttackNode( id=id, name=name, description=description, node_type=NodeType.LEAF, attributes=AttackAttributes( difficulty=difficulty, cost=cost, detection_risk=detection, time_hours=time_hours ), mitigations=mitigations or [] ) self._current().add_child(node) return self def end(self) -> 'AttackTreeBuilder': """Close current node, return to parent.""" if len(self._node_stack) > 1: self._node_stack.pop() return self def build(self) -> AttackTree: """Build the attack tree.""" if not self._root: raise ValueError("No root goal defined") return AttackTree( name=self.name, description=self.description, root=self._root ) def _current(self) -> AttackNode: if not self._node_stack: raise ValueError("No current node") return self._node_stack[-1] # Example usage def build_account_takeover_tree() -> AttackTree: """Build attack tree for account takeover scenario.""" return ( AttackTreeBuilder("Account Takeover", "Gain unauthorized access to user account") .goal("G1", "Take Over User Account") .or_node("S1", "Steal Credentials") .attack( "A1", "Phishing Attack", difficulty=Difficulty.LOW, cost=Cost.LOW, detection=DetectionRisk.MEDIUM, mitigations=["Security awareness training", "Email filtering"] ) .attack( "A2", "Credential Stuffing", difficulty=Difficulty.TRIVIAL, cost=Cost.LOW, detection=DetectionRisk.HIGH, mitigations=["Rate limiting", "MFA", "Password breach monitoring"] ) .attack( "A3", "Keylogger Malware", difficulty=Difficulty.MEDIUM, cost=Cost.MEDIUM, detection=DetectionRisk.MEDIUM, mitigations=["Endpoint protection", "MFA"] ) .end() .or_node("S2", "Bypass Authentication") .attack( "A4", "Session Hijacking", difficulty=Difficulty.MEDIUM, cost=Cost.LOW, detection=DetectionRisk.LOW, mitigations=["Secure session management", "HTTPS only"] ) .attack( "A5", "Authentication Bypass Vulnerability", difficulty=Difficulty.HIGH, cost=Cost.LOW, detection=DetectionRisk.LOW, mitigations=["Security testing", "Code review", "WAF"] ) .end() .or_node("S3", "Social Engineering") .and_node("S3.1", "Account Recovery Attack") .attack( "A6", "Gather Personal Information", difficulty=Difficulty.LOW, cost=Cost.FREE, detection=DetectionRisk.NONE ) .attack( "A7", "Call Support Desk", difficulty=Difficulty.MEDIUM, cost=Cost.FREE, detection=DetectionRisk.MEDIUM, mitigations=["Support verification procedures", "Security questions"] ) .end() .end() .build() ) ``` ### Template 3: Mermaid Diagram Generator ```python class MermaidExporter: """Export attack trees to Mermaid diagram format.""" def __init__(self, tree: AttackTree): self.tree = tree self._lines: List[str] = [] self._node_count = 0 def export(self) -> str: """Export tree to Mermaid flowchart.""" self._lines = ["flowchart TD"] self._export_node(self.tree.root, None) return "\n".join(self._lines) def _export_node(self, node: AttackNode, parent_id: Optional[str]) -> str: """Recursively export nodes.""" node_id = f"N{self._node_count}" self._node_count += 1 # Node shape based on type if node.node_type == NodeType.OR: shape = f"{node_id}(({node.name}))" elif node.node_type == NodeType.AND: shape = f"{node_id}[{node.name}]" else: # LEAF # Color based on difficulty style = self._get_leaf_style(node) shape = f"{node_id}[/{node.name}/]" self._lines.append(f" style {node_id} {style}") self._lines.append(f" {shape}") if parent_id: connector = "-->" if node.node_type != NodeType.AND else "==>" self._lines.append(f" {parent_id} {connector} {node_id}") for child in node.children: self._export_node(child, node_id) return node_id def _get_leaf_style(self, node: AttackNode) -> str: """Get style based on attack attributes.""" colors = { Difficulty.TRIVIAL: "fill:#ff6b6b", # Red - easy attack Difficulty.LOW: "fill:#ffa06b", Difficulty.MEDIUM: "fill:#ffd93d", Difficulty.HIGH: "fill:#6bcb77", Difficulty.EXPERT: "fill:#4d96ff", # Blue - hard attack } color = colors.get(node.attributes.difficulty, "fill:#gray") return color class PlantUMLExporter: """Export attack trees to PlantUML format.""" def __init__(self, tree: AttackTree): self.tree = tree def export(self) -> str: """Export tree to PlantUML.""" lines = [ "@startmindmap", f"* {self.tree.name}", ] self._export_node(self.tree.root, lines, 1) lines.append("@endmindmap") return "\n".join(lines) def _export_node(self, node: AttackNode, lines: List[str], depth: int) -> None: """Recursively export nodes.""" prefix = "*" * (depth + 1) if node.node_type == NodeType.OR: marker = "[OR]" elif node.node_type == NodeType.AND: marker = "[AND]" else: diff = node.attributes.difficulty.name marker = f"<<{diff}>>" lines.append(f"{prefix} {marker} {node.name}") for child in node.children: self._export_node(child, lines, depth + 1) ``` ### Template 4: Attack Path Analysis ```python from typing import Set, Tuple class AttackPathAnalyzer: """Analyze attack paths and coverage.""" def __init__(self, tree: AttackTree): self.tree = tree def get_all_paths(self) -> List[List[AttackNode]]: """Get all possible attack paths.""" paths = [] self._collect_paths(self.tree.root, [], paths) return paths def _collect_paths( self, node: AttackNode, current_path: List[AttackNode], all_paths: List[List[AttackNode]] ) -> None: """Recursively collect all paths.""" current_path = current_path + [node] if node.node_type == NodeType.LEAF: all_paths.append(current_path) return if not node.children: all_paths.append(current_path) return if node.node_type == NodeType.OR: # Each child is a separate path for child in node.children: self._collect_paths(child, current_path, all_paths) else: # AND # Must combine all children child_paths = [] for child in node.children: child_sub_paths = [] self._collect_paths(child, [], child_sub_paths) child_paths.append(child_sub_paths) # Combine paths from all AND children combined = self._combine_and_paths(child_paths) for combo in combined: all_paths.append(current_path + combo) def _combine_and_paths( self, child_paths: List[List[List[AttackNode]]] ) -> List[List[AttackNode]]: """Combine paths from AND node children.""" if not child_paths: return [[]] if len(child_paths) == 1: return [path for paths in child_paths for path in paths] # Cartesian product of all child path combinations result = [[]] for paths in child_paths: new_result = [] for existing in result: for path in paths: new_result.append(existing + path) result = new_result return result def calculate_path_metrics(self, path: List[AttackNode]) -> Dict: """Calculate metrics for a specific path.""" leaves = [n for n in path if n.node_type == NodeType.LEAF] total_difficulty = sum(n.attributes.difficulty.value for n in leaves) total_cost = sum(n.attributes.cost.value for n in leaves) total_time = sum(n.attributes.time_hours for n in leaves) max_detection = max((n.attributes.detection_risk.value for n in leaves), default=0) return { "steps": len(leaves), "total_difficulty": total_difficulty, "avg_difficulty": total_difficulty / len(leaves) if leaves else 0, "total_cost": total_cost, "total_time_hours": total_time, "max_detection_risk": max_detection, "requires_insider": any(n.attributes.requires_insider for n in leaves), "requires_physical": any(n.attributes.requires_physical for n in leaves), } def identify_critical_nodes(self) -> List[Tuple[AttackNode, int]]: """Find nodes that appear in the most paths.""" paths = self.get_all_paths() node_counts: Dict[str, Tuple[AttackNode, int]] = {} for path in paths: for node in path: if node.id not in node_counts: node_counts[node.id] = (node, 0) node_counts[node.id] = (node, node_counts[node.id][1] + 1) return sort
πŸ‘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
πŸ€–system promptβ€’7 months ago

threat-mitigation-mapping

Map identified threats to appropriate security controls and

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

code-review-excellence

Master effective code review practices to provide constructive

coding
⭐1
# Code Review Excellence Transform code reviews from gatekeeping to knowledge sharing through constructive feedback, systematic analysis, and collaborative improvement. ## When to Use This Skill - Reviewing pull requests and code changes - Establishing code review standards for teams - Mentoring junior developers through reviews - Conducting architecture reviews - Creating review checklists and guidelines - Improving team collaboration - Reducing code review cycle time - Maintaining code quality standards ## Core Principles ### 1. The Review Mindset **Goals of Code Review:** - Catch bugs and edge cases - Ensure code maintainability - Share knowledge across team - Enforce coding standards - Improve design and architecture - Build team culture **Not the Goals:** - Show off knowledge - Nitpick formatting (use linters) - Block progress unnecessarily - Rewrite to your preference ### 2. Effective Feedback **Good Feedback is:** - Specific and actionable - Educational, not judgmental - Focused on the code, not the person - Balanced (praise good work too) - Prioritized (critical vs nice-to-have) ```markdown ❌ Bad: "This is wrong." βœ… Good: "This could cause a race condition when multiple users access simultaneously. Consider using a mutex here." ❌ Bad: "Why didn't you use X pattern?" βœ… Good: "Have you considered the Repository pattern? It would make this easier to test. Here's an example: [link]" ❌ Bad: "Rename this variable." βœ… Good: "[nit] Consider `userCount` instead of `uc` for clarity. Not blocking if you prefer to keep it." ``` ### 3. Review Scope **What to Review:** - Logic correctness and edge cases - Security vulnerabilities - Performance implications - Test coverage and quality - Error handling - Documentation and comments - API design and naming - Architectural fit **What Not to Review Manually:** - Code formatting (use Prettier, Black, etc.) - Import organization - Linting violations - Simple typos ## Review Process ### Phase 1: Context Gathering (2-3 minutes) ```markdown Before diving into code, understand: 1. Read PR description and linked issue 2. Check PR size (>400 lines? Ask to split) 3. Review CI/CD status (tests passing?) 4. Understand the business requirement 5. Note any relevant architectural decisions ``` ### Phase 2: High-Level Review (5-10 minutes) ```markdown 1. **Architecture & Design** - Does the solution fit the problem? - Are there simpler approaches? - Is it consistent with existing patterns? - Will it scale? 2. **File Organization** - Are new files in the right places? - Is code grouped logically? - Are there duplicate files? 3. **Testing Strategy** - Are there tests? - Do tests cover edge cases? - Are tests readable? ``` ### Phase 3: Line-by-Line Review (10-20 minutes) ```markdown For each file: 1. **Logic & Correctness** - Edge cases handled? - Off-by-one errors? - Null/undefined checks? - Race conditions? 2. **Security** - Input validation? - SQL injection risks? - XSS vulnerabilities? - Sensitive data exposure? 3. **Performance** - N+1 queries? - Unnecessary loops? - Memory leaks? - Blocking operations? 4. **Maintainability** - Clear variable names? - Functions doing one thing? - Complex code commented? - Magic numbers extracted? ``` ### Phase 4: Summary & Decision (2-3 minutes) ```markdown 1. Summarize key concerns 2. Highlight what you liked 3. Make clear decision: - βœ… Approve - πŸ’¬ Comment (minor suggestions) - πŸ”„ Request Changes (must address) 4. Offer to pair if complex ``` ## Review Techniques ### Technique 1: The Checklist Method ```markdown ## Security Checklist - [ ] User input validated and sanitized - [ ] SQL queries use parameterization - [ ] Authentication/authorization checked - [ ] Secrets not hardcoded - [ ] Error messages don't leak info ## Performance Checklist - [ ] No N+1 queries - [ ] Database queries indexed - [ ] Large lists paginated - [ ] Expensive operations cached - [ ] No blocking I/O in hot paths ## Testing Checklist - [ ] Happy path tested - [ ] Edge cases covered - [ ] Error cases tested - [ ] Test names are descriptive - [ ] Tests are deterministic ``` ### Technique 2: The Question Approach Instead of stating problems, ask questions to encourage thinking: ```markdown ❌ "This will fail if the list is empty." βœ… "What happens if `items` is an empty array?" ❌ "You need error handling here." βœ… "How should this behave if the API call fails?" ❌ "This is inefficient." βœ… "I see this loops through all users. Have we considered the performance impact with 100k users?" ``` ### Technique 3: Suggest, Don't Command ````markdown ## Use Collaborative Language ❌ "You must change this to use async/await" βœ… "Suggestion: async/await might make this more readable: `typescript async function fetchUser(id: string) { const user = await db.query('SELECT * FROM users WHERE id = ?', id); return user; } ` What do you think?" ❌ "Extract this into a function" βœ… "This logic appears in 3 places. Would it make sense to extract it into a shared utility function?" ```` ### Technique 4: Differentiate Severity ```markdown Use labels to indicate priority: πŸ”΄ [blocking] - Must fix before merge 🟑 [important] - Should fix, discuss if disagree 🟒 [nit] - Nice to have, not blocking πŸ’‘ [suggestion] - Alternative approach to consider πŸ“š [learning] - Educational comment, no action needed πŸŽ‰ [praise] - Good work, keep it up! Example: "πŸ”΄ [blocking] This SQL query is vulnerable to injection. Please use parameterized queries." "🟒 [nit] Consider renaming `data` to `userData` for clarity." "πŸŽ‰ [praise] Excellent test coverage! This will catch edge cases." ``` ## Language-Specific Patterns ### Python Code Review ```python # Check for Python-specific issues # ❌ Mutable default arguments def add_item(item, items=[]): # Bug! Shared across calls items.append(item) return items # βœ… Use None as default def add_item(item, items=None): if items is None: items = [] items.append(item) return items # ❌ Catching too broad try: result = risky_operation() except: # Catches everything, even KeyboardInterrupt! pass # βœ… Catch specific exceptions try: result = risky_operation() except ValueError as e: logger.error(f"Invalid value: {e}") raise # ❌ Using mutable class attributes class User: permissions = [] # Shared across all instances! # βœ… Initialize in __init__ class User: def __init__(self): self.permissions = [] ``` ### TypeScript/JavaScript Code Review ```typescript // Check for TypeScript-specific issues // ❌ Using any defeats type safety function processData(data: any) { // Avoid any return data.value; } // βœ… Use proper types interface DataPayload { value: string; } function processData(data: DataPayload) { return data.value; } // ❌ Not handling async errors async function fetchUser(id: string) { const response = await fetch(`/api/users/${id}`); return response.json(); // What if network fails? } // βœ… Handle errors properly async function fetchUser(id: string): Promise<User> { try { const response = await fetch(`/api/users/${id}`); if (!response.ok) { throw new Error(`HTTP ${response.status}`); } return await response.json(); } catch (error) { console.error('Failed to fetch user:', error); throw error; } } // ❌ Mutation of props function UserProfile({ user }: Props) { user.lastViewed = new Date(); // Mutating prop! return <div>{user.name}</div>; } // βœ… Don't mutate props function UserProfile({ user, onView }: Props) { useEffect(() => { onView(user.id); // Notify parent to update }, [user.id]); return <div>{user.name}</div>; } ``` ## Advanced Review Patterns ### Pattern 1: Architectural Review ```markdown When reviewing significant changes: 1. **Design Document First** - For large features, request design doc before code - Review design with team before implementation - Agree on approach to avoid rework 2. **Review in Stages** - First PR: Core abstractions and interfaces - Second PR: Implementation - Third PR: Integration and tests - Easier to review, faster to iterate 3. **Consider Alternatives** - "Have we considered using [pattern/library]?" - "What's the tradeoff vs. the simpler approach?" - "How will this evolve as requirements change?" ``` ### Pattern 2: Test Quality Review ```typescript // ❌ Poor test: Implementation detail testing test('increments counter variable', () => { const component = render(<Counter />); const button = component.getByRole('button'); fireEvent.click(button); expect(component.state.counter).toBe(1); // Testing internal state }); // βœ… Good test: Behavior testing test('displays incremented count when clicked', () => { render(<Counter />); const button = screen.getByRole('button', { name: /increment/i }); fireEvent.click(button); expect(screen.getByText('Count: 1')).toBeInTheDocument(); }); // Review questions for tests: // - Do tests describe behavior, not implementation? // - Are test names clear and descriptive? // - Do tests cover edge cases? // - Are tests independent (no shared state)? // - Can tests run in any order? ``` ### Pattern 3: Security Review ```markdown ## Security Review Checklist ### Authentication & Authorization - [ ] Is authentication required where needed? - [ ] Are authorization checks before every action? - [ ] Is JWT validation proper (signature, expiry)? - [ ] Are API keys/secrets properly secured? ### Input Validation - [ ] All user inputs validated? - [ ] File uploads restricted (size, type)? - [ ] SQL queries parameterized? - [ ] XSS protection (escape output)? ### Data Protection - [ ] Passwords hashed (bcrypt/argon2)? - [ ] Sensitive data encrypted at rest? - [ ] HTTPS enforced for sensitive data? - [ ] PII handled according to regulations? ### Common Vulnerabilities - [ ] No eval() or similar dynamic execution? - [ ] No hardcoded secrets? - [ ] CSRF protection for state-changing operations? - [ ] Rate limiting on public endpoints? ``` ## Giving Difficult Feedback ### Pattern: The Sandwich Method (Modified) ```markdown Traditional: Praise + Criticism + Praise (feels fake) Better: Context + Specific Issue + Helpful Solution Example: "I noticed the payment processing logic is inline in the controller. This makes it harder to test and reuse. [Specific Issue] The calculateTotal() function mixes tax calculation, discount logic, and database queries, making it difficult to unit test and reason about. [Helpful Solution] Could we extract this into a PaymentService class? That would make it testable and reusable. I can pair with you on this if helpful." ``` ### Handling Disagreements ```markdown When author disagrees with your feedback: 1. **Seek to Understand** "Help me understand your approach. What led you to choose this pattern?" 2. **Acknowledge Valid Points** "That's a good point about X. I hadn't considered that." 3. **Provide Data** "I'm concerned about performance. Can we add a benchmark to validate the approach?" 4. **Escalate if Needed** "Let's get [architect/senior dev] to weigh in on this." 5. **Know When to Let Go** If it's working and not a critical issue, approve it. Perfection is the enemy of progress. ``` ## Best Practices 1. **Review Promptly**: Within 24 hours, ideally same day 2. **Limit PR Size**: 200-400 lines max for effective review 3. **Review in Time Blocks**: 60 minutes max, take breaks 4. **Use Review Tools**: GitHub, GitLab, or dedicated tools 5. **Automate What You Can**: Linters, formatters, security scans 6. **Build Rapport**: Emoji, praise, and empathy matter 7. **Be Available**: Offer to pair on complex issues 8. **Learn from Others**: Review others' review comments ## Common Pitfalls - **Perfectionism**: Blocking PRs for minor style preferences - **Scope Creep**: "While you're at it, can you also..." - **Inconsistency**: Different standards for different people - **Delayed Reviews**: Letting PRs sit for days - **Ghosting**: Requesting changes then disappearing - **Rubber Stamping**: Approving without actually reviewing - **Bike Shedding**: Debating trivial details extensively ## Templates ### PR Review Comment Template ```markdown ## Summary [Brief overview of what was reviewed] ## Strengths - [What was done well] - [Good patterns or approaches] ## Required Changes πŸ”΄ [Blocking issue 1] πŸ”΄ [Blocking issue 2] ## Suggestions πŸ’‘ [Improvement 1] πŸ’‘ [Improvement 2] ## Questions ❓ [Clarification needed on X] ❓ [Alternative approach consideration] ## Verdict βœ… Approve after addressing required changes ``` ## Resources - **references/code-review-best-practices.md**: Comprehensive review guidelines - **references/common-bugs-checklist.md**: Language-specific bugs to watch for - **references/security-review-guide.md**: Security-focused review checklist - **assets/pr-review-template.md**: Standard review comment template - **assets/review-checklist.md**: Quick reference checklist - **scripts/pr-analyzer.py**: Analyze PR complexity and suggest reviewers
πŸ‘0
πŸ‘οΈ3
πŸ€– Auto-discovered