Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

20 NODES
πŸ“textβ€’3 hours ago

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

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

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

Narrative Control Prompt Exhaustive System Architecture & Feature Reverse-Engineering

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

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

workflow-orchestration-patterns

Design durable workflows with Temporal for distributed systems.

coding
⭐1
# Workflow Orchestration Patterns Master workflow orchestration architecture with Temporal, covering fundamental design decisions, resilience patterns, and best practices for building reliable distributed systems. ## When to Use Workflow Orchestration ### Ideal Use Cases (Source: docs.temporal.io) - **Multi-step processes** spanning machines/services/databases - **Distributed transactions** requiring all-or-nothing semantics - **Long-running workflows** (hours to years) with automatic state persistence - **Failure recovery** that must resume from last successful step - **Business processes**: bookings, orders, campaigns, approvals - **Entity lifecycle management**: inventory tracking, account management, cart workflows - **Infrastructure automation**: CI/CD pipelines, provisioning, deployments - **Human-in-the-loop** systems requiring timeouts and escalations ### When NOT to Use - Simple CRUD operations (use direct API calls) - Pure data processing pipelines (use Airflow, batch processing) - Stateless request/response (use standard APIs) - Real-time streaming (use Kafka, event processors) ## Critical Design Decision: Workflows vs Activities **The Fundamental Rule** (Source: temporal.io/blog/workflow-engine-principles): - **Workflows** = Orchestration logic and decision-making - **Activities** = External interactions (APIs, databases, network calls) ### Workflows (Orchestration) **Characteristics:** - Contain business logic and coordination - **MUST be deterministic** (same inputs β†’ same outputs) - **Cannot** perform direct external calls - State automatically preserved across failures - Can run for years despite infrastructure failures **Example workflow tasks:** - Decide which steps to execute - Handle compensation logic - Manage timeouts and retries - Coordinate child workflows ### Activities (External Interactions) **Characteristics:** - Handle all external system interactions - Can be non-deterministic (API calls, DB writes) - Include built-in timeouts and retry logic - **Must be idempotent** (calling N times = calling once) - Short-lived (seconds to minutes typically) **Example activity tasks:** - Call payment gateway API - Write to database - Send emails or notifications - Query external services ### Design Decision Framework ``` Does it touch external systems? β†’ Activity Is it orchestration/decision logic? β†’ Workflow ``` ## Core Workflow Patterns ### 1. Saga Pattern with Compensation **Purpose**: Implement distributed transactions with rollback capability **Pattern** (Source: temporal.io/blog/compensating-actions-part-of-a-complete-breakfast-with-sagas): ``` For each step: 1. Register compensation BEFORE executing 2. Execute the step (via activity) 3. On failure, run all compensations in reverse order (LIFO) ``` **Example: Payment Workflow** 1. Reserve inventory (compensation: release inventory) 2. Charge payment (compensation: refund payment) 3. Fulfill order (compensation: cancel fulfillment) **Critical Requirements:** - Compensations must be idempotent - Register compensation BEFORE executing step - Run compensations in reverse order - Handle partial failures gracefully ### 2. Entity Workflows (Actor Model) **Purpose**: Long-lived workflow representing single entity instance **Pattern** (Source: docs.temporal.io/evaluate/use-cases-design-patterns): - One workflow execution = one entity (cart, account, inventory item) - Workflow persists for entity lifetime - Receives signals for state changes - Supports queries for current state **Example Use Cases:** - Shopping cart (add items, checkout, expiration) - Bank account (deposits, withdrawals, balance checks) - Product inventory (stock updates, reservations) **Benefits:** - Encapsulates entity behavior - Guarantees consistency per entity - Natural event sourcing ### 3. Fan-Out/Fan-In (Parallel Execution) **Purpose**: Execute multiple tasks in parallel, aggregate results **Pattern:** - Spawn child workflows or parallel activities - Wait for all to complete - Aggregate results - Handle partial failures **Scaling Rule** (Source: temporal.io/blog/workflow-engine-principles): - Don't scale individual workflows - For 1M tasks: spawn 1K child workflows Γ— 1K tasks each - Keep each workflow bounded ### 4. Async Callback Pattern **Purpose**: Wait for external event or human approval **Pattern:** - Workflow sends request and waits for signal - External system processes asynchronously - Sends signal to resume workflow - Workflow continues with response **Use Cases:** - Human approval workflows - Webhook callbacks - Long-running external processes ## State Management and Determinism ### Automatic State Preservation **How Temporal Works** (Source: docs.temporal.io/workflows): - Complete program state preserved automatically - Event History records every command and event - Seamless recovery from crashes - Applications restore pre-failure state ### Determinism Constraints **Workflows Execute as State Machines**: - Replay behavior must be consistent - Same inputs β†’ identical outputs every time **Prohibited in Workflows** (Source: docs.temporal.io/workflows): - ❌ Threading, locks, synchronization primitives - ❌ Random number generation (`random()`) - ❌ Global state or static variables - ❌ System time (`datetime.now()`) - ❌ Direct file I/O or network calls - ❌ Non-deterministic libraries **Allowed in Workflows**: - βœ… `workflow.now()` (deterministic time) - βœ… `workflow.random()` (deterministic random) - βœ… Pure functions and calculations - βœ… Calling activities (non-deterministic operations) ### Versioning Strategies **Challenge**: Changing workflow code while old executions still running **Solutions**: 1. **Versioning API**: Use `workflow.get_version()` for safe changes 2. **New Workflow Type**: Create new workflow, route new executions to it 3. **Backward Compatibility**: Ensure old events replay correctly ## Resilience and Error Handling ### Retry Policies **Default Behavior**: Temporal retries activities forever **Configure Retry**: - Initial retry interval - Backoff coefficient (exponential backoff) - Maximum interval (cap retry delay) - Maximum attempts (eventually fail) **Non-Retryable Errors**: - Invalid input (validation failures) - Business rule violations - Permanent failures (resource not found) ### Idempotency Requirements **Why Critical** (Source: docs.temporal.io/activities): - Activities may execute multiple times - Network failures trigger retries - Duplicate execution must be safe **Implementation Strategies**: - Idempotency keys (deduplication) - Check-then-act with unique constraints - Upsert operations instead of insert - Track processed request IDs ### Activity Heartbeats **Purpose**: Detect stalled long-running activities **Pattern**: - Activity sends periodic heartbeat - Includes progress information - Timeout if no heartbeat received - Enables progress-based retry ## Best Practices ### Workflow Design 1. **Keep workflows focused** - Single responsibility per workflow 2. **Small workflows** - Use child workflows for scalability 3. **Clear boundaries** - Workflow orchestrates, activities execute 4. **Test locally** - Use time-skipping test environment ### Activity Design 1. **Idempotent operations** - Safe to retry 2. **Short-lived** - Seconds to minutes, not hours 3. **Timeout configuration** - Always set timeouts 4. **Heartbeat for long tasks** - Report progress 5. **Error handling** - Distinguish retryable vs non-retryable ### Common Pitfalls **Workflow Violations**: - Using `datetime.now()` instead of `workflow.now()` - Threading or async operations in workflow code - Calling external APIs directly from workflow - Non-deterministic logic in workflows **Activity Mistakes**: - Non-idempotent operations (can't handle retries) - Missing timeouts (activities run forever) - No error classification (retry validation errors) - Ignoring payload limits (2MB per argument) ### Operational Considerations **Monitoring**: - Workflow execution duration - Activity failure rates - Retry attempts and backoff - Pending workflow counts **Scalability**: - Horizontal scaling with workers - Task queue partitioning - Child workflow decomposition - Activity batching when appropriate ## Additional Resources **Official Documentation**: - Temporal Core Concepts: docs.temporal.io/workflows - Workflow Patterns: docs.temporal.io/evaluate/use-cases-design-patterns - Best Practices: docs.temporal.io/develop/best-practices - Saga Pattern: temporal.io/blog/saga-pattern-made-easy **Key Principles**: 1. Workflows = orchestration, Activities = external calls 2. Determinism is non-negotiable for workflows 3. Idempotency is critical for activities 4. State preservation is automatic 5. Design for failure and recovery
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

workflow-patterns

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

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

bazel-build-optimization

Optimize Bazel builds for large-scale monorepos. Use when

coding
⭐1
# Bazel Build Optimization Production patterns for Bazel in large-scale monorepos. ## When to Use This Skill - Setting up Bazel for monorepos - Configuring remote caching/execution - Optimizing build times - Writing custom Bazel rules - Debugging build issues - Migrating to Bazel ## Core Concepts ### 1. Bazel Architecture ``` workspace/ β”œβ”€β”€ WORKSPACE.bazel # External dependencies β”œβ”€β”€ .bazelrc # Build configurations β”œβ”€β”€ .bazelversion # Bazel version β”œβ”€β”€ BUILD.bazel # Root build file β”œβ”€β”€ apps/ β”‚ └── web/ β”‚ └── BUILD.bazel β”œβ”€β”€ libs/ β”‚ └── utils/ β”‚ └── BUILD.bazel └── tools/ └── bazel/ └── rules/ ``` ### 2. Key Concepts | Concept | Description | | ----------- | -------------------------------------- | | **Target** | Buildable unit (library, binary, test) | | **Package** | Directory with BUILD file | | **Label** | Target identifier `//path/to:target` | | **Rule** | Defines how to build a target | | **Aspect** | Cross-cutting build behavior | ## Templates ### Template 1: WORKSPACE Configuration ```python # WORKSPACE.bazel workspace(name = "myproject") load("@bazel_tools//tools/build_defs/repo:http.bzl", "http_archive") # Rules for JavaScript/TypeScript http_archive( name = "aspect_rules_js", sha256 = "...", strip_prefix = "rules_js-1.34.0", url = "https://github.com/aspect-build/rules_js/releases/download/v1.34.0/rules_js-v1.34.0.tar.gz", ) load("@aspect_rules_js//js:repositories.bzl", "rules_js_dependencies") rules_js_dependencies() load("@rules_nodejs//nodejs:repositories.bzl", "nodejs_register_toolchains") nodejs_register_toolchains( name = "nodejs", node_version = "20.9.0", ) load("@aspect_rules_js//npm:repositories.bzl", "npm_translate_lock") npm_translate_lock( name = "npm", pnpm_lock = "//:pnpm-lock.yaml", verify_node_modules_ignored = "//:.bazelignore", ) load("@npm//:repositories.bzl", "npm_repositories") npm_repositories() # Rules for Python http_archive( name = "rules_python", sha256 = "...", strip_prefix = "rules_python-0.27.0", url = "https://github.com/bazelbuild/rules_python/releases/download/0.27.0/rules_python-0.27.0.tar.gz", ) load("@rules_python//python:repositories.bzl", "py_repositories") py_repositories() ``` ### Template 2: .bazelrc Configuration ```bash # .bazelrc # Build settings build --enable_platform_specific_config build --incompatible_enable_cc_toolchain_resolution build --experimental_strict_conflict_checks # Performance build --jobs=auto build --local_cpu_resources=HOST_CPUS*.75 build --local_ram_resources=HOST_RAM*.75 # Caching build --disk_cache=~/.cache/bazel-disk build --repository_cache=~/.cache/bazel-repo # Remote caching (optional) build:remote-cache --remote_cache=grpcs://cache.example.com build:remote-cache --remote_upload_local_results=true build:remote-cache --remote_timeout=3600 # Remote execution (optional) build:remote-exec --remote_executor=grpcs://remote.example.com build:remote-exec --remote_instance_name=projects/myproject/instances/default build:remote-exec --jobs=500 # Platform configurations build:linux --platforms=//platforms:linux_x86_64 build:macos --platforms=//platforms:macos_arm64 # CI configuration build:ci --config=remote-cache build:ci --build_metadata=ROLE=CI build:ci --bes_results_url=https://results.example.com/invocation/ build:ci --bes_backend=grpcs://bes.example.com # Test settings test --test_output=errors test --test_summary=detailed # Coverage coverage --combined_report=lcov coverage --instrumentation_filter="//..." # Convenience aliases build:opt --compilation_mode=opt build:dbg --compilation_mode=dbg # Import user settings try-import %workspace%/user.bazelrc ``` ### Template 3: TypeScript Library BUILD ```python # libs/utils/BUILD.bazel load("@aspect_rules_ts//ts:defs.bzl", "ts_project") load("@aspect_rules_js//js:defs.bzl", "js_library") load("@npm//:defs.bzl", "npm_link_all_packages") npm_link_all_packages(name = "node_modules") ts_project( name = "utils_ts", srcs = glob(["src/**/*.ts"]), declaration = True, source_map = True, tsconfig = "//:tsconfig.json", deps = [ ":node_modules/@types/node", ], ) js_library( name = "utils", srcs = [":utils_ts"], visibility = ["//visibility:public"], ) # Tests load("@aspect_rules_jest//jest:defs.bzl", "jest_test") jest_test( name = "utils_test", config = "//:jest.config.js", data = [ ":utils", "//:node_modules/jest", ], node_modules = "//:node_modules", ) ``` ### Template 4: Python Library BUILD ```python # libs/ml/BUILD.bazel load("@rules_python//python:defs.bzl", "py_library", "py_test", "py_binary") load("@pip//:requirements.bzl", "requirement") py_library( name = "ml", srcs = glob(["src/**/*.py"]), deps = [ requirement("numpy"), requirement("pandas"), requirement("scikit-learn"), "//libs/utils:utils_py", ], visibility = ["//visibility:public"], ) py_test( name = "ml_test", srcs = glob(["tests/**/*.py"]), deps = [ ":ml", requirement("pytest"), ], size = "medium", timeout = "moderate", ) py_binary( name = "train", srcs = ["train.py"], deps = [":ml"], data = ["//data:training_data"], ) ``` ### Template 5: Custom Rule for Docker ```python # tools/bazel/rules/docker.bzl def _docker_image_impl(ctx): dockerfile = ctx.file.dockerfile base_image = ctx.attr.base_image layers = ctx.files.layers # Build the image output = ctx.actions.declare_file(ctx.attr.name + ".tar") args = ctx.actions.args() args.add("--dockerfile", dockerfile) args.add("--output", output) args.add("--base", base_image) args.add_all("--layer", layers) ctx.actions.run( inputs = [dockerfile] + layers, outputs = [output], executable = ctx.executable._builder, arguments = [args], mnemonic = "DockerBuild", progress_message = "Building Docker image %s" % ctx.label, ) return [DefaultInfo(files = depset([output]))] docker_image = rule( implementation = _docker_image_impl, attrs = { "dockerfile": attr.label( allow_single_file = [".dockerfile", "Dockerfile"], mandatory = True, ), "base_image": attr.string(mandatory = True), "layers": attr.label_list(allow_files = True), "_builder": attr.label( default = "//tools/docker:builder", executable = True, cfg = "exec", ), }, ) ``` ### Template 6: Query and Dependency Analysis ```bash # Find all dependencies of a target bazel query "deps(//apps/web:web)" # Find reverse dependencies (what depends on this) bazel query "rdeps(//..., //libs/utils:utils)" # Find all targets in a package bazel query "//libs/..." # Find changed targets since commit bazel query "rdeps(//..., set($(git diff --name-only HEAD~1 | sed 's/.*/"&"/' | tr '\n' ' ')))" # Generate dependency graph bazel query "deps(//apps/web:web)" --output=graph | dot -Tpng > deps.png # Find all test targets bazel query "kind('.*_test', //...)" # Find targets with specific tag bazel query "attr(tags, 'integration', //...)" # Compute build graph size bazel query "deps(//...)" --output=package | wc -l ``` ### Template 7: Remote Execution Setup ```python # platforms/BUILD.bazel platform( name = "linux_x86_64", constraint_values = [ "@platforms//os:linux", "@platforms//cpu:x86_64", ], exec_properties = { "container-image": "docker://gcr.io/myproject/bazel-worker:latest", "OSFamily": "Linux", }, ) platform( name = "remote_linux", parents = [":linux_x86_64"], exec_properties = { "Pool": "default", "dockerNetwork": "standard", }, ) # toolchains/BUILD.bazel toolchain( name = "cc_toolchain_linux", exec_compatible_with = [ "@platforms//os:linux", "@platforms//cpu:x86_64", ], target_compatible_with = [ "@platforms//os:linux", "@platforms//cpu:x86_64", ], toolchain = "@remotejdk11_linux//:jdk", toolchain_type = "@bazel_tools//tools/jdk:runtime_toolchain_type", ) ``` ## Performance Optimization ```bash # Profile build bazel build //... --profile=profile.json bazel analyze-profile profile.json # Identify slow actions bazel build //... --execution_log_json_file=exec_log.json # Memory profiling bazel build //... --memory_profile=memory.json # Skip analysis cache bazel build //... --notrack_incremental_state ``` ## Best Practices ### Do's - **Use fine-grained targets** - Better caching - **Pin dependencies** - Reproducible builds - **Enable remote caching** - Share build artifacts - **Use visibility wisely** - Enforce architecture - **Write BUILD files per directory** - Standard convention ### Don'ts - **Don't use glob for deps** - Explicit is better - **Don't commit bazel-\* dirs** - Add to .gitignore - **Don't skip WORKSPACE setup** - Foundation of build - **Don't ignore build warnings** - Technical debt ## Resources - [Bazel Documentation](https://bazel.build/docs) - [Bazel Remote Execution](https://bazel.build/docs/remote-execution) - [rules_js](https://github.com/aspect-build/rules_js)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

architecture-decision-records

Write and maintain Architecture Decision Records (ADRs) following

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

embedding-strategies

Select and optimize embedding models for semantic search and RAG

coding
⭐1
# Embedding Strategies Guide to selecting and optimizing embedding models for vector search applications. ## When to Use This Skill - Choosing embedding models for RAG - Optimizing chunking strategies - Fine-tuning embeddings for domains - Comparing embedding model performance - Reducing embedding dimensions - Handling multilingual content ## Core Concepts ### 1. Embedding Model Comparison (2026) | Model | Dimensions | Max Tokens | Best For | | -------------------------- | ---------- | ---------- | ----------------------------------- | | **voyage-3-large** | 1024 | 32000 | Claude apps (Anthropic recommended) | | **voyage-3** | 1024 | 32000 | Claude apps, cost-effective | | **voyage-code-3** | 1024 | 32000 | Code search | | **voyage-finance-2** | 1024 | 32000 | Financial documents | | **voyage-law-2** | 1024 | 32000 | Legal documents | | **text-embedding-3-large** | 3072 | 8191 | OpenAI apps, high accuracy | | **text-embedding-3-small** | 1536 | 8191 | OpenAI apps, cost-effective | | **bge-large-en-v1.5** | 1024 | 512 | Open source, local deployment | | **all-MiniLM-L6-v2** | 384 | 256 | Fast, lightweight | | **multilingual-e5-large** | 1024 | 512 | Multi-language | ### 2. Embedding Pipeline ``` Document β†’ Chunking β†’ Preprocessing β†’ Embedding Model β†’ Vector ↓ [Overlap, Size] [Clean, Normalize] [API/Local] ``` ## Templates ### Template 1: Voyage AI Embeddings (Recommended for Claude) ```python from langchain_voyageai import VoyageAIEmbeddings from typing import List import os # Initialize Voyage AI embeddings (recommended by Anthropic for Claude) embeddings = VoyageAIEmbeddings( model="voyage-3-large", voyage_api_key=os.environ.get("VOYAGE_API_KEY") ) def get_embeddings(texts: List[str]) -> List[List[float]]: """Get embeddings from Voyage AI.""" return embeddings.embed_documents(texts) def get_query_embedding(query: str) -> List[float]: """Get single query embedding.""" return embeddings.embed_query(query) # Specialized models for domains code_embeddings = VoyageAIEmbeddings(model="voyage-code-3") finance_embeddings = VoyageAIEmbeddings(model="voyage-finance-2") legal_embeddings = VoyageAIEmbeddings(model="voyage-law-2") ``` ### Template 2: OpenAI Embeddings ```python from openai import OpenAI from typing import List import numpy as np client = OpenAI() def get_embeddings( texts: List[str], model: str = "text-embedding-3-small", dimensions: int = None ) -> List[List[float]]: """Get embeddings from OpenAI with optional dimension reduction.""" # Handle batching for large lists batch_size = 100 all_embeddings = [] for i in range(0, len(texts), batch_size): batch = texts[i:i + batch_size] kwargs = {"input": batch, "model": model} if dimensions: # Matryoshka dimensionality reduction kwargs["dimensions"] = dimensions response = client.embeddings.create(**kwargs) embeddings = [item.embedding for item in response.data] all_embeddings.extend(embeddings) return all_embeddings def get_embedding(text: str, **kwargs) -> List[float]: """Get single embedding.""" return get_embeddings([text], **kwargs)[0] # Dimension reduction with Matryoshka embeddings def get_reduced_embedding(text: str, dimensions: int = 512) -> List[float]: """Get embedding with reduced dimensions (Matryoshka).""" return get_embedding( text, model="text-embedding-3-small", dimensions=dimensions ) ``` ### Template 3: Local Embeddings with Sentence Transformers ```python from sentence_transformers import SentenceTransformer from typing import List, Optional import numpy as np class LocalEmbedder: """Local embedding with sentence-transformers.""" def __init__( self, model_name: str = "BAAI/bge-large-en-v1.5", device: str = "cuda" ): self.model = SentenceTransformer(model_name, device=device) self.model_name = model_name def embed( self, texts: List[str], normalize: bool = True, show_progress: bool = False ) -> np.ndarray: """Embed texts with optional normalization.""" embeddings = self.model.encode( texts, normalize_embeddings=normalize, show_progress_bar=show_progress, convert_to_numpy=True ) return embeddings def embed_query(self, query: str) -> np.ndarray: """Embed a query with appropriate prefix for retrieval models.""" # BGE and similar models benefit from query prefix if "bge" in self.model_name.lower(): query = f"Represent this sentence for searching relevant passages: {query}" return self.embed([query])[0] def embed_documents(self, documents: List[str]) -> np.ndarray: """Embed documents for indexing.""" return self.embed(documents) # E5 model with instructions class E5Embedder: def __init__(self, model_name: str = "intfloat/multilingual-e5-large"): self.model = SentenceTransformer(model_name) def embed_query(self, query: str) -> np.ndarray: """E5 requires 'query:' prefix for queries.""" return self.model.encode(f"query: {query}") def embed_document(self, document: str) -> np.ndarray: """E5 requires 'passage:' prefix for documents.""" return self.model.encode(f"passage: {document}") ``` ### Template 4: Chunking Strategies ```python from typing import List, Tuple import re def chunk_by_tokens( text: str, chunk_size: int = 512, chunk_overlap: int = 50, tokenizer=None ) -> List[str]: """Chunk text by token count.""" import tiktoken tokenizer = tokenizer or tiktoken.get_encoding("cl100k_base") tokens = tokenizer.encode(text) chunks = [] start = 0 while start < len(tokens): end = start + chunk_size chunk_tokens = tokens[start:end] chunk_text = tokenizer.decode(chunk_tokens) chunks.append(chunk_text) start = end - chunk_overlap return chunks def chunk_by_sentences( text: str, max_chunk_size: int = 1000, min_chunk_size: int = 100 ) -> List[str]: """Chunk text by sentences, respecting size limits.""" import nltk sentences = nltk.sent_tokenize(text) chunks = [] current_chunk = [] current_size = 0 for sentence in sentences: sentence_size = len(sentence) if current_size + sentence_size > max_chunk_size and current_chunk: chunks.append(" ".join(current_chunk)) current_chunk = [] current_size = 0 current_chunk.append(sentence) current_size += sentence_size if current_chunk: chunks.append(" ".join(current_chunk)) return chunks def chunk_by_semantic_sections( text: str, headers_pattern: str = r'^#{1,3}\s+.+$' ) -> List[Tuple[str, str]]: """Chunk markdown by headers, preserving hierarchy.""" lines = text.split('\n') chunks = [] current_header = "" current_content = [] for line in lines: if re.match(headers_pattern, line, re.MULTILINE): if current_content: chunks.append((current_header, '\n'.join(current_content))) current_header = line current_content = [] else: current_content.append(line) if current_content: chunks.append((current_header, '\n'.join(current_content))) return chunks def recursive_character_splitter( text: str, chunk_size: int = 1000, chunk_overlap: int = 200, separators: List[str] = None ) -> List[str]: """LangChain-style recursive splitter.""" separators = separators or ["\n\n", "\n", ". ", " ", ""] def split_text(text: str, separators: List[str]) -> List[str]: if not text: return [] separator = separators[0] remaining_separators = separators[1:] if separator == "": # Character-level split return [text[i:i+chunk_size] for i in range(0, len(text), chunk_size - chunk_overlap)] splits = text.split(separator) chunks = [] current_chunk = [] current_length = 0 for split in splits: split_length = len(split) + len(separator) if current_length + split_length > chunk_size and current_chunk: chunk_text = separator.join(current_chunk) # Recursively split if still too large if len(chunk_text) > chunk_size and remaining_separators: chunks.extend(split_text(chunk_text, remaining_separators)) else: chunks.append(chunk_text) # Start new chunk with overlap overlap_splits = [] overlap_length = 0 for s in reversed(current_chunk): if overlap_length + len(s) <= chunk_overlap: overlap_splits.insert(0, s) overlap_length += len(s) else: break current_chunk = overlap_splits current_length = overlap_length current_chunk.append(split) current_length += split_length if current_chunk: chunks.append(separator.join(current_chunk)) return chunks return split_text(text, separators) ``` ### Template 5: Domain-Specific Embedding Pipeline ```python import re from typing import List, Optional from dataclasses import dataclass @dataclass class EmbeddedDocument: id: str document_id: str chunk_index: int text: str embedding: List[float] metadata: dict class DomainEmbeddingPipeline: """Pipeline for domain-specific embeddings.""" def __init__( self, embedding_model: str = "voyage-3-large", chunk_size: int = 512, chunk_overlap: int = 50, preprocessing_fn=None ): self.embeddings = VoyageAIEmbeddings(model=embedding_model) self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap self.preprocess = preprocessing_fn or self._default_preprocess def _default_preprocess(self, text: str) -> str: """Default preprocessing.""" # Remove excessive whitespace text = re.sub(r'\s+', ' ', text) # Remove special characters (customize for your domain) text = re.sub(r'[^\w\s.,!?-]', '', text) return text.strip() async def process_documents( self, documents: List[dict], id_field: str = "id", content_field: str = "content", metadata_fields: Optional[List[str]] = None ) -> List[EmbeddedDocument]: """Process documents for vector storage.""" processed = [] for doc in documents: content = doc[content_field] doc_id = doc[id_field] # Preprocess cleaned = self.preprocess(content) # Chunk chunks = chunk_by_tokens( cleaned, self.chunk_size, self.chunk_overlap ) # Create embeddings embeddings = await self.embeddings.aembed_documents(chunks) # Create records for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)): metadata = {"document_id": doc_id, "chunk_index": i} # Add specified metadata fields if metadata_fields: for field in metadata_fields: if field in doc: metadata[field] = doc[field] processed.append(EmbeddedDocument( id=f"{doc_id}_chunk_{i}", document_id=doc_id, chunk_index=i, text=chunk, embedding=embedding, metadata=metadata )) return processed # Code-specific pipeline class CodeEmbeddingPipeline: """Specialized pipeline for code embeddings.""" def __init__(self): # Use Voyage's code-specific model self.embeddings = VoyageAIEmbeddings(model="voyage-code-3") def chunk_code(self, code: str, language: str) -> List[dict]: """Chunk code by functions/classes using tree-sitter.""" try: import tree_sitter_languages parser = tree_sitter_languages.get_parser(language) tree = parser.parse(bytes(code, "utf8")) chunks = [] # Extract function and class definitions self._extract_nodes(tree.root_node, code, chunks) return chunks except ImportError: # Fallback to simple chunking return [{"text": code, "type": "module"}] def _extract_nodes(self, node, source_code: str, chunks: list): """Recursively extract function/class definitions.""" if node.type in ['function_definition', 'class_definition', 'method_definition']: text = source_code[node.start_byte:node.end_byte] chunks.append({ "text": text, "type": node.type, "name": self._get_name(node), "start_line": node.start_point[0], "end_line": node.end_point[0] }) for child in node.children: self._extract_nodes(child, source_code, chunks) def _get_name(self, node) -> str: """Extract name from function/class node.""" for child in node.children: if child.type == 'identifier' or child.type == 'name': return child.text.decode('utf8') return "unknown" async def embed_with_context( self, chunk: str, context: str = "" ) -> List[float]: """Embed code with surrounding context.""" if context: combined = f"Context: {context}\n\nCode:\n{chunk}" else: combined = chunk return await self.embeddings.aembed_query(combined) ``` ### Template 6: Embedding Quality Evaluation ```python import numpy as np from typing import List, Dict def evaluate_retrieval_quality( queries: List[str], relevant_docs: List[List[str]], # List of relevant doc IDs per query retrieved_docs: List[List[str]], # List of retrieved doc IDs per query k: int = 10 ) -> Dict[str, float]: """Evaluate embedding quality for retrieval.""" def precision_at_k(relevant: set, retrieved: List[str], k: int) -> float: retrieved_k = retrieved[:k] relevant_retrieved = len(set(retrieved_k) & relevant) return relevant_retrieved / k if k > 0 else 0 def recall_at_k(relevant: set, retrieved: List[str], k: int) -> float: retrieved_k = retrieved[:k] relevant_retrieved = len(set(retrieved_k) & relevant) return relevant_retrieved / len(relevant) if relevant else 0 def mrr(relevant: set, retrieved: List[str]) -> float: for i, doc in enumerate(retrieved): if doc in relevant: return 1 / (i + 1) return 0 def ndcg_at_k(relevant: set, retrieved: List[str], k: int) -> float: dcg = sum( 1 / np.log2(i + 2) if doc in relevant else 0 for i, doc in enumerate(retrieved[:k]) ) ideal_dcg = sum(1 / np.log2(i + 2) for i in range(min(len(relevant), k))) return dcg / ideal_dcg if ideal_dcg > 0 else 0 metrics = { f"precision@{k}": [], f"recall@{k}": [], "mrr": [], f"ndcg@{k}": [] } for relevant, retrieved in zip(relevant_docs, retrieved_docs): relevant_set = set(relevant) metrics[f"precision@{k}"].append(precision_at_k(relevant_set, retrieved, k)) metrics[f"recall@{k}"].append(recall_at_k(relevant_set, retrieved, k)) metrics["mrr"].append(mrr(relevant_set, retrieved)) metrics[f"ndcg@{k}"].append(ndcg_at_k(relevant_set, retrieved, k)) return {name: np.mean(values) for name, values in metrics.items()} def compute_embedding_similarity( embeddings1: np.ndarray, embeddings2: np.ndarray, metric: str = "cosine" ) -> np.ndarray: """Compute similarity matrix between embedding sets.""" if metric == "cosine": # Normalize and compute dot product norm1 = embeddings1 / np.linalg.norm(embeddings1, axis=1, keepdims=True) norm2 = embeddings2 / np.linalg.norm(embeddings2, axis=1, keepdims=True) return norm1 @ norm2.T elif metric == "euclidean": from scipy.spatial.distance import cdist return -cdist(embeddings1, embeddings2, metric='euclidean') elif metric == "dot": return embeddings1 @ embeddings2.T else: raise ValueError(f"Unknown metric: {metric}") def compare_embedding_models( texts: List[str], models: Dict[str, callable], queries: List[str], relevant_indices: List[List[int]], k: int = 5 ) -> Dict[str, Dict[str, float]]: """Compare multiple embedding models on retrieval quality.""" results = {} for model_name, embed_fn in models.items(): # Embed all texts doc_embeddings = np.array(embed_fn(texts)) retrieved_per_query = [] for query in queries: query_embedding = np.array(embed_fn([query])[0]) # Compute similarities similarities = compute_embedding_similarity( query_embedding.reshape(1, -1), doc_embeddings, metric="cosine" )[0] # Get top-k indices top_k_indices = np.argsort(similarities)[::-1][:k] retrieved_per_query.append([str(i) for i in top_k_indices]) # Convert relevant indices to string IDs relevant_docs = [[str(i) for i in indices] for indices in relevant_indices] results[model_name] = evaluate_retrieval_quality( queries, relevant_docs, retrieved_per_query, k ) return results ``` ## Best Practices ### Do's - **Match model to use case**: Code vs prose vs multilingual - **Chunk thoughtfully**: Preserve semantic boundaries - **Normalize embeddings**: For cosine similarity search - **Batch requests**: More efficient than one-by-one - **Cache embeddings**: Avoid recomputing for static content - **Use Voyage AI for Claude apps**: Recommended by Anthropic ### Don'ts - **Don't ignore token limits**: Truncation loses information - **Don't mix embedding models**: Incompatible vector spaces - **Don't skip preprocessing**: Garbage in, garbage out - **Don't over-chunk**: Lose important context - **Don't forget metadata**: Essential for filtering and debugging ## Resources - [Voyage AI Documentation](https://docs.voyageai.com/) - [OpenAI Embeddings Guide](https://platform.openai.com/docs/guides/embeddings) - [Sentence Transformers](https://www.sbert.net/) - [MTEB Benchmar
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

hybrid-search-implementation

Combine vector and keyword search for improved retrieval. Use when

coding
⭐1
# Hybrid Search Implementation Patterns for combining vector similarity and keyword-based search. ## When to Use This Skill - Building RAG systems with improved recall - Combining semantic understanding with exact matching - Handling queries with specific terms (names, codes) - Improving search for domain-specific vocabulary - When pure vector search misses keyword matches ## Core Concepts ### 1. Hybrid Search Architecture ``` Query β†’ ┬─► Vector Search ──► Candidates ─┐ β”‚ β”‚ └─► Keyword Search ─► Candidates ─┴─► Fusion ─► Results ``` ### 2. Fusion Methods | Method | Description | Best For | | ----------------- | ------------------------ | --------------- | | **RRF** | Reciprocal Rank Fusion | General purpose | | **Linear** | Weighted sum of scores | Tunable balance | | **Cross-encoder** | Rerank with neural model | Highest quality | | **Cascade** | Filter then rerank | Efficiency | ## Templates ### Template 1: Reciprocal Rank Fusion ```python from typing import List, Dict, Tuple from collections import defaultdict def reciprocal_rank_fusion( result_lists: List[List[Tuple[str, float]]], k: int = 60, weights: List[float] = None ) -> List[Tuple[str, float]]: """ Combine multiple ranked lists using RRF. Args: result_lists: List of (doc_id, score) tuples per search method k: RRF constant (higher = more weight to lower ranks) weights: Optional weights per result list Returns: Fused ranking as (doc_id, score) tuples """ if weights is None: weights = [1.0] * len(result_lists) scores = defaultdict(float) for result_list, weight in zip(result_lists, weights): for rank, (doc_id, _) in enumerate(result_list): # RRF formula: 1 / (k + rank) scores[doc_id] += weight * (1.0 / (k + rank + 1)) # Sort by fused score return sorted(scores.items(), key=lambda x: x[1], reverse=True) def linear_combination( vector_results: List[Tuple[str, float]], keyword_results: List[Tuple[str, float]], alpha: float = 0.5 ) -> List[Tuple[str, float]]: """ Combine results with linear interpolation. Args: vector_results: (doc_id, similarity_score) from vector search keyword_results: (doc_id, bm25_score) from keyword search alpha: Weight for vector search (1-alpha for keyword) """ # Normalize scores to [0, 1] def normalize(results): if not results: return {} scores = [s for _, s in results] min_s, max_s = min(scores), max(scores) range_s = max_s - min_s if max_s != min_s else 1 return {doc_id: (score - min_s) / range_s for doc_id, score in results} vector_scores = normalize(vector_results) keyword_scores = normalize(keyword_results) # Combine all_docs = set(vector_scores.keys()) | set(keyword_scores.keys()) combined = {} for doc_id in all_docs: v_score = vector_scores.get(doc_id, 0) k_score = keyword_scores.get(doc_id, 0) combined[doc_id] = alpha * v_score + (1 - alpha) * k_score return sorted(combined.items(), key=lambda x: x[1], reverse=True) ``` ### Template 2: PostgreSQL Hybrid Search ```python import asyncpg from typing import List, Dict, Optional import numpy as np class PostgresHybridSearch: """Hybrid search with pgvector and full-text search.""" def __init__(self, pool: asyncpg.Pool): self.pool = pool async def setup_schema(self): """Create tables and indexes.""" async with self.pool.acquire() as conn: await conn.execute(""" CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE IF NOT EXISTS documents ( id TEXT PRIMARY KEY, content TEXT NOT NULL, embedding vector(1536), metadata JSONB DEFAULT '{}', ts_content tsvector GENERATED ALWAYS AS ( to_tsvector('english', content) ) STORED ); -- Vector index (HNSW) CREATE INDEX IF NOT EXISTS documents_embedding_idx ON documents USING hnsw (embedding vector_cosine_ops); -- Full-text index (GIN) CREATE INDEX IF NOT EXISTS documents_fts_idx ON documents USING gin (ts_content); """) async def hybrid_search( self, query: str, query_embedding: List[float], limit: int = 10, vector_weight: float = 0.5, filter_metadata: Optional[Dict] = None ) -> List[Dict]: """ Perform hybrid search combining vector and full-text. Uses RRF fusion for combining results. """ async with self.pool.acquire() as conn: # Build filter clause where_clause = "1=1" params = [query_embedding, query, limit * 3] if filter_metadata: for key, value in filter_metadata.items(): params.append(value) where_clause += f" AND metadata->>'{key}' = ${len(params)}" results = await conn.fetch(f""" WITH vector_search AS ( SELECT id, content, metadata, ROW_NUMBER() OVER (ORDER BY embedding <=> $1::vector) as vector_rank, 1 - (embedding <=> $1::vector) as vector_score FROM documents WHERE {where_clause} ORDER BY embedding <=> $1::vector LIMIT $3 ), keyword_search AS ( SELECT id, content, metadata, ROW_NUMBER() OVER (ORDER BY ts_rank(ts_content, websearch_to_tsquery('english', $2)) DESC) as keyword_rank, ts_rank(ts_content, websearch_to_tsquery('english', $2)) as keyword_score FROM documents WHERE ts_content @@ websearch_to_tsquery('english', $2) AND {where_clause} ORDER BY ts_rank(ts_content, websearch_to_tsquery('english', $2)) DESC LIMIT $3 ) SELECT COALESCE(v.id, k.id) as id, COALESCE(v.content, k.content) as content, COALESCE(v.metadata, k.metadata) as metadata, v.vector_score, k.keyword_score, -- RRF fusion COALESCE(1.0 / (60 + v.vector_rank), 0) * $4::float + COALESCE(1.0 / (60 + k.keyword_rank), 0) * (1 - $4::float) as rrf_score FROM vector_search v FULL OUTER JOIN keyword_search k ON v.id = k.id ORDER BY rrf_score DESC LIMIT $3 / 3 """, *params, vector_weight) return [dict(row) for row in results] async def search_with_rerank( self, query: str, query_embedding: List[float], limit: int = 10, rerank_candidates: int = 50 ) -> List[Dict]: """Hybrid search with cross-encoder reranking.""" from sentence_transformers import CrossEncoder # Get candidates candidates = await self.hybrid_search( query, query_embedding, limit=rerank_candidates ) if not candidates: return [] # Rerank with cross-encoder model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') pairs = [(query, c["content"]) for c in candidates] scores = model.predict(pairs) for candidate, score in zip(candidates, scores): candidate["rerank_score"] = float(score) # Sort by rerank score and return top results reranked = sorted(candidates, key=lambda x: x["rerank_score"], reverse=True) return reranked[:limit] ``` ### Template 3: Elasticsearch Hybrid Search ```python from elasticsearch import Elasticsearch from typing import List, Dict, Optional class ElasticsearchHybridSearch: """Hybrid search with Elasticsearch and dense vectors.""" def __init__( self, es_client: Elasticsearch, index_name: str = "documents" ): self.es = es_client self.index_name = index_name def create_index(self, vector_dims: int = 1536): """Create index with dense vector and text fields.""" mapping = { "mappings": { "properties": { "content": { "type": "text", "analyzer": "english" }, "embedding": { "type": "dense_vector", "dims": vector_dims, "index": True, "similarity": "cosine" }, "metadata": { "type": "object", "enabled": True } } } } self.es.indices.create(index=self.index_name, body=mapping, ignore=400) def hybrid_search( self, query: str, query_embedding: List[float], limit: int = 10, boost_vector: float = 1.0, boost_text: float = 1.0, filter: Optional[Dict] = None ) -> List[Dict]: """ Hybrid search using Elasticsearch's built-in capabilities. """ # Build the hybrid query search_body = { "size": limit, "query": { "bool": { "should": [ # Vector search (kNN) { "script_score": { "query": {"match_all": {}}, "script": { "source": f"cosineSimilarity(params.query_vector, 'embedding') * {boost_vector} + 1.0", "params": {"query_vector": query_embedding} } } }, # Text search (BM25) { "match": { "content": { "query": query, "boost": boost_text } } } ], "minimum_should_match": 1 } } } # Add filter if provided if filter: search_body["query"]["bool"]["filter"] = filter response = self.es.search(index=self.index_name, body=search_body) return [ { "id": hit["_id"], "content": hit["_source"]["content"], "metadata": hit["_source"].get("metadata", {}), "score": hit["_score"] } for hit in response["hits"]["hits"] ] def hybrid_search_rrf( self, query: str, query_embedding: List[float], limit: int = 10, window_size: int = 100 ) -> List[Dict]: """ Hybrid search using Elasticsearch 8.x RRF. """ search_body = { "size": limit, "sub_searches": [ { "query": { "match": { "content": query } } }, { "query": { "knn": { "field": "embedding", "query_vector": query_embedding, "k": window_size, "num_candidates": window_size * 2 } } } ], "rank": { "rrf": { "window_size": window_size, "rank_constant": 60 } } } response = self.es.search(index=self.index_name, body=search_body) return [ { "id": hit["_id"], "content": hit["_source"]["content"], "score": hit["_score"] } for hit in response["hits"]["hits"] ] ``` ### Template 4: Custom Hybrid RAG Pipeline ```python from typing import List, Dict, Optional, Callable from dataclasses import dataclass @dataclass class SearchResult: id: str content: str score: float source: str # "vector", "keyword", "hybrid" metadata: Dict = None class HybridRAGPipeline: """Complete hybrid search pipeline for RAG.""" def __init__( self, vector_store, keyword_store, embedder, reranker=None, fusion_method: str = "rrf", vector_weight: float = 0.5 ): self.vector_store = vector_store self.keyword_store = keyword_store self.embedder = embedder self.reranker = reranker self.fusion_method = fusion_method self.vector_weight = vector_weight async def search( self, query: str, top_k: int = 10, filter: Optional[Dict] = None, use_rerank: bool = True ) -> List[SearchResult]: """Execute hybrid search pipeline.""" # Step 1: Get query embedding query_embedding = self.embedder.embed(query) # Step 2: Execute parallel searches vector_results, keyword_results = await asyncio.gather( self._vector_search(query_embedding, top_k * 3, filter), self._keyword_search(query, top_k * 3, filter) ) # Step 3: Fuse results if self.fusion_method == "rrf": fused = self._rrf_fusion(vector_results, keyword_results) else: fused = self._linear_fusion(vector_results, keyword_results) # Step 4: Rerank if enabled if use_rerank and self.reranker: fused = await self._rerank(query, fused[:top_k * 2]) return fused[:top_k] async def _vector_search( self, embedding: List[float], limit: int, filter: Dict ) -> List[SearchResult]: results = await self.vector_store.search(embedding, limit, filter) return [ SearchResult( id=r["id"], content=r["content"], score=r["score"], source="vector", metadata=r.get("metadata") ) for r in results ] async def _keyword_search( self, query: str, limit: int, filter: Dict ) -> List[SearchResult]: results = await self.keyword_store.search(query, limit, filter) return [ SearchResult( id=r["id"], content=r["content"], score=r["score"], source="keyword", metadata=r.get("metadata") ) for r in results ] def _rrf_fusion( self, vector_results: List[SearchResult], keyword_results: List[SearchResult] ) -> List[SearchResult]: """Fuse with RRF.""" k = 60 scores = {} content_map = {} for rank, result in enumerate(vector_results): scores[result.id] = scores.get(result.id, 0) + 1 / (k + rank + 1) content_map[result.id] = result for rank, result in enumerate(keyword_results): scores[result.id] = scores.get(result.id, 0) + 1 / (k + rank + 1) if result.id not in content_map: content_map[result.id] = result sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True) return [ SearchResult( id=doc_id, content=content_map[doc_id].content, score=scores[doc_id], source="hybrid", metadata=content_map[doc_id].metadata ) for doc_id in sorted_ids ] async def _rerank( self, query: str, results: List[SearchResult] ) -> List[SearchResult]: """Rerank with cross-encoder.""" if not results: return results pairs = [(query, r.content) for r in results] scores = self.reranker.predict(pairs) for result, score in zip(results, scores): result.score = float(score) return sorted(results, key=lambda x: x.score, reverse=True) ``` ## Best Practices ### Do's - **Tune weights empirically** - Test on your data - **Use RRF for simplicity** - Works well without tuning - **Add reranking** - Significant quality improvement - **Log both scores** - Helps with debugging - **A/B test** - Measure real user impact ### Don'ts - **Don't assume one size fits all** - Different queries need different weights - **Don't skip keyword search** - Handles exact matches better - **Don't over-fetch** - Balance recall vs latency - **Don't ignore edge cases** - Empty results, single word queries ## Resources - [RRF Paper](https://plg.uwaterloo.ca/~gvcormac/cormacksigir09-rrf.pdf) - [Vespa Hybrid Search](https://blog.vespa.ai/improving-text-ranking-with-few-shot-prompting/) - [Cohere Rerank](https://docs.cohere.com/docs/reranking)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

rag-implementation

Build Retrieval-Augmented Generation (RAG) systems for LLM

coding
⭐1
# RAG Implementation Master Retrieval-Augmented Generation (RAG) to build LLM applications that provide accurate, grounded responses using external knowledge sources. ## When to Use This Skill - Building Q&A systems over proprietary documents - Creating chatbots with current, factual information - Implementing semantic search with natural language queries - Reducing hallucinations with grounded responses - Enabling LLMs to access domain-specific knowledge - Building documentation assistants - Creating research tools with source citation ## Core Components ### 1. Vector Databases **Purpose**: Store and retrieve document embeddings efficiently **Options:** - **Pinecone**: Managed, scalable, serverless - **Weaviate**: Open-source, hybrid search, GraphQL - **Milvus**: High performance, on-premise - **Chroma**: Lightweight, easy to use, local development - **Qdrant**: Fast, filtered search, Rust-based - **pgvector**: PostgreSQL extension, SQL integration ### 2. Embeddings **Purpose**: Convert text to numerical vectors for similarity search **Models (2026):** | Model | Dimensions | Best For | |-------|------------|----------| | **voyage-3-large** | 1024 | Claude apps (Anthropic recommended) | | **voyage-code-3** | 1024 | Code search | | **text-embedding-3-large** | 3072 | OpenAI apps, high accuracy | | **text-embedding-3-small** | 1536 | OpenAI apps, cost-effective | | **bge-large-en-v1.5** | 1024 | Open source, local deployment | | **multilingual-e5-large** | 1024 | Multi-language support | ### 3. Retrieval Strategies **Approaches:** - **Dense Retrieval**: Semantic similarity via embeddings - **Sparse Retrieval**: Keyword matching (BM25, TF-IDF) - **Hybrid Search**: Combine dense + sparse with weighted fusion - **Multi-Query**: Generate multiple query variations - **HyDE**: Generate hypothetical documents for better retrieval ### 4. Reranking **Purpose**: Improve retrieval quality by reordering results **Methods:** - **Cross-Encoders**: BERT-based reranking (ms-marco-MiniLM) - **Cohere Rerank**: API-based reranking - **Maximal Marginal Relevance (MMR)**: Diversity + relevance - **LLM-based**: Use LLM to score relevance ## Quick Start with LangGraph ```python from langgraph.graph import StateGraph, START, END from langchain_anthropic import ChatAnthropic from langchain_voyageai import VoyageAIEmbeddings from langchain_pinecone import PineconeVectorStore from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from langchain_text_splitters import RecursiveCharacterTextSplitter from typing import TypedDict, Annotated class RAGState(TypedDict): question: str context: list[Document] answer: str # Initialize components llm = ChatAnthropic(model="claude-sonnet-4-6") embeddings = VoyageAIEmbeddings(model="voyage-3-large") vectorstore = PineconeVectorStore(index_name="docs", embedding=embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 4}) # RAG prompt rag_prompt = ChatPromptTemplate.from_template( """Answer based on the context below. If you cannot answer, say so. Context: {context} Question: {question} Answer:""" ) async def retrieve(state: RAGState) -> RAGState: """Retrieve relevant documents.""" docs = await retriever.ainvoke(state["question"]) return {"context": docs} async def generate(state: RAGState) -> RAGState: """Generate answer from context.""" context_text = "\n\n".join(doc.page_content for doc in state["context"]) messages = rag_prompt.format_messages( context=context_text, question=state["question"] ) response = await llm.ainvoke(messages) return {"answer": response.content} # Build RAG graph builder = StateGraph(RAGState) builder.add_node("retrieve", retrieve) builder.add_node("generate", generate) builder.add_edge(START, "retrieve") builder.add_edge("retrieve", "generate") builder.add_edge("generate", END) rag_chain = builder.compile() # Use result = await rag_chain.ainvoke({"question": "What are the main features?"}) print(result["answer"]) ``` ## Advanced RAG Patterns ### Pattern 1: Hybrid Search with RRF ```python from langchain_community.retrievers import BM25Retriever from langchain.retrievers import EnsembleRetriever # Sparse retriever (BM25 for keyword matching) bm25_retriever = BM25Retriever.from_documents(documents) bm25_retriever.k = 10 # Dense retriever (embeddings for semantic search) dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 10}) # Combine with Reciprocal Rank Fusion weights ensemble_retriever = EnsembleRetriever( retrievers=[bm25_retriever, dense_retriever], weights=[0.3, 0.7] # 30% keyword, 70% semantic ) ``` ### Pattern 2: Multi-Query Retrieval ```python from langchain.retrievers.multi_query import MultiQueryRetriever # Generate multiple query perspectives for better recall multi_query_retriever = MultiQueryRetriever.from_llm( retriever=vectorstore.as_retriever(search_kwargs={"k": 5}), llm=llm ) # Single query β†’ multiple variations β†’ combined results results = await multi_query_retriever.ainvoke("What is the main topic?") ``` ### Pattern 3: Contextual Compression ```python from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor # Compressor extracts only relevant portions compressor = LLMChainExtractor.from_llm(llm) compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10}) ) # Returns only relevant parts of documents compressed_docs = await compression_retriever.ainvoke("specific query") ``` ### Pattern 4: Parent Document Retriever ```python from langchain.retrievers import ParentDocumentRetriever from langchain.storage import InMemoryStore from langchain_text_splitters import RecursiveCharacterTextSplitter # Small chunks for precise retrieval, large chunks for context child_splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=50) parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200) # Store for parent documents docstore = InMemoryStore() parent_retriever = ParentDocumentRetriever( vectorstore=vectorstore, docstore=docstore, child_splitter=child_splitter, parent_splitter=parent_splitter ) # Add documents (splits children, stores parents) await parent_retriever.aadd_documents(documents) # Retrieval returns parent documents with full context results = await parent_retriever.ainvoke("query") ``` ### Pattern 5: HyDE (Hypothetical Document Embeddings) ```python from langchain_core.prompts import ChatPromptTemplate class HyDEState(TypedDict): question: str hypothetical_doc: str context: list[Document] answer: str hyde_prompt = ChatPromptTemplate.from_template( """Write a detailed passage that would answer this question: Question: {question} Passage:""" ) async def generate_hypothetical(state: HyDEState) -> HyDEState: """Generate hypothetical document for better retrieval.""" messages = hyde_prompt.format_messages(question=state["question"]) response = await llm.ainvoke(messages) return {"hypothetical_doc": response.content} async def retrieve_with_hyde(state: HyDEState) -> HyDEState: """Retrieve using hypothetical document.""" # Use hypothetical doc for retrieval instead of original query docs = await retriever.ainvoke(state["hypothetical_doc"]) return {"context": docs} # Build HyDE RAG graph builder = StateGraph(HyDEState) builder.add_node("hypothetical", generate_hypothetical) builder.add_node("retrieve", retrieve_with_hyde) builder.add_node("generate", generate) builder.add_edge(START, "hypothetical") builder.add_edge("hypothetical", "retrieve") builder.add_edge("retrieve", "generate") builder.add_edge("generate", END) hyde_rag = builder.compile() ``` ## Document Chunking Strategies ### Recursive Character Text Splitter ```python from langchain_text_splitters import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, length_function=len, separators=["\n\n", "\n", ". ", " ", ""] # Try in order ) chunks = splitter.split_documents(documents) ``` ### Token-Based Splitting ```python from langchain_text_splitters import TokenTextSplitter splitter = TokenTextSplitter( chunk_size=512, chunk_overlap=50, encoding_name="cl100k_base" # OpenAI tiktoken encoding ) ``` ### Semantic Chunking ```python from langchain_experimental.text_splitter import SemanticChunker splitter = SemanticChunker( embeddings=embeddings, breakpoint_threshold_type="percentile", breakpoint_threshold_amount=95 ) ``` ### Markdown Header Splitter ```python from langchain_text_splitters import MarkdownHeaderTextSplitter headers_to_split_on = [ ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"), ] splitter = MarkdownHeaderTextSplitter( headers_to_split_on=headers_to_split_on, strip_headers=False ) ``` ## Vector Store Configurations ### Pinecone (Serverless) ```python from pinecone import Pinecone, ServerlessSpec from langchain_pinecone import PineconeVectorStore # Initialize Pinecone client pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"]) # Create index if needed if "my-index" not in pc.list_indexes().names(): pc.create_index( name="my-index", dimension=1024, # voyage-3-large dimensions metric="cosine", spec=ServerlessSpec(cloud="aws", region="us-east-1") ) # Create vector store index = pc.Index("my-index") vectorstore = PineconeVectorStore(index=index, embedding=embeddings) ``` ### Weaviate ```python import weaviate from langchain_weaviate import WeaviateVectorStore client = weaviate.connect_to_local() # or connect_to_weaviate_cloud() vectorstore = WeaviateVectorStore( client=client, index_name="Documents", text_key="content", embedding=embeddings ) ``` ### Chroma (Local Development) ```python from langchain_chroma import Chroma vectorstore = Chroma( collection_name="my_collection", embedding_function=embeddings, persist_directory="./chroma_db" ) ``` ### pgvector (PostgreSQL) ```python from langchain_postgres.vectorstores import PGVector connection_string = "postgresql+psycopg://user:pass@localhost:5432/vectordb" vectorstore = PGVector( embeddings=embeddings, collection_name="documents", connection=connection_string, ) ``` ## Retrieval Optimization ### 1. Metadata Filtering ```python from langchain_core.documents import Document # Add metadata during indexing docs_with_metadata = [] for doc in documents: doc.metadata.update({ "source": doc.metadata.get("source", "unknown"), "category": determine_category(doc.page_content), "date": datetime.now().isoformat() }) docs_with_metadata.append(doc) # Filter during retrieval results = await vectorstore.asimilarity_search( "query", filter={"category": "technical"}, k=5 ) ``` ### 2. Maximal Marginal Relevance (MMR) ```python # Balance relevance with diversity results = await vectorstore.amax_marginal_relevance_search( "query", k=5, fetch_k=20, # Fetch 20, return top 5 diverse lambda_mult=0.5 # 0=max diversity, 1=max relevance ) ``` ### 3. Reranking with Cross-Encoder ```python from sentence_transformers import CrossEncoder reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') async def retrieve_and_rerank(query: str, k: int = 5) -> list[Document]: # Get initial results candidates = await vectorstore.asimilarity_search(query, k=20) # Rerank pairs = [[query, doc.page_content] for doc in candidates] scores = reranker.predict(pairs) # Sort by score and take top k ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True) return [doc for doc, score in ranked[:k]] ``` ### 4. Cohere Rerank ```python from langchain.retrievers import CohereRerank from langchain_cohere import CohereRerank reranker = CohereRerank(model="rerank-english-v3.0", top_n=5) # Wrap retriever with reranking reranked_retriever = ContextualCompressionRetriever( base_compressor=reranker, base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20}) ) ``` ## Prompt Engineering for RAG ### Contextual Prompt with Citations ```python rag_prompt = ChatPromptTemplate.from_template( """Answer the question based on the context below. Include citations using [1], [2], etc. If you cannot answer based on the context, say "I don't have enough information." Context: {context} Question: {question} Instructions: 1. Use only information from the context 2. Cite sources with [1], [2] format 3. If uncertain, express uncertainty Answer (with citations):""" ) ``` ### Structured Output for RAG ```python from pydantic import BaseModel, Field class RAGResponse(BaseModel): answer: str = Field(description="The answer based on context") confidence: float = Field(description="Confidence score 0-1") sources: list[str] = Field(description="Source document IDs used") reasoning: str = Field(description="Brief reasoning for the answer") # Use with structured output structured_llm = llm.with_structured_output(RAGResponse) ``` ## Evaluation Metrics ```python from typing import TypedDict class RAGEvalMetrics(TypedDict): retrieval_precision: float # Relevant docs / retrieved docs retrieval_recall: float # Retrieved relevant / total relevant answer_relevance: float # Answer addresses question faithfulness: float # Answer grounded in context context_relevance: float # Context relevant to question async def evaluate_rag_system( rag_chain, test_cases: list[dict] ) -> RAGEvalMetrics: """Evaluate RAG system on test cases.""" metrics = {k: [] for k in RAGEvalMetrics.__annotations__} for test in test_cases: result = await rag_chain.ainvoke({"question": test["question"]}) # Retrieval metrics retrieved_ids = {doc.metadata["id"] for doc in result["context"]} relevant_ids = set(test["relevant_doc_ids"]) precision = len(retrieved_ids & relevant_ids) / len(retrieved_ids) recall = len(retrieved_ids & relevant_ids) / len(relevant_ids) metrics["retrieval_precision"].append(precision) metrics["retrieval_recall"].append(recall) # Use LLM-as-judge for quality metrics quality = await evaluate_answer_quality( question=test["question"], answer=result["answer"], context=result["context"], expected=test.get("expected_answer") ) metrics["answer_relevance"].append(quality["relevance"]) metrics["faithfulness"].append(quality["faithfulness"]) metrics["context_relevance"].append(quality["context_relevance"]) return {k: sum(v) / len(v) for k, v in metrics.items()} ``` ## Resources - [LangChain RAG Tutorial](https://python.langchain.com/docs/tutorials/rag/) - [LangGraph RAG Examples](https://langchain-ai.github.io/langgraph/tutorials/rag/) - [Pinecone Best Practices](https://docs.pinecone.io/guides/get-started/overview) - [Voyage AI Embeddings](https://docs.voyageai.com/) - [RAG Evaluation Guide](https://docs.ragas.io/) ## Best Practices 1. **Chunk Size**: Balance between context (larger) and specificity (smaller) - typically 500-1000 tokens 2. **Overlap**: Use 10-20% overlap to preserve context at boundaries 3. **Metadata**: Include source, page, timestamp for filtering and debugging 4. **Hybrid Search**: Combine semantic and keyword search for best recall 5. **Reranking**: Use cross-encoder reranking for precision-critical applications 6. **Citations**: Always return source documents for transparency 7. **Evaluation**: Continuously test retrieval quality and answer accuracy 8. **Monitoring**: Track retrieval metrics and latency in production ## Common Issues - **Poor Retrieval**: Check embedding quality, chunk size, query formulation - **Irrelevant Results**: Add metadata filtering, use hybrid search, rerank - **Missing Information**: Ensure documents are properly indexed, check chunking - **Slow Queries**: Optimize vector store, use caching, reduce k - **Hallucinations**: Improve grounding prompt, add verification step - **Context Too Long**: Use compression or parent document retriever
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

similarity-search-patterns

Implement efficient similarity search with vector databases. Use

coding
⭐1
# Similarity Search Patterns Patterns for implementing efficient similarity search in production systems. ## When to Use This Skill - Building semantic search systems - Implementing RAG retrieval - Creating recommendation engines - Optimizing search latency - Scaling to millions of vectors - Combining semantic and keyword search ## Core Concepts ### 1. Distance Metrics | Metric | Formula | Best For | | ------------------ | ------------------ | --------------------- | --- | -------------- | | **Cosine** | 1 - (AΒ·B)/(β€–Aβ€–β€–Bβ€–) | Normalized embeddings | | **Euclidean (L2)** | √Σ(a-b)Β² | Raw embeddings | | **Dot Product** | AΒ·B | Magnitude matters | | **Manhattan (L1)** | Ξ£ | a-b | | Sparse vectors | ### 2. Index Types ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Index Types β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ Flat β”‚ HNSW β”‚ IVF+PQ β”‚ β”‚ (Exact) β”‚ (Graph-based) β”‚ (Quantized) β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ O(n) search β”‚ O(log n) β”‚ O(√n) β”‚ β”‚ 100% recall β”‚ ~95-99% β”‚ ~90-95% β”‚ β”‚ Small data β”‚ Medium-Large β”‚ Very Large β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` ## Templates ### Template 1: Pinecone Implementation ```python from pinecone import Pinecone, ServerlessSpec from typing import List, Dict, Optional import hashlib class PineconeVectorStore: def __init__( self, api_key: str, index_name: str, dimension: int = 1536, metric: str = "cosine" ): self.pc = Pinecone(api_key=api_key) # Create index if not exists if index_name not in self.pc.list_indexes().names(): self.pc.create_index( name=index_name, dimension=dimension, metric=metric, spec=ServerlessSpec(cloud="aws", region="us-east-1") ) self.index = self.pc.Index(index_name) def upsert( self, vectors: List[Dict], namespace: str = "" ) -> int: """ Upsert vectors. vectors: [{"id": str, "values": List[float], "metadata": dict}] """ # Batch upsert batch_size = 100 total = 0 for i in range(0, len(vectors), batch_size): batch = vectors[i:i + batch_size] self.index.upsert(vectors=batch, namespace=namespace) total += len(batch) return total def search( self, query_vector: List[float], top_k: int = 10, namespace: str = "", filter: Optional[Dict] = None, include_metadata: bool = True ) -> List[Dict]: """Search for similar vectors.""" results = self.index.query( vector=query_vector, top_k=top_k, namespace=namespace, filter=filter, include_metadata=include_metadata ) return [ { "id": match.id, "score": match.score, "metadata": match.metadata } for match in results.matches ] def search_with_rerank( self, query: str, query_vector: List[float], top_k: int = 10, rerank_top_n: int = 50, namespace: str = "" ) -> List[Dict]: """Search and rerank results.""" # Over-fetch for reranking initial_results = self.search( query_vector, top_k=rerank_top_n, namespace=namespace ) # Rerank with cross-encoder or LLM reranked = self._rerank(query, initial_results) return reranked[:top_k] def _rerank(self, query: str, results: List[Dict]) -> List[Dict]: """Rerank results using cross-encoder.""" from sentence_transformers import CrossEncoder model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') pairs = [(query, r["metadata"]["text"]) for r in results] scores = model.predict(pairs) for result, score in zip(results, scores): result["rerank_score"] = float(score) return sorted(results, key=lambda x: x["rerank_score"], reverse=True) def delete(self, ids: List[str], namespace: str = ""): """Delete vectors by ID.""" self.index.delete(ids=ids, namespace=namespace) def delete_by_filter(self, filter: Dict, namespace: str = ""): """Delete vectors matching filter.""" self.index.delete(filter=filter, namespace=namespace) ``` ### Template 2: Qdrant Implementation ```python from qdrant_client import QdrantClient from qdrant_client.http import models from typing import List, Dict, Optional class QdrantVectorStore: def __init__( self, url: str = "localhost", port: int = 6333, collection_name: str = "documents", vector_size: int = 1536 ): self.client = QdrantClient(url=url, port=port) self.collection_name = collection_name # Create collection if not exists collections = self.client.get_collections().collections if collection_name not in [c.name for c in collections]: self.client.create_collection( collection_name=collection_name, vectors_config=models.VectorParams( size=vector_size, distance=models.Distance.COSINE ), # Optional: enable quantization for memory efficiency quantization_config=models.ScalarQuantization( scalar=models.ScalarQuantizationConfig( type=models.ScalarType.INT8, quantile=0.99, always_ram=True ) ) ) def upsert(self, points: List[Dict]) -> int: """ Upsert points. points: [{"id": str/int, "vector": List[float], "payload": dict}] """ qdrant_points = [ models.PointStruct( id=p["id"], vector=p["vector"], payload=p.get("payload", {}) ) for p in points ] self.client.upsert( collection_name=self.collection_name, points=qdrant_points ) return len(points) def search( self, query_vector: List[float], limit: int = 10, filter: Optional[models.Filter] = None, score_threshold: Optional[float] = None ) -> List[Dict]: """Search for similar vectors.""" results = self.client.search( collection_name=self.collection_name, query_vector=query_vector, limit=limit, query_filter=filter, score_threshold=score_threshold ) return [ { "id": r.id, "score": r.score, "payload": r.payload } for r in results ] def search_with_filter( self, query_vector: List[float], must_conditions: List[Dict] = None, should_conditions: List[Dict] = None, must_not_conditions: List[Dict] = None, limit: int = 10 ) -> List[Dict]: """Search with complex filters.""" conditions = [] if must_conditions: conditions.extend([ models.FieldCondition( key=c["key"], match=models.MatchValue(value=c["value"]) ) for c in must_conditions ]) filter = models.Filter(must=conditions) if conditions else None return self.search(query_vector, limit=limit, filter=filter) def search_with_sparse( self, dense_vector: List[float], sparse_vector: Dict[int, float], limit: int = 10, dense_weight: float = 0.7 ) -> List[Dict]: """Hybrid search with dense and sparse vectors.""" # Requires collection with named vectors results = self.client.search( collection_name=self.collection_name, query_vector=models.NamedVector( name="dense", vector=dense_vector ), limit=limit ) return [{"id": r.id, "score": r.score, "payload": r.payload} for r in results] ``` ### Template 3: pgvector with PostgreSQL ```python import asyncpg from typing import List, Dict, Optional import numpy as np class PgVectorStore: def __init__(self, connection_string: str): self.connection_string = connection_string async def init(self): """Initialize connection pool and extension.""" self.pool = await asyncpg.create_pool(self.connection_string) async with self.pool.acquire() as conn: # Enable extension await conn.execute("CREATE EXTENSION IF NOT EXISTS vector") # Create table await conn.execute(""" CREATE TABLE IF NOT EXISTS documents ( id TEXT PRIMARY KEY, content TEXT, metadata JSONB, embedding vector(1536) ) """) # Create index (HNSW for better performance) await conn.execute(""" CREATE INDEX IF NOT EXISTS documents_embedding_idx ON documents USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64) """) async def upsert(self, documents: List[Dict]): """Upsert documents with embeddings.""" async with self.pool.acquire() as conn: await conn.executemany( """ INSERT INTO documents (id, content, metadata, embedding) VALUES ($1, $2, $3, $4) ON CONFLICT (id) DO UPDATE SET content = EXCLUDED.content, metadata = EXCLUDED.metadata, embedding = EXCLUDED.embedding """, [ ( doc["id"], doc["content"], doc.get("metadata", {}), np.array(doc["embedding"]).tolist() ) for doc in documents ] ) async def search( self, query_embedding: List[float], limit: int = 10, filter_metadata: Optional[Dict] = None ) -> List[Dict]: """Search for similar documents.""" query = """ SELECT id, content, metadata, 1 - (embedding <=> $1::vector) as similarity FROM documents """ params = [query_embedding] if filter_metadata: conditions = [] for key, value in filter_metadata.items(): params.append(value) conditions.append(f"metadata->>'{key}' = ${len(params)}") query += " WHERE " + " AND ".join(conditions) query += f" ORDER BY embedding <=> $1::vector LIMIT ${len(params) + 1}" params.append(limit) async with self.pool.acquire() as conn: rows = await conn.fetch(query, *params) return [ { "id": row["id"], "content": row["content"], "metadata": row["metadata"], "score": row["similarity"] } for row in rows ] async def hybrid_search( self, query_embedding: List[float], query_text: str, limit: int = 10, vector_weight: float = 0.5 ) -> List[Dict]: """Hybrid search combining vector and full-text.""" async with self.pool.acquire() as conn: rows = await conn.fetch( """ WITH vector_results AS ( SELECT id, content, metadata, 1 - (embedding <=> $1::vector) as vector_score FROM documents ORDER BY embedding <=> $1::vector LIMIT $3 * 2 ), text_results AS ( SELECT id, content, metadata, ts_rank(to_tsvector('english', content), plainto_tsquery('english', $2)) as text_score FROM documents WHERE to_tsvector('english', content) @@ plainto_tsquery('english', $2) LIMIT $3 * 2 ) SELECT COALESCE(v.id, t.id) as id, COALESCE(v.content, t.content) as content, COALESCE(v.metadata, t.metadata) as metadata, COALESCE(v.vector_score, 0) * $4 + COALESCE(t.text_score, 0) * (1 - $4) as combined_score FROM vector_results v FULL OUTER JOIN text_results t ON v.id = t.id ORDER BY combined_score DESC LIMIT $3 """, query_embedding, query_text, limit, vector_weight ) return [dict(row) for row in rows] ``` ### Template 4: Weaviate Implementation ```python import weaviate from weaviate.util import generate_uuid5 from typing import List, Dict, Optional class WeaviateVectorStore: def __init__( self, url: str = "http://localhost:8080", class_name: str = "Document" ): self.client = weaviate.Client(url=url) self.class_name = class_name self._ensure_schema() def _ensure_schema(self): """Create schema if not exists.""" schema = { "class": self.class_name, "vectorizer": "none", # We provide vectors "properties": [ {"name": "content", "dataType": ["text"]}, {"name": "source", "dataType": ["string"]}, {"name": "chunk_id", "dataType": ["int"]} ] } if not self.client.schema.exists(self.class_name): self.client.schema.create_class(schema) def upsert(self, documents: List[Dict]): """Batch upsert documents.""" with self.client.batch as batch: batch.batch_size = 100 for doc in documents: batch.add_data_object( data_object={ "content": doc["content"], "source": doc.get("source", ""), "chunk_id": doc.get("chunk_id", 0) }, class_name=self.class_name, uuid=generate_uuid5(doc["id"]), vector=doc["embedding"] ) def search( self, query_vector: List[float], limit: int = 10, where_filter: Optional[Dict] = None ) -> List[Dict]: """Vector search.""" query = ( self.client.query .get(self.class_name, ["content", "source", "chunk_id"]) .with_near_vector({"vector": query_vector}) .with_limit(limit) .with_additional(["distance", "id"]) ) if where_filter: query = query.with_where(where_filter) results = query.do() return [ { "id": item["_additional"]["id"], "content": item["content"], "source": item["source"], "score": 1 - item["_additional"]["distance"] } for item in results["data"]["Get"][self.class_name] ] def hybrid_search( self, query: str, query_vector: List[float], limit: int = 10, alpha: float = 0.5 # 0 = keyword, 1 = vector ) -> List[Dict]: """Hybrid search combining BM25 and vector.""" results = ( self.client.query .get(self.class_name, ["content", "source"]) .with_hybrid(query=query, vector=query_vector, alpha=alpha) .with_limit(limit) .with_additional(["score"]) .do() ) return [ { "content": item["content"], "source": item["source"], "score": item["_additional"]["score"] } for item in results["data"]["Get"][self.class_name] ] ``` ## Best Practices ### Do's - **Use appropriate index** - HNSW for most cases - **Tune parameters** - ef_search, nprobe for recall/speed - **Implement hybrid search** - Combine with keyword search - **Monitor recall** - Measure search quality - **Pre-filter when possible** - Reduce search space ### Don'ts - **Don't skip evaluation** - Measure before optimizing - **Don't over-index** - Start with flat, scale up - **Don't ignore latency** - P99 matters for UX - **Don't forget costs** - Vector storage adds up ## Resources - [Pinecone Docs](https://docs.pinecone.io/) - [Qdrant Docs](https://qdrant.tech/documentation/) - [pgvector](https://github.com/pgvector/pgvector) - [Weaviate Docs](https://weaviate.io/developers/weaviate)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

microservices-patterns

Design microservices architectures with service boundaries,

coding
⭐1
# Microservices Patterns Master microservices architecture patterns including service boundaries, inter-service communication, data management, and resilience patterns for building distributed systems. ## When to Use This Skill - Decomposing monoliths into microservices - Designing service boundaries and contracts - Implementing inter-service communication - Managing distributed data and transactions - Building resilient distributed systems - Implementing service discovery and load balancing - Designing event-driven architectures ## Core Concepts ### 1. Service Decomposition Strategies **By Business Capability** - Organize services around business functions - Each service owns its domain - Example: OrderService, PaymentService, InventoryService **By Subdomain (DDD)** - Core domain, supporting subdomains - Bounded contexts map to services - Clear ownership and responsibility **Strangler Fig Pattern** - Gradually extract from monolith - New functionality as microservices - Proxy routes to old/new systems ### 2. Communication Patterns **Synchronous (Request/Response)** - REST APIs - gRPC - GraphQL **Asynchronous (Events/Messages)** - Event streaming (Kafka) - Message queues (RabbitMQ, SQS) - Pub/Sub patterns ### 3. Data Management **Database Per Service** - Each service owns its data - No shared databases - Loose coupling **Saga Pattern** - Distributed transactions - Compensating actions - Eventual consistency ### 4. Resilience Patterns **Circuit Breaker** - Fail fast on repeated errors - Prevent cascade failures **Retry with Backoff** - Transient fault handling - Exponential backoff **Bulkhead** - Isolate resources - Limit impact of failures ## Service Decomposition Patterns ### Pattern 1: By Business Capability ```python # E-commerce example # Order Service class OrderService: """Handles order lifecycle.""" async def create_order(self, order_data: dict) -> Order: order = Order.create(order_data) # Publish event for other services await self.event_bus.publish( OrderCreatedEvent( order_id=order.id, customer_id=order.customer_id, items=order.items, total=order.total ) ) return order # Payment Service (separate service) class PaymentService: """Handles payment processing.""" async def process_payment(self, payment_request: PaymentRequest) -> PaymentResult: # Process payment result = await self.payment_gateway.charge( amount=payment_request.amount, customer=payment_request.customer_id ) if result.success: await self.event_bus.publish( PaymentCompletedEvent( order_id=payment_request.order_id, transaction_id=result.transaction_id ) ) return result # Inventory Service (separate service) class InventoryService: """Handles inventory management.""" async def reserve_items(self, order_id: str, items: List[OrderItem]) -> ReservationResult: # Check availability for item in items: available = await self.inventory_repo.get_available(item.product_id) if available < item.quantity: return ReservationResult( success=False, error=f"Insufficient inventory for {item.product_id}" ) # Reserve items reservation = await self.create_reservation(order_id, items) await self.event_bus.publish( InventoryReservedEvent( order_id=order_id, reservation_id=reservation.id ) ) return ReservationResult(success=True, reservation=reservation) ``` ### Pattern 2: API Gateway ```python from fastapi import FastAPI, HTTPException, Depends import httpx from circuitbreaker import circuit app = FastAPI() class APIGateway: """Central entry point for all client requests.""" def __init__(self): self.order_service_url = "http://order-service:8000" self.payment_service_url = "http://payment-service:8001" self.inventory_service_url = "http://inventory-service:8002" self.http_client = httpx.AsyncClient(timeout=5.0) @circuit(failure_threshold=5, recovery_timeout=30) async def call_order_service(self, path: str, method: str = "GET", **kwargs): """Call order service with circuit breaker.""" response = await self.http_client.request( method, f"{self.order_service_url}{path}", **kwargs ) response.raise_for_status() return response.json() async def create_order_aggregate(self, order_id: str) -> dict: """Aggregate data from multiple services.""" # Parallel requests order, payment, inventory = await asyncio.gather( self.call_order_service(f"/orders/{order_id}"), self.call_payment_service(f"/payments/order/{order_id}"), self.call_inventory_service(f"/reservations/order/{order_id}"), return_exceptions=True ) # Handle partial failures result = {"order": order} if not isinstance(payment, Exception): result["payment"] = payment if not isinstance(inventory, Exception): result["inventory"] = inventory return result @app.post("/api/orders") async def create_order( order_data: dict, gateway: APIGateway = Depends() ): """API Gateway endpoint.""" try: # Route to order service order = await gateway.call_order_service( "/orders", method="POST", json=order_data ) return {"order": order} except httpx.HTTPError as e: raise HTTPException(status_code=503, detail="Order service unavailable") ``` ## Communication Patterns ### Pattern 1: Synchronous REST Communication ```python # Service A calls Service B import httpx from tenacity import retry, stop_after_attempt, wait_exponential class ServiceClient: """HTTP client with retries and timeout.""" def __init__(self, base_url: str): self.base_url = base_url self.client = httpx.AsyncClient( timeout=httpx.Timeout(5.0, connect=2.0), limits=httpx.Limits(max_keepalive_connections=20) ) @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10) ) async def get(self, path: str, **kwargs): """GET with automatic retries.""" response = await self.client.get(f"{self.base_url}{path}", **kwargs) response.raise_for_status() return response.json() async def post(self, path: str, **kwargs): """POST request.""" response = await self.client.post(f"{self.base_url}{path}", **kwargs) response.raise_for_status() return response.json() # Usage payment_client = ServiceClient("http://payment-service:8001") result = await payment_client.post("/payments", json=payment_data) ``` ### Pattern 2: Asynchronous Event-Driven ```python # Event-driven communication with Kafka from aiokafka import AIOKafkaProducer, AIOKafkaConsumer import json from dataclasses import dataclass, asdict from datetime import datetime @dataclass class DomainEvent: event_id: str event_type: str aggregate_id: str occurred_at: datetime data: dict class EventBus: """Event publishing and subscription.""" def __init__(self, bootstrap_servers: List[str]): self.bootstrap_servers = bootstrap_servers self.producer = None async def start(self): self.producer = AIOKafkaProducer( bootstrap_servers=self.bootstrap_servers, value_serializer=lambda v: json.dumps(v).encode() ) await self.producer.start() async def publish(self, event: DomainEvent): """Publish event to Kafka topic.""" topic = event.event_type await self.producer.send_and_wait( topic, value=asdict(event), key=event.aggregate_id.encode() ) async def subscribe(self, topic: str, handler: callable): """Subscribe to events.""" consumer = AIOKafkaConsumer( topic, bootstrap_servers=self.bootstrap_servers, value_deserializer=lambda v: json.loads(v.decode()), group_id="my-service" ) await consumer.start() try: async for message in consumer: event_data = message.value await handler(event_data) finally: await consumer.stop() # Order Service publishes event async def create_order(order_data: dict): order = await save_order(order_data) event = DomainEvent( event_id=str(uuid.uuid4()), event_type="OrderCreated", aggregate_id=order.id, occurred_at=datetime.now(), data={ "order_id": order.id, "customer_id": order.customer_id, "total": order.total } ) await event_bus.publish(event) # Inventory Service listens for OrderCreated async def handle_order_created(event_data: dict): """React to order creation.""" order_id = event_data["data"]["order_id"] items = event_data["data"]["items"] # Reserve inventory await reserve_inventory(order_id, items) ``` ### Pattern 3: Saga Pattern (Distributed Transactions) ```python # Saga orchestration for order fulfillment from enum import Enum from typing import List, Callable class SagaStep: """Single step in saga.""" def __init__( self, name: str, action: Callable, compensation: Callable ): self.name = name self.action = action self.compensation = compensation class SagaStatus(Enum): PENDING = "pending" COMPLETED = "completed" COMPENSATING = "compensating" FAILED = "failed" class OrderFulfillmentSaga: """Orchestrated saga for order fulfillment.""" def __init__(self): self.steps: List[SagaStep] = [ SagaStep( "create_order", action=self.create_order, compensation=self.cancel_order ), SagaStep( "reserve_inventory", action=self.reserve_inventory, compensation=self.release_inventory ), SagaStep( "process_payment", action=self.process_payment, compensation=self.refund_payment ), SagaStep( "confirm_order", action=self.confirm_order, compensation=self.cancel_order_confirmation ) ] async def execute(self, order_data: dict) -> SagaResult: """Execute saga steps.""" completed_steps = [] context = {"order_data": order_data} try: for step in self.steps: # Execute step result = await step.action(context) if not result.success: # Compensate await self.compensate(completed_steps, context) return SagaResult( status=SagaStatus.FAILED, error=result.error ) completed_steps.append(step) context.update(result.data) return SagaResult(status=SagaStatus.COMPLETED, data=context) except Exception as e: # Compensate on error await self.compensate(completed_steps, context) return SagaResult(status=SagaStatus.FAILED, error=str(e)) async def compensate(self, completed_steps: List[SagaStep], context: dict): """Execute compensating actions in reverse order.""" for step in reversed(completed_steps): try: await step.compensation(context) except Exception as e: # Log compensation failure print(f"Compensation failed for {step.name}: {e}") # Step implementations async def create_order(self, context: dict) -> StepResult: order = await order_service.create(context["order_data"]) return StepResult(success=True, data={"order_id": order.id}) async def cancel_order(self, context: dict): await order_service.cancel(context["order_id"]) async def reserve_inventory(self, context: dict) -> StepResult: result = await inventory_service.reserve( context["order_id"], context["order_data"]["items"] ) return StepResult( success=result.success, data={"reservation_id": result.reservation_id} ) async def release_inventory(self, context: dict): await inventory_service.release(context["reservation_id"]) async def process_payment(self, context: dict) -> StepResult: result = await payment_service.charge( context["order_id"], context["order_data"]["total"] ) return StepResult( success=result.success, data={"transaction_id": result.transaction_id}, error=result.error ) async def refund_payment(self, context: dict): await payment_service.refund(context["transaction_id"]) ``` ## Resilience Patterns ### Circuit Breaker Pattern ```python from enum import Enum from datetime import datetime, timedelta from typing import Callable, Any class CircuitState(Enum): CLOSED = "closed" # Normal operation OPEN = "open" # Failing, reject requests HALF_OPEN = "half_open" # Testing if recovered class CircuitBreaker: """Circuit breaker for service calls.""" def __init__( self, failure_threshold: int = 5, recovery_timeout: int = 30, success_threshold: int = 2 ): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.success_threshold = success_threshold self.failure_count = 0 self.success_count = 0 self.state = CircuitState.CLOSED self.opened_at = None async def call(self, func: Callable, *args, **kwargs) -> Any: """Execute function with circuit breaker.""" if self.state == CircuitState.OPEN: if self._should_attempt_reset(): self.state = CircuitState.HALF_OPEN else: raise CircuitBreakerOpenError("Circuit breaker is open") try: result = await func(*args, **kwargs) self._on_success() return result except Exception as e: self._on_failure() raise def _on_success(self): """Handle successful call.""" self.failure_count = 0 if self.state == CircuitState.HALF_OPEN: self.success_count += 1 if self.success_count >= self.success_threshold: self.state = CircuitState.CLOSED self.success_count = 0 def _on_failure(self): """Handle failed call.""" self.failure_count += 1 if self.failure_count >= self.failure_threshold: self.state = CircuitState.OPEN self.opened_at = datetime.now() if self.state == CircuitState.HALF_OPEN: self.state = CircuitState.OPEN self.opened_at = datetime.now() def _should_attempt_reset(self) -> bool: """Check if enough time passed to try again.""" return ( datetime.now() - self.opened_at > timedelta(seconds=self.recovery_timeout) ) # Usage breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30) async def call_payment_service(payment_data: dict): return await breaker.call( payment_client.process_payment, payment_data ) ``` ## Resources - **references/service-decomposition-guide.md**: Breaking down monoliths - **references/communication-patterns.md**: Sync vs async patterns - **references/saga-implementation.md**: Distributed transactions - **assets/circuit-breaker.py**: Production circuit breaker - **assets/event-bus-template.py**: Kafka event bus implementation - **assets/api-gateway-template.py**: Complete API gateway ## Best Practices 1. **Service Boundaries**: Align with business capabilities 2. **Database Per Service**: No shared databases 3. **API Contracts**: Versioned, backward compatible 4. **Async When Possible**: Events over direct calls 5. **Circuit Breakers**: Fail fast on service failures 6. **Distributed Tracing**: Track requests across services 7. **Service Registry**: Dynamic service discovery 8. **Health Checks**: Liveness and readiness probes ## Common Pitfalls - **Distributed Monolith**: Tightly coupled services - **Chatty Services**: Too many inter-service calls - **Shared Databases**: Tight coupling through data - **No Circuit Breakers**: Cascade failures - **Synchronous Everything**: Tight coupling, poor resilience - **Premature Microservices**: Starting with microservices - **Ignoring Network Failures**: Assuming reliable network - **No Compensation Logic**: Can't undo failed transactions
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

python-testing-patterns

Implement comprehensive testing strategies with pytest, fixtures,

coding
⭐1
# Python Testing Patterns Comprehensive guide to implementing robust testing strategies in Python using pytest, fixtures, mocking, parameterization, and test-driven development practices. ## When to Use This Skill - Writing unit tests for Python code - Setting up test suites and test infrastructure - Implementing test-driven development (TDD) - Creating integration tests for APIs and services - Mocking external dependencies and services - Testing async code and concurrent operations - Setting up continuous testing in CI/CD - Implementing property-based testing - Testing database operations - Debugging failing tests ## Core Concepts ### 1. Test Types - **Unit Tests**: Test individual functions/classes in isolation - **Integration Tests**: Test interaction between components - **Functional Tests**: Test complete features end-to-end - **Performance Tests**: Measure speed and resource usage ### 2. Test Structure (AAA Pattern) - **Arrange**: Set up test data and preconditions - **Act**: Execute the code under test - **Assert**: Verify the results ### 3. Test Coverage - Measure what code is exercised by tests - Identify untested code paths - Aim for meaningful coverage, not just high percentages ### 4. Test Isolation - Tests should be independent - No shared state between tests - Each test should clean up after itself ## Quick Start ```python # test_example.py def add(a, b): return a + b def test_add(): """Basic test example.""" result = add(2, 3) assert result == 5 def test_add_negative(): """Test with negative numbers.""" assert add(-1, 1) == 0 # Run with: pytest test_example.py ``` ## Fundamental Patterns ### Pattern 1: Basic pytest Tests ```python # test_calculator.py import pytest class Calculator: """Simple calculator for testing.""" def add(self, a: float, b: float) -> float: return a + b def subtract(self, a: float, b: float) -> float: return a - b def multiply(self, a: float, b: float) -> float: return a * b def divide(self, a: float, b: float) -> float: if b == 0: raise ValueError("Cannot divide by zero") return a / b def test_addition(): """Test addition.""" calc = Calculator() assert calc.add(2, 3) == 5 assert calc.add(-1, 1) == 0 assert calc.add(0, 0) == 0 def test_subtraction(): """Test subtraction.""" calc = Calculator() assert calc.subtract(5, 3) == 2 assert calc.subtract(0, 5) == -5 def test_multiplication(): """Test multiplication.""" calc = Calculator() assert calc.multiply(3, 4) == 12 assert calc.multiply(0, 5) == 0 def test_division(): """Test division.""" calc = Calculator() assert calc.divide(6, 3) == 2 assert calc.divide(5, 2) == 2.5 def test_division_by_zero(): """Test division by zero raises error.""" calc = Calculator() with pytest.raises(ValueError, match="Cannot divide by zero"): calc.divide(5, 0) ``` ### Pattern 2: Fixtures for Setup and Teardown ```python # test_database.py import pytest from typing import Generator class Database: """Simple database class.""" def __init__(self, connection_string: str): self.connection_string = connection_string self.connected = False def connect(self): """Connect to database.""" self.connected = True def disconnect(self): """Disconnect from database.""" self.connected = False def query(self, sql: str) -> list: """Execute query.""" if not self.connected: raise RuntimeError("Not connected") return [{"id": 1, "name": "Test"}] @pytest.fixture def db() -> Generator[Database, None, None]: """Fixture that provides connected database.""" # Setup database = Database("sqlite:///:memory:") database.connect() # Provide to test yield database # Teardown database.disconnect() def test_database_query(db): """Test database query with fixture.""" results = db.query("SELECT * FROM users") assert len(results) == 1 assert results[0]["name"] == "Test" @pytest.fixture(scope="session") def app_config(): """Session-scoped fixture - created once per test session.""" return { "database_url": "postgresql://localhost/test", "api_key": "test-key", "debug": True } @pytest.fixture(scope="module") def api_client(app_config): """Module-scoped fixture - created once per test module.""" # Setup expensive resource client = {"config": app_config, "session": "active"} yield client # Cleanup client["session"] = "closed" def test_api_client(api_client): """Test using api client fixture.""" assert api_client["session"] == "active" assert api_client["config"]["debug"] is True ``` ### Pattern 3: Parameterized Tests ```python # test_validation.py import pytest def is_valid_email(email: str) -> bool: """Check if email is valid.""" return "@" in email and "." in email.split("@")[1] @pytest.mark.parametrize("email,expected", [ ("user@example.com", True), ("test.user@domain.co.uk", True), ("invalid.email", False), ("@example.com", False), ("user@domain", False), ("", False), ]) def test_email_validation(email, expected): """Test email validation with various inputs.""" assert is_valid_email(email) == expected @pytest.mark.parametrize("a,b,expected", [ (2, 3, 5), (0, 0, 0), (-1, 1, 0), (100, 200, 300), (-5, -5, -10), ]) def test_addition_parameterized(a, b, expected): """Test addition with multiple parameter sets.""" from test_calculator import Calculator calc = Calculator() assert calc.add(a, b) == expected # Using pytest.param for special cases @pytest.mark.parametrize("value,expected", [ pytest.param(1, True, id="positive"), pytest.param(0, False, id="zero"), pytest.param(-1, False, id="negative"), ]) def test_is_positive(value, expected): """Test with custom test IDs.""" assert (value > 0) == expected ``` ### Pattern 4: Mocking with unittest.mock ```python # test_api_client.py import pytest from unittest.mock import Mock, patch, MagicMock import requests class APIClient: """Simple API client.""" def __init__(self, base_url: str): self.base_url = base_url def get_user(self, user_id: int) -> dict: """Fetch user from API.""" response = requests.get(f"{self.base_url}/users/{user_id}") response.raise_for_status() return response.json() def create_user(self, data: dict) -> dict: """Create new user.""" response = requests.post(f"{self.base_url}/users", json=data) response.raise_for_status() return response.json() def test_get_user_success(): """Test successful API call with mock.""" client = APIClient("https://api.example.com") mock_response = Mock() mock_response.json.return_value = {"id": 1, "name": "John Doe"} mock_response.raise_for_status.return_value = None with patch("requests.get", return_value=mock_response) as mock_get: user = client.get_user(1) assert user["id"] == 1 assert user["name"] == "John Doe" mock_get.assert_called_once_with("https://api.example.com/users/1") def test_get_user_not_found(): """Test API call with 404 error.""" client = APIClient("https://api.example.com") mock_response = Mock() mock_response.raise_for_status.side_effect = requests.HTTPError("404 Not Found") with patch("requests.get", return_value=mock_response): with pytest.raises(requests.HTTPError): client.get_user(999) @patch("requests.post") def test_create_user(mock_post): """Test user creation with decorator syntax.""" client = APIClient("https://api.example.com") mock_post.return_value.json.return_value = {"id": 2, "name": "Jane Doe"} mock_post.return_value.raise_for_status.return_value = None user_data = {"name": "Jane Doe", "email": "jane@example.com"} result = client.create_user(user_data) assert result["id"] == 2 mock_post.assert_called_once() call_args = mock_post.call_args assert call_args.kwargs["json"] == user_data ``` ### Pattern 5: Testing Exceptions ```python # test_exceptions.py import pytest def divide(a: float, b: float) -> float: """Divide a by b.""" if b == 0: raise ZeroDivisionError("Division by zero") if not isinstance(a, (int, float)) or not isinstance(b, (int, float)): raise TypeError("Arguments must be numbers") return a / b def test_zero_division(): """Test exception is raised for division by zero.""" with pytest.raises(ZeroDivisionError): divide(10, 0) def test_zero_division_with_message(): """Test exception message.""" with pytest.raises(ZeroDivisionError, match="Division by zero"): divide(5, 0) def test_type_error(): """Test type error exception.""" with pytest.raises(TypeError, match="must be numbers"): divide("10", 5) def test_exception_info(): """Test accessing exception info.""" with pytest.raises(ValueError) as exc_info: int("not a number") assert "invalid literal" in str(exc_info.value) ``` ## Advanced Patterns ### Pattern 6: Testing Async Code ```python # test_async.py import pytest import asyncio async def fetch_data(url: str) -> dict: """Fetch data asynchronously.""" await asyncio.sleep(0.1) return {"url": url, "data": "result"} @pytest.mark.asyncio async def test_fetch_data(): """Test async function.""" result = await fetch_data("https://api.example.com") assert result["url"] == "https://api.example.com" assert "data" in result @pytest.mark.asyncio async def test_concurrent_fetches(): """Test concurrent async operations.""" urls = ["url1", "url2", "url3"] tasks = [fetch_data(url) for url in urls] results = await asyncio.gather(*tasks) assert len(results) == 3 assert all("data" in r for r in results) @pytest.fixture async def async_client(): """Async fixture.""" client = {"connected": True} yield client client["connected"] = False @pytest.mark.asyncio async def test_with_async_fixture(async_client): """Test using async fixture.""" assert async_client["connected"] is True ``` ### Pattern 7: Monkeypatch for Testing ```python # test_environment.py import os import pytest def get_database_url() -> str: """Get database URL from environment.""" return os.environ.get("DATABASE_URL", "sqlite:///:memory:") def test_database_url_default(): """Test default database URL.""" # Will use actual environment variable if set url = get_database_url() assert url def test_database_url_custom(monkeypatch): """Test custom database URL with monkeypatch.""" monkeypatch.setenv("DATABASE_URL", "postgresql://localhost/test") assert get_database_url() == "postgresql://localhost/test" def test_database_url_not_set(monkeypatch): """Test when env var is not set.""" monkeypatch.delenv("DATABASE_URL", raising=False) assert get_database_url() == "sqlite:///:memory:" class Config: """Configuration class.""" def __init__(self): self.api_key = "production-key" def get_api_key(self): return self.api_key def test_monkeypatch_attribute(monkeypatch): """Test monkeypatching object attributes.""" config = Config() monkeypatch.setattr(config, "api_key", "test-key") assert config.get_api_key() == "test-key" ``` ### Pattern 8: Temporary Files and Directories ```python # test_file_operations.py import pytest from pathlib import Path def save_data(filepath: Path, data: str): """Save data to file.""" filepath.write_text(data) def load_data(filepath: Path) -> str: """Load data from file.""" return filepath.read_text() def test_file_operations(tmp_path): """Test file operations with temporary directory.""" # tmp_path is a pathlib.Path object test_file = tmp_path / "test_data.txt" # Save data save_data(test_file, "Hello, World!") # Verify file exists assert test_file.exists() # Load and verify data data = load_data(test_file) assert data == "Hello, World!" def test_multiple_files(tmp_path): """Test with multiple temporary files.""" files = { "file1.txt": "Content 1", "file2.txt": "Content 2", "file3.txt": "Content 3" } for filename, content in files.items(): filepath = tmp_path / filename save_data(filepath, content) # Verify all files created assert len(list(tmp_path.iterdir())) == 3 # Verify contents for filename, expected_content in files.items(): filepath = tmp_path / filename assert load_data(filepath) == expected_content ``` ### Pattern 9: Custom Fixtures and Conftest ```python # conftest.py """Shared fixtures for all tests.""" import pytest @pytest.fixture(scope="session") def database_url(): """Provide database URL for all tests.""" return "postgresql://localhost/test_db" @pytest.fixture(autouse=True) def reset_database(database_url): """Auto-use fixture that runs before each test.""" # Setup: Clear database print(f"Clearing database: {database_url}") yield # Teardown: Clean up print("Test completed") @pytest.fixture def sample_user(): """Provide sample user data.""" return { "id": 1, "name": "Test User", "email": "test@example.com" } @pytest.fixture def sample_users(): """Provide list of sample users.""" return [ {"id": 1, "name": "User 1"}, {"id": 2, "name": "User 2"}, {"id": 3, "name": "User 3"}, ] # Parametrized fixture @pytest.fixture(params=["sqlite", "postgresql", "mysql"]) def db_backend(request): """Fixture that runs tests with different database backends.""" return request.param def test_with_db_backend(db_backend): """This test will run 3 times with different backends.""" print(f"Testing with {db_backend}") assert db_backend in ["sqlite", "postgresql", "mysql"] ``` ### Pattern 10: Property-Based Testing ```python # test_properties.py from hypothesis import given, strategies as st import pytest def reverse_string(s: str) -> str: """Reverse a string.""" return s[::-1] @given(st.text()) def test_reverse_twice_is_original(s): """Property: reversing twice returns original.""" assert reverse_string(reverse_string(s)) == s @given(st.text()) def test_reverse_length(s): """Property: reversed string has same length.""" assert len(reverse_string(s)) == len(s) @given(st.integers(), st.integers()) def test_addition_commutative(a, b): """Property: addition is commutative.""" assert a + b == b + a @given(st.lists(st.integers())) def test_sorted_list_properties(lst): """Property: sorted list is ordered.""" sorted_lst = sorted(lst) # Same length assert len(sorted_lst) == len(lst) # All elements present assert set(sorted_lst) == set(lst) # Is ordered for i in range(len(sorted_lst) - 1): assert sorted_lst[i] <= sorted_lst[i + 1] ``` ## Test Design Principles ### One Behavior Per Test Each test should verify exactly one behavior. This makes failures easy to diagnose and tests easy to maintain. ```python # BAD - testing multiple behaviors def test_user_service(): user = service.create_user(data) assert user.id is not None assert user.email == data["email"] updated = service.update_user(user.id, {"name": "New"}) assert updated.name == "New" # GOOD - focused tests def test_create_user_assigns_id(): user = service.create_user(data) assert user.id is not None def test_create_user_stores_email(): user = service.create_user(data) assert user.email == data["email"] def test_update_user_changes_name(): user = service.create_user(data) updated = service.update_user(user.id, {"name": "New"}) assert updated.name == "New" ``` ### Test Error Paths Always test failure cases, not just happy paths. ```python def test_get_user_raises_not_found(): with pytest.raises(UserNotFoundError) as exc_info: service.get_user("nonexistent-id") assert "nonexistent-id" in str(exc_info.value) def test_create_user_rejects_invalid_email(): with pytest.raises(ValueError, match="Invalid email format"): service.create_user({"email": "not-an-email"}) ``` ## Testing Best Practices ### Test Organization ```python # tests/ # __init__.py # conftest.py # Shared fixtures # test_unit/ # Unit tests # test_models.py # test_utils.py # test_integration/ # Integration tests # test_api.py # test_database.py # test_e2e/ # End-to-end tests # test_workflows.py ``` ### Test Naming Convention A common pattern: `test_<unit>_<scenario>_<expected_outcome>`. Adapt to your team's preferences. ```python # Pattern: test_<unit>_<scenario>_<expected> def test_create_user_with_valid_data_returns_user(): ... def test_create_user_with_duplicate_email_raises_conflict(): ... def test_get_user_with_unknown_id_returns_none(): ... # Good test names - clear and descriptive def test_user_creation_with_valid_data(): """Clear name describes what is being tested.""" pass def test_login_fails_with_invalid_password(): """Name describes expected behavior.""" pass def test_api_returns_404_for_missing_resource(): """Specific about inputs and expected outcomes.""" pass # Bad test names - avoid these def test_1(): # Not descriptive pass def test_user(): # Too vague pass def test_function(): # Doesn't explain what's tested pass ``` ### Testing Retry Behavior Verify that retry logic works correctly using mock side effects. ```python from unittest.mock import Mock def test_retries_on_transient_error(): """Test that service retries on transient failures.""" client = Mock() # Fail twice, then succeed client.request.side_effect = [ ConnectionError("Failed"), ConnectionError("Failed"), {"status": "ok"}, ] service = ServiceWithRetry(client, max_retries=3) result = service.fetch() assert result == {"status": "ok"} assert client.request.call_count == 3 def test_gives_up_after_max_retries(): """Test that service stops retrying after max attempts.""" client = Mock() client.request.side_effect = ConnectionError("Failed") service = ServiceWithRetry(client, max_retries=3) with pytest.raises(ConnectionError): service.fetch() assert client.request.call_count == 3 def test_does_not_retry_on_permanent_error(): """Test that permanent errors are not retried.""" client = Mock() client.request.side_effect = ValueError("Invalid input") service = ServiceWithRetry(client, max_retries=3) with pytest.raises(ValueError): service.fetch() # Only called once - no retry for ValueError assert client.request.call_count == 1 ``` ### Mocking Time with Freezegun Use freezegun
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

anti-reversing-techniques

Understand anti-reversing, obfuscation, and protection techniques

security
⭐1
> **AUTHORIZED USE ONLY**: This skill contains dual-use security techniques. Before proceeding with any bypass or analysis: > > 1. **Verify authorization**: Confirm you have explicit written permission from the software owner, or are operating within a legitimate security context (CTF, authorized pentest, malware analysis, security research) > 2. **Document scope**: Ensure your activities fall within the defined scope of your authorization > 3. **Legal compliance**: Understand that unauthorized bypassing of software protection may violate laws (CFAA, DMCA anti-circumvention, etc.) > > **Legitimate use cases**: Malware analysis, authorized penetration testing, CTF competitions, academic security research, analyzing software you own/have rights to # Anti-Reversing Techniques Understanding protection mechanisms encountered during authorized software analysis, security research, and malware analysis. This knowledge helps analysts bypass protections to complete legitimate analysis tasks. ## Anti-Debugging Techniques ### Windows Anti-Debugging #### API-Based Detection ```c // IsDebuggerPresent if (IsDebuggerPresent()) { exit(1); } // CheckRemoteDebuggerPresent BOOL debugged = FALSE; CheckRemoteDebuggerPresent(GetCurrentProcess(), &debugged); if (debugged) exit(1); // NtQueryInformationProcess typedef NTSTATUS (NTAPI *pNtQueryInformationProcess)( HANDLE, PROCESSINFOCLASS, PVOID, ULONG, PULONG); DWORD debugPort = 0; NtQueryInformationProcess( GetCurrentProcess(), ProcessDebugPort, // 7 &debugPort, sizeof(debugPort), NULL ); if (debugPort != 0) exit(1); // Debug flags DWORD debugFlags = 0; NtQueryInformationProcess( GetCurrentProcess(), ProcessDebugFlags, // 0x1F &debugFlags, sizeof(debugFlags), NULL ); if (debugFlags == 0) exit(1); // 0 means being debugged ``` **Bypass Approaches:** ```python # x64dbg: ScyllaHide plugin # Patches common anti-debug checks # Manual patching in debugger: # - Set IsDebuggerPresent return to 0 # - Patch PEB.BeingDebugged to 0 # - Hook NtQueryInformationProcess # IDAPython: Patch checks ida_bytes.patch_byte(check_addr, 0x90) # NOP ``` #### PEB-Based Detection ```c // Direct PEB access #ifdef _WIN64 PPEB peb = (PPEB)__readgsqword(0x60); #else PPEB peb = (PPEB)__readfsdword(0x30); #endif // BeingDebugged flag if (peb->BeingDebugged) exit(1); // NtGlobalFlag // Debugged: 0x70 (FLG_HEAP_ENABLE_TAIL_CHECK | // FLG_HEAP_ENABLE_FREE_CHECK | // FLG_HEAP_VALIDATE_PARAMETERS) if (peb->NtGlobalFlag & 0x70) exit(1); // Heap flags PDWORD heapFlags = (PDWORD)((PBYTE)peb->ProcessHeap + 0x70); if (*heapFlags & 0x50000062) exit(1); ``` **Bypass Approaches:** ```assembly ; In debugger, modify PEB directly ; x64dbg: dump at gs:[60] (x64) or fs:[30] (x86) ; Set BeingDebugged (offset 2) to 0 ; Clear NtGlobalFlag (offset 0xBC for x64) ``` #### Timing-Based Detection ```c // RDTSC timing uint64_t start = __rdtsc(); // ... some code ... uint64_t end = __rdtsc(); if ((end - start) > THRESHOLD) exit(1); // QueryPerformanceCounter LARGE_INTEGER start, end, freq; QueryPerformanceFrequency(&freq); QueryPerformanceCounter(&start); // ... code ... QueryPerformanceCounter(&end); double elapsed = (double)(end.QuadPart - start.QuadPart) / freq.QuadPart; if (elapsed > 0.1) exit(1); // Too slow = debugger // GetTickCount DWORD start = GetTickCount(); // ... code ... if (GetTickCount() - start > 1000) exit(1); ``` **Bypass Approaches:** ``` - Use hardware breakpoints instead of software - Patch timing checks - Use VM with controlled time - Hook timing APIs to return consistent values ``` #### Exception-Based Detection ```c // SEH-based detection __try { __asm { int 3 } // Software breakpoint } __except(EXCEPTION_EXECUTE_HANDLER) { // Normal execution: exception caught return; } // Debugger ate the exception exit(1); // VEH-based detection LONG CALLBACK VectoredHandler(PEXCEPTION_POINTERS ep) { if (ep->ExceptionRecord->ExceptionCode == EXCEPTION_BREAKPOINT) { ep->ContextRecord->Rip++; // Skip INT3 return EXCEPTION_CONTINUE_EXECUTION; } return EXCEPTION_CONTINUE_SEARCH; } ``` ### Linux Anti-Debugging ```c // ptrace self-trace if (ptrace(PTRACE_TRACEME, 0, NULL, NULL) == -1) { // Already being traced exit(1); } // /proc/self/status FILE *f = fopen("/proc/self/status", "r"); char line[256]; while (fgets(line, sizeof(line), f)) { if (strncmp(line, "TracerPid:", 10) == 0) { int tracer_pid = atoi(line + 10); if (tracer_pid != 0) exit(1); } } // Parent process check if (getppid() != 1 && strcmp(get_process_name(getppid()), "bash") != 0) { // Unusual parent (might be debugger) } ``` **Bypass Approaches:** ```bash # LD_PRELOAD to hook ptrace # Compile: gcc -shared -fPIC -o hook.so hook.c long ptrace(int request, ...) { return 0; // Always succeed } # Usage LD_PRELOAD=./hook.so ./target ``` ## Anti-VM Detection ### Hardware Fingerprinting ```c // CPUID-based detection int cpuid_info[4]; __cpuid(cpuid_info, 1); // Check hypervisor bit (bit 31 of ECX) if (cpuid_info[2] & (1 << 31)) { // Running in hypervisor } // CPUID brand string __cpuid(cpuid_info, 0x40000000); char vendor[13] = {0}; memcpy(vendor, &cpuid_info[1], 12); // "VMwareVMware", "Microsoft Hv", "KVMKVMKVM", "VBoxVBoxVBox" // MAC address prefix // VMware: 00:0C:29, 00:50:56 // VirtualBox: 08:00:27 // Hyper-V: 00:15:5D ``` ### Registry/File Detection ```c // Windows registry keys // HKLM\SOFTWARE\VMware, Inc.\VMware Tools // HKLM\SOFTWARE\Oracle\VirtualBox Guest Additions // HKLM\HARDWARE\ACPI\DSDT\VBOX__ // Files // C:\Windows\System32\drivers\vmmouse.sys // C:\Windows\System32\drivers\vmhgfs.sys // C:\Windows\System32\drivers\VBoxMouse.sys // Processes // vmtoolsd.exe, vmwaretray.exe // VBoxService.exe, VBoxTray.exe ``` ### Timing-Based VM Detection ```c // VM exits cause timing anomalies uint64_t start = __rdtsc(); __cpuid(cpuid_info, 0); // Causes VM exit uint64_t end = __rdtsc(); if ((end - start) > 500) { // Likely in VM (CPUID takes longer) } ``` **Bypass Approaches:** ``` - Use bare-metal analysis environment - Harden VM (remove guest tools, change MAC) - Patch detection code - Use specialized analysis VMs (FLARE-VM) ``` ## Code Obfuscation ### Control Flow Obfuscation #### Control Flow Flattening ```c // Original if (cond) { func_a(); } else { func_b(); } func_c(); // Flattened int state = 0; while (1) { switch (state) { case 0: state = cond ? 1 : 2; break; case 1: func_a(); state = 3; break; case 2: func_b(); state = 3; break; case 3: func_c(); return; } } ``` **Analysis Approach:** - Identify state variable - Map state transitions - Reconstruct original flow - Tools: D-810 (IDA), SATURN #### Opaque Predicates ```c // Always true, but complex to analyze int x = rand(); if ((x * x) >= 0) { // Always true real_code(); } else { junk_code(); // Dead code } // Always false if ((x * (x + 1)) % 2 == 1) { // Product of consecutive = even junk_code(); } ``` **Analysis Approach:** - Identify constant expressions - Symbolic execution to prove predicates - Pattern matching for known opaque predicates ### Data Obfuscation #### String Encryption ```c // XOR encryption char decrypt_string(char *enc, int len, char key) { char *dec = malloc(len + 1); for (int i = 0; i < len; i++) { dec[i] = enc[i] ^ key; } dec[len] = 0; return dec; } // Stack strings char url[20]; url[0] = 'h'; url[1] = 't'; url[2] = 't'; url[3] = 'p'; url[4] = ':'; url[5] = '/'; url[6] = '/'; // ... ``` **Analysis Approach:** ```python # FLOSS for automatic string deobfuscation floss malware.exe # IDAPython string decryption def decrypt_xor(ea, length, key): result = "" for i in range(length): byte = ida_bytes.get_byte(ea + i) result += chr(byte ^ key) return result ``` #### API Obfuscation ```c // Dynamic API resolution typedef HANDLE (WINAPI *pCreateFileW)(LPCWSTR, DWORD, DWORD, LPSECURITY_ATTRIBUTES, DWORD, DWORD, HANDLE); HMODULE kernel32 = LoadLibraryA("kernel32.dll"); pCreateFileW myCreateFile = (pCreateFileW)GetProcAddress( kernel32, "CreateFileW"); // API hashing DWORD hash_api(char *name) { DWORD hash = 0; while (*name) { hash = ((hash >> 13) | (hash << 19)) + *name++; } return hash; } // Resolve by hash comparison instead of string ``` **Analysis Approach:** - Identify hash algorithm - Build hash database of known APIs - Use HashDB plugin for IDA - Dynamic analysis to resolve at runtime ### Instruction-Level Obfuscation #### Dead Code Insertion ```asm ; Original mov eax, 1 ; With dead code push ebx ; Dead mov eax, 1 pop ebx ; Dead xor ecx, ecx ; Dead add ecx, ecx ; Dead ``` #### Instruction Substitution ```asm ; Original: xor eax, eax (set to 0) ; Substitutions: sub eax, eax mov eax, 0 and eax, 0 lea eax, [0] ; Original: mov eax, 1 ; Substitutions: xor eax, eax inc eax push 1 pop eax ``` ## Packing and Encryption ### Common Packers ``` UPX - Open source, easy to unpack Themida - Commercial, VM-based protection VMProtect - Commercial, code virtualization ASPack - Compression packer PECompact - Compression packer Enigma - Commercial protector ``` ### Unpacking Methodology ``` 1. Identify packer (DIE, Exeinfo PE, PEiD) 2. Static unpacking (if known packer): - UPX: upx -d packed.exe - Use existing unpackers 3. Dynamic unpacking: a. Find Original Entry Point (OEP) b. Set breakpoint on OEP c. Dump memory when OEP reached d. Fix import table (Scylla, ImpREC) 4. OEP finding techniques: - Hardware breakpoint on stack (ESP trick) - Break on common API calls (GetCommandLineA) - Trace and look for typical entry patterns ``` ### Manual Unpacking Example ``` 1. Load packed binary in x64dbg 2. Note entry point (packer stub) 3. Use ESP trick: - Run to entry - Set hardware breakpoint on [ESP] - Run until breakpoint hits (after PUSHAD/POPAD) 4. Look for JMP to OEP 5. At OEP, use Scylla to: - Dump process - Find imports (IAT autosearch) - Fix dump ``` ## Virtualization-Based Protection ### Code Virtualization ``` Original x86 code is converted to custom bytecode interpreted by embedded VM at runtime. Original: VM Protected: mov eax, 1 push vm_context add eax, 2 call vm_entry ; VM interprets bytecode ; equivalent to original ``` ### Analysis Approaches ``` 1. Identify VM components: - VM entry (dispatcher) - Handler table - Bytecode location - Virtual registers/stack 2. Trace execution: - Log handler calls - Map bytecode to operations - Understand instruction set 3. Lifting/devirtualization: - Map VM instructions back to native - Tools: VMAttack, SATURN, NoVmp 4. Symbolic execution: - Analyze VM semantically - angr, Triton ``` ## Bypass Strategies Summary ### General Principles 1. **Understand the protection**: Identify what technique is used 2. **Find the check**: Locate protection code in binary 3. **Patch or hook**: Modify check to always pass 4. **Use appropriate tools**: ScyllaHide, x64dbg plugins 5. **Document findings**: Keep notes on bypassed protections ### Tool Recommendations ``` Anti-debug bypass: ScyllaHide, TitanHide Unpacking: x64dbg + Scylla, OllyDumpEx Deobfuscation: D-810, SATURN, miasm VM analysis: VMAttack, NoVmp, manual tracing String decryption: FLOSS, custom scripts Symbolic execution: angr, Triton ``` ### Ethical Considerations This knowledge should only be used for: - Authorized security research - Malware analysis (defensive) - CTF competitions - Understanding protections for legitimate purposes - Educational purposes Never use to bypass protections for: - Software piracy - Unauthorized access - Malicious purposes
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

binary-analysis-patterns

Master binary analysis patterns including disassembly,

security
⭐1
# Binary Analysis Patterns Comprehensive patterns and techniques for analyzing compiled binaries, understanding assembly code, and reconstructing program logic. ## Disassembly Fundamentals ### x86-64 Instruction Patterns #### Function Prologue/Epilogue ```asm ; Standard prologue push rbp ; Save base pointer mov rbp, rsp ; Set up stack frame sub rsp, 0x20 ; Allocate local variables ; Leaf function (no calls) ; May skip frame pointer setup sub rsp, 0x18 ; Just allocate locals ; Standard epilogue mov rsp, rbp ; Restore stack pointer pop rbp ; Restore base pointer ret ; Leave instruction (equivalent) leave ; mov rsp, rbp; pop rbp ret ``` #### Calling Conventions **System V AMD64 (Linux, macOS)** ```asm ; Arguments: RDI, RSI, RDX, RCX, R8, R9, then stack ; Return: RAX (and RDX for 128-bit) ; Caller-saved: RAX, RCX, RDX, RSI, RDI, R8-R11 ; Callee-saved: RBX, RBP, R12-R15 ; Example: func(a, b, c, d, e, f, g) mov rdi, [a] ; 1st arg mov rsi, [b] ; 2nd arg mov rdx, [c] ; 3rd arg mov rcx, [d] ; 4th arg mov r8, [e] ; 5th arg mov r9, [f] ; 6th arg push [g] ; 7th arg on stack call func ``` **Microsoft x64 (Windows)** ```asm ; Arguments: RCX, RDX, R8, R9, then stack ; Shadow space: 32 bytes reserved on stack ; Return: RAX ; Example: func(a, b, c, d, e) sub rsp, 0x28 ; Shadow space + alignment mov rcx, [a] ; 1st arg mov rdx, [b] ; 2nd arg mov r8, [c] ; 3rd arg mov r9, [d] ; 4th arg mov [rsp+0x20], [e] ; 5th arg on stack call func add rsp, 0x28 ``` ### ARM Assembly Patterns #### ARM64 (AArch64) Calling Convention ```asm ; Arguments: X0-X7 ; Return: X0 (and X1 for 128-bit) ; Frame pointer: X29 ; Link register: X30 ; Function prologue stp x29, x30, [sp, #-16]! ; Save FP and LR mov x29, sp ; Set frame pointer ; Function epilogue ldp x29, x30, [sp], #16 ; Restore FP and LR ret ``` #### ARM32 Calling Convention ```asm ; Arguments: R0-R3, then stack ; Return: R0 (and R1 for 64-bit) ; Link register: LR (R14) ; Function prologue push {fp, lr} add fp, sp, #4 ; Function epilogue pop {fp, pc} ; Return by popping PC ``` ## Control Flow Patterns ### Conditional Branches ```asm ; if (a == b) cmp eax, ebx jne skip_block ; ... if body ... skip_block: ; if (a < b) - signed cmp eax, ebx jge skip_block ; Jump if greater or equal ; ... if body ... skip_block: ; if (a < b) - unsigned cmp eax, ebx jae skip_block ; Jump if above or equal ; ... if body ... skip_block: ``` ### Loop Patterns ```asm ; for (int i = 0; i < n; i++) xor ecx, ecx ; i = 0 loop_start: cmp ecx, [n] ; i < n jge loop_end ; ... loop body ... inc ecx ; i++ jmp loop_start loop_end: ; while (condition) jmp loop_check loop_body: ; ... body ... loop_check: cmp eax, ebx jl loop_body ; do-while loop_body: ; ... body ... cmp eax, ebx jl loop_body ``` ### Switch Statement Patterns ```asm ; Jump table pattern mov eax, [switch_var] cmp eax, max_case ja default_case jmp [jump_table + eax*8] ; Sequential comparison (small switch) cmp eax, 1 je case_1 cmp eax, 2 je case_2 cmp eax, 3 je case_3 jmp default_case ``` ## Data Structure Patterns ### Array Access ```asm ; array[i] - 4-byte elements mov eax, [rbx + rcx*4] ; rbx=base, rcx=index ; array[i] - 8-byte elements mov rax, [rbx + rcx*8] ; Multi-dimensional array[i][j] ; arr[i][j] = base + (i * cols + j) * element_size imul eax, [cols] add eax, [j] mov edx, [rbx + rax*4] ``` ### Structure Access ```c struct Example { int a; // offset 0 char b; // offset 4 // padding // offset 5-7 long c; // offset 8 short d; // offset 16 }; ``` ```asm ; Accessing struct fields mov rdi, [struct_ptr] mov eax, [rdi] ; s->a (offset 0) movzx eax, byte [rdi+4] ; s->b (offset 4) mov rax, [rdi+8] ; s->c (offset 8) movzx eax, word [rdi+16] ; s->d (offset 16) ``` ### Linked List Traversal ```asm ; while (node != NULL) list_loop: test rdi, rdi ; node == NULL? jz list_done ; ... process node ... mov rdi, [rdi+8] ; node = node->next (assuming next at offset 8) jmp list_loop list_done: ``` ## Common Code Patterns ### String Operations ```asm ; strlen pattern xor ecx, ecx strlen_loop: cmp byte [rdi + rcx], 0 je strlen_done inc ecx jmp strlen_loop strlen_done: ; ecx contains length ; strcpy pattern strcpy_loop: mov al, [rsi] mov [rdi], al test al, al jz strcpy_done inc rsi inc rdi jmp strcpy_loop strcpy_done: ; memcpy using rep movsb mov rdi, dest mov rsi, src mov rcx, count rep movsb ``` ### Arithmetic Patterns ```asm ; Multiplication by constant ; x * 3 lea eax, [rax + rax*2] ; x * 5 lea eax, [rax + rax*4] ; x * 10 lea eax, [rax + rax*4] ; x * 5 add eax, eax ; * 2 ; Division by power of 2 (signed) mov eax, [x] cdq ; Sign extend to EDX:EAX and edx, 7 ; For divide by 8 add eax, edx ; Adjust for negative sar eax, 3 ; Arithmetic shift right ; Modulo power of 2 and eax, 7 ; x % 8 ``` ### Bit Manipulation ```asm ; Test specific bit test eax, 0x80 ; Test bit 7 jnz bit_set ; Set bit or eax, 0x10 ; Set bit 4 ; Clear bit and eax, ~0x10 ; Clear bit 4 ; Toggle bit xor eax, 0x10 ; Toggle bit 4 ; Count leading zeros bsr eax, ecx ; Bit scan reverse xor eax, 31 ; Convert to leading zeros ; Population count (popcnt) popcnt eax, ecx ; Count set bits ``` ## Decompilation Patterns ### Variable Recovery ```asm ; Local variable at rbp-8 mov qword [rbp-8], rax ; Store to local mov rax, [rbp-8] ; Load from local ; Stack-allocated array lea rax, [rbp-0x40] ; Array starts at rbp-0x40 mov [rax], edx ; array[0] = value mov [rax+4], ecx ; array[1] = value ``` ### Function Signature Recovery ```asm ; Identify parameters by register usage func: ; rdi used as first param (System V) mov [rbp-8], rdi ; Save param to local ; rsi used as second param mov [rbp-16], rsi ; Identify return by RAX at end mov rax, [result] ret ``` ### Type Recovery ```asm ; 1-byte operations suggest char/bool movzx eax, byte [rdi] ; Zero-extend byte movsx eax, byte [rdi] ; Sign-extend byte ; 2-byte operations suggest short movzx eax, word [rdi] movsx eax, word [rdi] ; 4-byte operations suggest int/float mov eax, [rdi] movss xmm0, [rdi] ; Float ; 8-byte operations suggest long/double/pointer mov rax, [rdi] movsd xmm0, [rdi] ; Double ``` ## Ghidra Analysis Tips ### Improving Decompilation ```java // In Ghidra scripting // Fix function signature Function func = getFunctionAt(toAddr(0x401000)); func.setReturnType(IntegerDataType.dataType, SourceType.USER_DEFINED); // Create structure type StructureDataType struct = new StructureDataType("MyStruct", 0); struct.add(IntegerDataType.dataType, "field_a", null); struct.add(PointerDataType.dataType, "next", null); // Apply to memory createData(toAddr(0x601000), struct); ``` ### Pattern Matching Scripts ```python # Find all calls to dangerous functions for func in currentProgram.getFunctionManager().getFunctions(True): for ref in getReferencesTo(func.getEntryPoint()): if func.getName() in ["strcpy", "sprintf", "gets"]: print(f"Dangerous call at {ref.getFromAddress()}") ``` ## IDA Pro Patterns ### IDAPython Analysis ```python import idaapi import idautils import idc # Find all function calls def find_calls(func_name): for func_ea in idautils.Functions(): for head in idautils.Heads(func_ea, idc.find_func_end(func_ea)): if idc.print_insn_mnem(head) == "call": target = idc.get_operand_value(head, 0) if idc.get_func_name(target) == func_name: print(f"Call to {func_name} at {hex(head)}") # Rename functions based on strings def auto_rename(): for s in idautils.Strings(): for xref in idautils.XrefsTo(s.ea): func = idaapi.get_func(xref.frm) if func and "sub_" in idc.get_func_name(func.start_ea): # Use string as hint for naming pass ``` ## Best Practices ### Analysis Workflow 1. **Initial triage**: File type, architecture, imports/exports 2. **String analysis**: Identify interesting strings, error messages 3. **Function identification**: Entry points, exports, cross-references 4. **Control flow mapping**: Understand program structure 5. **Data structure recovery**: Identify structs, arrays, globals 6. **Algorithm identification**: Crypto, hashing, compression 7. **Documentation**: Comments, renamed symbols, type definitions ### Common Pitfalls - **Optimizer artifacts**: Code may not match source structure - **Inline functions**: Functions may be expanded inline - **Tail call optimization**: `jmp` instead of `call` + `ret` - **Dead code**: Unreachable code from optimization - **Position-independent code**: RIP-relative addressing
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

protocol-reverse-engineering

Master network protocol reverse engineering including packet

security
⭐1
# Protocol Reverse Engineering Comprehensive techniques for capturing, analyzing, and documenting network protocols for security research, interoperability, and debugging. ## Traffic Capture ### Wireshark Capture ```bash # Capture on specific interface wireshark -i eth0 -k # Capture with filter wireshark -i eth0 -k -f "port 443" # Capture to file tshark -i eth0 -w capture.pcap # Ring buffer capture (rotate files) tshark -i eth0 -b filesize:100000 -b files:10 -w capture.pcap ``` ### tcpdump Capture ```bash # Basic capture tcpdump -i eth0 -w capture.pcap # With filter tcpdump -i eth0 port 8080 -w capture.pcap # Capture specific bytes tcpdump -i eth0 -s 0 -w capture.pcap # Full packet # Real-time display tcpdump -i eth0 -X port 80 ``` ### Man-in-the-Middle Capture ```bash # mitmproxy for HTTP/HTTPS mitmproxy --mode transparent -p 8080 # SSL/TLS interception mitmproxy --mode transparent --ssl-insecure # Dump to file mitmdump -w traffic.mitm # Burp Suite # Configure browser proxy to 127.0.0.1:8080 ``` ## Protocol Analysis ### Wireshark Analysis ``` # Display filters tcp.port == 8080 http.request.method == "POST" ip.addr == 192.168.1.1 tcp.flags.syn == 1 && tcp.flags.ack == 0 frame contains "password" # Following streams Right-click > Follow > TCP Stream Right-click > Follow > HTTP Stream # Export objects File > Export Objects > HTTP # Decryption Edit > Preferences > Protocols > TLS - (Pre)-Master-Secret log filename - RSA keys list ``` ### tshark Analysis ```bash # Extract specific fields tshark -r capture.pcap -T fields -e ip.src -e ip.dst -e tcp.port # Statistics tshark -r capture.pcap -q -z conv,tcp tshark -r capture.pcap -q -z endpoints,ip # Filter and extract tshark -r capture.pcap -Y "http" -T json > http_traffic.json # Protocol hierarchy tshark -r capture.pcap -q -z io,phs ``` ### Scapy for Custom Analysis ```python from scapy.all import * # Read pcap packets = rdpcap("capture.pcap") # Analyze packets for pkt in packets: if pkt.haslayer(TCP): print(f"Src: {pkt[IP].src}:{pkt[TCP].sport}") print(f"Dst: {pkt[IP].dst}:{pkt[TCP].dport}") if pkt.haslayer(Raw): print(f"Data: {pkt[Raw].load[:50]}") # Filter packets http_packets = [p for p in packets if p.haslayer(TCP) and (p[TCP].sport == 80 or p[TCP].dport == 80)] # Create custom packets pkt = IP(dst="target")/TCP(dport=80)/Raw(load="GET / HTTP/1.1\r\n") send(pkt) ``` ## Protocol Identification ### Common Protocol Signatures ``` HTTP - "HTTP/1." or "GET " or "POST " at start TLS/SSL - 0x16 0x03 (record layer) DNS - UDP port 53, specific header format SMB - 0xFF 0x53 0x4D 0x42 ("SMB" signature) SSH - "SSH-2.0" banner FTP - "220 " response, "USER " command SMTP - "220 " banner, "EHLO" command MySQL - 0x00 length prefix, protocol version PostgreSQL - 0x00 0x00 0x00 startup length Redis - "*" RESP array prefix MongoDB - BSON documents with specific header ``` ### Protocol Header Patterns ``` +--------+--------+--------+--------+ | Magic number / Signature | +--------+--------+--------+--------+ | Version | Flags | +--------+--------+--------+--------+ | Length | Message Type | +--------+--------+--------+--------+ | Sequence Number / Session ID | +--------+--------+--------+--------+ | Payload... | +--------+--------+--------+--------+ ``` ## Binary Protocol Analysis ### Structure Identification ```python # Common patterns in binary protocols # Length-prefixed message struct Message { uint32_t length; # Total message length uint16_t msg_type; # Message type identifier uint8_t flags; # Flags/options uint8_t reserved; # Padding/alignment uint8_t payload[]; # Variable-length payload }; # Type-Length-Value (TLV) struct TLV { uint8_t type; # Field type uint16_t length; # Field length uint8_t value[]; # Field data }; # Fixed header + variable payload struct Packet { uint8_t magic[4]; # "ABCD" signature uint32_t version; uint32_t payload_len; uint32_t checksum; # CRC32 or similar uint8_t payload[]; }; ``` ### Python Protocol Parser ```python import struct from dataclasses import dataclass @dataclass class MessageHeader: magic: bytes version: int msg_type: int length: int @classmethod def from_bytes(cls, data: bytes): magic, version, msg_type, length = struct.unpack( ">4sHHI", data[:12] ) return cls(magic, version, msg_type, length) def parse_messages(data: bytes): offset = 0 messages = [] while offset < len(data): header = MessageHeader.from_bytes(data[offset:]) payload = data[offset+12:offset+12+header.length] messages.append((header, payload)) offset += 12 + header.length return messages # Parse TLV structure def parse_tlv(data: bytes): fields = [] offset = 0 while offset < len(data): field_type = data[offset] length = struct.unpack(">H", data[offset+1:offset+3])[0] value = data[offset+3:offset+3+length] fields.append((field_type, value)) offset += 3 + length return fields ``` ### Hex Dump Analysis ```python def hexdump(data: bytes, width: int = 16): """Format binary data as hex dump.""" lines = [] for i in range(0, len(data), width): chunk = data[i:i+width] hex_part = ' '.join(f'{b:02x}' for b in chunk) ascii_part = ''.join( chr(b) if 32 <= b < 127 else '.' for b in chunk ) lines.append(f'{i:08x} {hex_part:<{width*3}} {ascii_part}') return '\n'.join(lines) # Example output: # 00000000 48 54 54 50 2f 31 2e 31 20 32 30 30 20 4f 4b 0d HTTP/1.1 200 OK. # 00000010 0a 43 6f 6e 74 65 6e 74 2d 54 79 70 65 3a 20 74 .Content-Type: t ``` ## Encryption Analysis ### Identifying Encryption ```python # Entropy analysis - high entropy suggests encryption/compression import math from collections import Counter def entropy(data: bytes) -> float: if not data: return 0.0 counter = Counter(data) probs = [count / len(data) for count in counter.values()] return -sum(p * math.log2(p) for p in probs) # Entropy thresholds: # < 6.0: Likely plaintext or structured data # 6.0-7.5: Possibly compressed # > 7.5: Likely encrypted or random # Common encryption indicators # - High, uniform entropy # - No obvious structure or patterns # - Length often multiple of block size (16 for AES) # - Possible IV at start (16 bytes for AES-CBC) ``` ### TLS Analysis ```bash # Extract TLS metadata tshark -r capture.pcap -Y "ssl.handshake" \ -T fields -e ip.src -e ssl.handshake.ciphersuite # JA3 fingerprinting (client) tshark -r capture.pcap -Y "ssl.handshake.type == 1" \ -T fields -e ssl.handshake.ja3 # JA3S fingerprinting (server) tshark -r capture.pcap -Y "ssl.handshake.type == 2" \ -T fields -e ssl.handshake.ja3s # Certificate extraction tshark -r capture.pcap -Y "ssl.handshake.certificate" \ -T fields -e x509sat.printableString ``` ### Decryption Approaches ```bash # Pre-master secret log (browser) export SSLKEYLOGFILE=/tmp/keys.log # Configure Wireshark # Edit > Preferences > Protocols > TLS # (Pre)-Master-Secret log filename: /tmp/keys.log # Decrypt with private key (if available) # Only works for RSA key exchange # Edit > Preferences > Protocols > TLS > RSA keys list ``` ## Custom Protocol Documentation ### Protocol Specification Template ```markdown # Protocol Name Specification ## Overview Brief description of protocol purpose and design. ## Transport - Layer: TCP/UDP - Port: XXXX - Encryption: TLS 1.2+ ## Message Format ### Header (12 bytes) | Offset | Size | Field | Description | | ------ | ---- | ------- | ----------------------- | | 0 | 4 | Magic | 0x50524F54 ("PROT") | | 4 | 2 | Version | Protocol version (1) | | 6 | 2 | Type | Message type identifier | | 8 | 4 | Length | Payload length in bytes | ### Message Types | Type | Name | Description | | ---- | --------- | ---------------------- | | 0x01 | HELLO | Connection initiation | | 0x02 | HELLO_ACK | Connection accepted | | 0x03 | DATA | Application data | | 0x04 | CLOSE | Connection termination | ### Type 0x01: HELLO | Offset | Size | Field | Description | | ------ | ---- | ---------- | ------------------------ | | 0 | 4 | ClientID | Unique client identifier | | 4 | 2 | Flags | Connection flags | | 6 | var | Extensions | TLV-encoded extensions | ## State Machine ``` [INIT] --HELLO--> [WAIT_ACK] --HELLO_ACK--> [CONNECTED] | DATA/DATA | [CLOSED] <--CLOSE--+ ``` ## Examples ### Connection Establishment ``` Client -> Server: HELLO (ClientID=0x12345678) Server -> Client: HELLO_ACK (Status=OK) Client -> Server: DATA (payload) ``` ``` ### Wireshark Dissector (Lua) ```lua -- custom_protocol.lua local proto = Proto("custom", "Custom Protocol") -- Define fields local f_magic = ProtoField.string("custom.magic", "Magic") local f_version = ProtoField.uint16("custom.version", "Version") local f_type = ProtoField.uint16("custom.type", "Type") local f_length = ProtoField.uint32("custom.length", "Length") local f_payload = ProtoField.bytes("custom.payload", "Payload") proto.fields = { f_magic, f_version, f_type, f_length, f_payload } -- Message type names local msg_types = { [0x01] = "HELLO", [0x02] = "HELLO_ACK", [0x03] = "DATA", [0x04] = "CLOSE" } function proto.dissector(buffer, pinfo, tree) pinfo.cols.protocol = "CUSTOM" local subtree = tree:add(proto, buffer()) -- Parse header subtree:add(f_magic, buffer(0, 4)) subtree:add(f_version, buffer(4, 2)) local msg_type = buffer(6, 2):uint() subtree:add(f_type, buffer(6, 2)):append_text( " (" .. (msg_types[msg_type] or "Unknown") .. ")" ) local length = buffer(8, 4):uint() subtree:add(f_length, buffer(8, 4)) if length > 0 then subtree:add(f_payload, buffer(12, length)) end end -- Register for TCP port local tcp_table = DissectorTable.get("tcp.port") tcp_table:add(8888, proto) ``` ## Active Testing ### Fuzzing with Boofuzz ```python from boofuzz import * def main(): session = Session( target=Target( connection=TCPSocketConnection("target", 8888) ) ) # Define protocol structure s_initialize("HELLO") s_static(b"\x50\x52\x4f\x54") # Magic s_word(1, name="version") # Version s_word(0x01, name="type") # Type (HELLO) s_size("payload", length=4) # Length field s_block_start("payload") s_dword(0x12345678, name="client_id") s_word(0, name="flags") s_block_end() session.connect(s_get("HELLO")) session.fuzz() if __name__ == "__main__": main() ``` ### Replay and Modification ```python from scapy.all import * # Replay captured traffic packets = rdpcap("capture.pcap") for pkt in packets: if pkt.haslayer(TCP) and pkt[TCP].dport == 8888: send(pkt) # Modify and replay for pkt in packets: if pkt.haslayer(Raw): # Modify payload original = pkt[Raw].load modified = original.replace(b"client", b"CLIENT") pkt[Raw].load = modified # Recalculate checksums del pkt[IP].chksum del pkt[TCP].chksum send(pkt) ``` ## Best Practices ### Analysis Workflow 1. **Capture traffic**: Multiple sessions, different scenarios 2. **Identify boundaries**: Message start/end markers 3. **Map structure**: Fixed header, variable payload 4. **Identify fields**: Compare multiple samples 5. **Document format**: Create specification 6. **Validate understanding**: Implement parser/generator 7. **Test edge cases**: Fuzzing, boundary conditions ### Common Patterns to Look For - Magic numbers/signatures at message start - Version fields for compatibility - Length fields (often before variable data) - Type/opcode fields for message identification - Sequence numbers for ordering - Checksums/CRCs for integrity - Timestamps for timing - Session/connection identifiers
πŸ‘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

wcag-audit-patterns

Conduct WCAG 2.2 accessibility audits with automated testing,

coding
⭐1
# WCAG Audit Patterns Comprehensive guide to auditing web content against WCAG 2.2 guidelines with actionable remediation strategies. ## When to Use This Skill - Conducting accessibility audits - Fixing WCAG violations - Implementing accessible components - Preparing for accessibility lawsuits - Meeting ADA/Section 508 requirements - Achieving VPAT compliance ## Core Concepts ### 1. WCAG Conformance Levels | Level | Description | Required For | | ------- | ---------------------- | ----------------- | | **A** | Minimum accessibility | Legal baseline | | **AA** | Standard conformance | Most regulations | | **AAA** | Enhanced accessibility | Specialized needs | ### 2. POUR Principles ``` Perceivable: Can users perceive the content? Operable: Can users operate the interface? Understandable: Can users understand the content? Robust: Does it work with assistive tech? ``` ### 3. Common Violations by Impact ``` Critical (Blockers): β”œβ”€β”€ Missing alt text for functional images β”œβ”€β”€ No keyboard access to interactive elements β”œβ”€β”€ Missing form labels └── Auto-playing media without controls Serious: β”œβ”€β”€ Insufficient color contrast β”œβ”€β”€ Missing skip links β”œβ”€β”€ Inaccessible custom widgets └── Missing page titles Moderate: β”œβ”€β”€ Missing language attribute β”œβ”€β”€ Unclear link text β”œβ”€β”€ Missing landmarks └── Improper heading hierarchy ``` ## Audit Checklist ### Perceivable (Principle 1) ````markdown ## 1.1 Text Alternatives ### 1.1.1 Non-text Content (Level A) - [ ] All images have alt text - [ ] Decorative images have alt="" - [ ] Complex images have long descriptions - [ ] Icons with meaning have accessible names - [ ] CAPTCHAs have alternatives Check: ```html <!-- Good --> <img src="chart.png" alt="Sales increased 25% from Q1 to Q2" /> <img src="decorative-line.png" alt="" /> <!-- Bad --> <img src="chart.png" /> <img src="decorative-line.png" alt="decorative line" /> ``` ```` ## 1.2 Time-based Media ### 1.2.1 Audio-only and Video-only (Level A) - [ ] Audio has text transcript - [ ] Video has audio description or transcript ### 1.2.2 Captions (Level A) - [ ] All video has synchronized captions - [ ] Captions are accurate and complete - [ ] Speaker identification included ### 1.2.3 Audio Description (Level A) - [ ] Video has audio description for visual content ## 1.3 Adaptable ### 1.3.1 Info and Relationships (Level A) - [ ] Headings use proper tags (h1-h6) - [ ] Lists use ul/ol/dl - [ ] Tables have headers - [ ] Form inputs have labels - [ ] ARIA landmarks present Check: ```html <!-- Heading hierarchy --> <h1>Page Title</h1> <h2>Section</h2> <h3>Subsection</h3> <h2>Another Section</h2> <!-- Table headers --> <table> <thead> <tr> <th scope="col">Name</th> <th scope="col">Price</th> </tr> </thead> </table> ``` ### 1.3.2 Meaningful Sequence (Level A) - [ ] Reading order is logical - [ ] CSS positioning doesn't break order - [ ] Focus order matches visual order ### 1.3.3 Sensory Characteristics (Level A) - [ ] Instructions don't rely on shape/color alone - [ ] "Click the red button" β†’ "Click Submit (red button)" ## 1.4 Distinguishable ### 1.4.1 Use of Color (Level A) - [ ] Color is not only means of conveying info - [ ] Links distinguishable without color - [ ] Error states not color-only ### 1.4.3 Contrast (Minimum) (Level AA) - [ ] Text: 4.5:1 contrast ratio - [ ] Large text (18pt+): 3:1 ratio - [ ] UI components: 3:1 ratio Tools: WebAIM Contrast Checker, axe DevTools ### 1.4.4 Resize Text (Level AA) - [ ] Text resizes to 200% without loss - [ ] No horizontal scrolling at 320px - [ ] Content reflows properly ### 1.4.10 Reflow (Level AA) - [ ] Content reflows at 400% zoom - [ ] No two-dimensional scrolling - [ ] All content accessible at 320px width ### 1.4.11 Non-text Contrast (Level AA) - [ ] UI components have 3:1 contrast - [ ] Focus indicators visible - [ ] Graphical objects distinguishable ### 1.4.12 Text Spacing (Level AA) - [ ] No content loss with increased spacing - [ ] Line height 1.5x font size - [ ] Paragraph spacing 2x font size - [ ] Letter spacing 0.12x font size - [ ] Word spacing 0.16x font size ```` ### Operable (Principle 2) ```markdown ## 2.1 Keyboard Accessible ### 2.1.1 Keyboard (Level A) - [ ] All functionality keyboard accessible - [ ] No keyboard traps - [ ] Tab order is logical - [ ] Custom widgets are keyboard operable Check: ```javascript // Custom button must be keyboard accessible <div role="button" tabindex="0" onkeydown="if(event.key === 'Enter' || event.key === ' ') activate()"> ```` ### 2.1.2 No Keyboard Trap (Level A) - [ ] Focus can move away from all components - [ ] Modal dialogs trap focus correctly - [ ] Focus returns after modal closes ## 2.2 Enough Time ### 2.2.1 Timing Adjustable (Level A) - [ ] Session timeouts can be extended - [ ] User warned before timeout - [ ] Option to disable auto-refresh ### 2.2.2 Pause, Stop, Hide (Level A) - [ ] Moving content can be paused - [ ] Auto-updating content can be paused - [ ] Animations respect prefers-reduced-motion ```css @media (prefers-reduced-motion: reduce) { * { animation: none !important; transition: none !important; } } ``` ## 2.3 Seizures and Physical Reactions ### 2.3.1 Three Flashes (Level A) - [ ] No content flashes more than 3 times/second - [ ] Flashing area is small (<25% viewport) ## 2.4 Navigable ### 2.4.1 Bypass Blocks (Level A) - [ ] Skip to main content link present - [ ] Landmark regions defined - [ ] Proper heading structure ```html <a href="#main" class="skip-link">Skip to main content</a> <main id="main">...</main> ``` ### 2.4.2 Page Titled (Level A) - [ ] Unique, descriptive page titles - [ ] Title reflects page content ### 2.4.3 Focus Order (Level A) - [ ] Focus order matches visual order - [ ] tabindex used correctly ### 2.4.4 Link Purpose (In Context) (Level A) - [ ] Links make sense out of context - [ ] No "click here" or "read more" alone ```html <!-- Bad --> <a href="report.pdf">Click here</a> <!-- Good --> <a href="report.pdf">Download Q4 Sales Report (PDF)</a> ``` ### 2.4.6 Headings and Labels (Level AA) - [ ] Headings describe content - [ ] Labels describe purpose ### 2.4.7 Focus Visible (Level AA) - [ ] Focus indicator visible on all elements - [ ] Custom focus styles meet contrast ```css :focus { outline: 3px solid #005fcc; outline-offset: 2px; } ``` ### 2.4.11 Focus Not Obscured (Level AA) - WCAG 2.2 - [ ] Focused element not fully hidden - [ ] Sticky headers don't obscure focus ```` ### Understandable (Principle 3) ```markdown ## 3.1 Readable ### 3.1.1 Language of Page (Level A) - [ ] HTML lang attribute set - [ ] Language correct for content ```html <html lang="en"> ```` ### 3.1.2 Language of Parts (Level AA) - [ ] Language changes marked ```html <p>The French word <span lang="fr">bonjour</span> means hello.</p> ``` ## 3.2 Predictable ### 3.2.1 On Focus (Level A) - [ ] No context change on focus alone - [ ] No unexpected popups on focus ### 3.2.2 On Input (Level A) - [ ] No automatic form submission - [ ] User warned before context change ### 3.2.3 Consistent Navigation (Level AA) - [ ] Navigation consistent across pages - [ ] Repeated components same order ### 3.2.4 Consistent Identification (Level AA) - [ ] Same functionality = same label - [ ] Icons used consistently ## 3.3 Input Assistance ### 3.3.1 Error Identification (Level A) - [ ] Errors clearly identified - [ ] Error message describes problem - [ ] Error linked to field ```html <input aria-describedby="email-error" aria-invalid="true" /> <span id="email-error" role="alert">Please enter valid email</span> ``` ### 3.3.2 Labels or Instructions (Level A) - [ ] All inputs have visible labels - [ ] Required fields indicated - [ ] Format hints provided ### 3.3.3 Error Suggestion (Level AA) - [ ] Errors include correction suggestion - [ ] Suggestions are specific ### 3.3.4 Error Prevention (Level AA) - [ ] Legal/financial forms reversible - [ ] Data checked before submission - [ ] User can review before submit ```` ### Robust (Principle 4) ```markdown ## 4.1 Compatible ### 4.1.1 Parsing (Level A) - Obsolete in WCAG 2.2 - [ ] Valid HTML (good practice) - [ ] No duplicate IDs - [ ] Complete start/end tags ### 4.1.2 Name, Role, Value (Level A) - [ ] Custom widgets have accessible names - [ ] ARIA roles correct - [ ] State changes announced ```html <!-- Accessible custom checkbox --> <div role="checkbox" aria-checked="false" tabindex="0" aria-labelledby="label"> </div> <span id="label">Accept terms</span> ```` ### 4.1.3 Status Messages (Level AA) - [ ] Status updates announced - [ ] Live regions used correctly ```html <div role="status" aria-live="polite">3 items added to cart</div> <div role="alert" aria-live="assertive">Error: Form submission failed</div> ``` ```` ## Automated Testing ```javascript // axe-core integration const axe = require('axe-core'); async function runAccessibilityAudit(page) { await page.addScriptTag({ path: require.resolve('axe-core') }); const results = await page.evaluate(async () => { return await axe.run(document, { runOnly: { type: 'tag', values: ['wcag2a', 'wcag2aa', 'wcag21aa', 'wcag22aa'] } }); }); return { violations: results.violations, passes: results.passes, incomplete: results.incomplete }; } // Playwright test example test('should have no accessibility violations', async ({ page }) => { await page.goto('/'); const results = await runAccessibilityAudit(page); expect(results.violations).toHaveLength(0); }); ```` ```bash # CLI tools npx @axe-core/cli https://example.com npx pa11y https://example.com lighthouse https://example.com --only-categories=accessibility ``` ## Remediation Patterns ### Fix: Missing Form Labels ```html <!-- Before --> <input type="email" placeholder="Email" /> <!-- After: Option 1 - Visible label --> <label for="email">Email address</label> <input id="email" type="email" /> <!-- After: Option 2 - aria-label --> <input type="email" aria-label="Email address" /> <!-- After: Option 3 - aria-labelledby --> <span id="email-label">Email</span> <input type="email" aria-labelledby="email-label" /> ``` ### Fix: Insufficient Color Contrast ```css /* Before: 2.5:1 contrast */ .text { color: #767676; } /* After: 4.5:1 contrast */ .text { color: #595959; } /* Or add background */ .text { color: #767676; background: #000; } ``` ### Fix: Keyboard Navigation ```javascript // Make custom element keyboard accessible class AccessibleDropdown extends HTMLElement { connectedCallback() { this.setAttribute("tabindex", "0"); this.setAttribute("role", "combobox"); this.setAttribute("aria-expanded", "false"); this.addEventListener("keydown", (e) => { switch (e.key) { case "Enter": case " ": this.toggle(); e.preventDefault(); break; case "Escape": this.close(); break; case "ArrowDown": this.focusNext(); e.preventDefault(); break; case "ArrowUp": this.focusPrevious(); e.preventDefault(); break; } }); } } ``` ## Best Practices ### Do's - **Start early** - Accessibility from design phase - **Test with real users** - Disabled users provide best feedback - **Automate what you can** - 30-50% issues detectable - **Use semantic HTML** - Reduces ARIA needs - **Document patterns** - Build accessible component library ### Don'ts - **Don't rely only on automated testing** - Manual testing required - **Don't use ARIA as first solution** - Native HTML first - **Don't hide focus outlines** - Keyboard users need them - **Don't disable zoom** - Users need to resize - **Don't use color alone** - Multiple indicators needed ## Resources - [WCAG 2.2 Guidelines](https://www.w3.org/TR/WCAG22/) - [WebAIM](https://webaim.org/) - [A11y Project Checklist](https://www.a11yproject.com/checklist/) - [axe DevTools](https://www.deque.com/axe/)
πŸ‘1
πŸ‘οΈ1
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

paypal-integration

Integrate PayPal payment processing with support for express

business
⭐1
# PayPal Integration Master PayPal payment integration including Express Checkout, IPN handling, recurring billing, and refund workflows. ## When to Use This Skill - Integrating PayPal as a payment option - Implementing express checkout flows - Setting up recurring billing with PayPal - Processing refunds and payment disputes - Handling PayPal webhooks (IPN) - Supporting international payments - Implementing PayPal subscriptions ## Core Concepts ### 1. Payment Products **PayPal Checkout** - One-time payments - Express checkout experience - Guest and PayPal account payments **PayPal Subscriptions** - Recurring billing - Subscription plans - Automatic renewals **PayPal Payouts** - Send money to multiple recipients - Marketplace and platform payments ### 2. Integration Methods **Client-Side (JavaScript SDK)** - Smart Payment Buttons - Hosted payment flow - Minimal backend code **Server-Side (REST API)** - Full control over payment flow - Custom checkout UI - Advanced features ### 3. IPN (Instant Payment Notification) - Webhook-like payment notifications - Asynchronous payment updates - Verification required ## Quick Start ```javascript // Frontend - PayPal Smart Buttons <div id="paypal-button-container"></div> <script src="https://www.paypal.com/sdk/js?client-id=YOUR_CLIENT_ID&currency=USD"></script> <script> paypal.Buttons({ createOrder: function(data, actions) { return actions.order.create({ purchase_units: [{ amount: { value: '25.00' } }] }); }, onApprove: function(data, actions) { return actions.order.capture().then(function(details) { // Payment successful console.log('Transaction completed by ' + details.payer.name.given_name); // Send to backend for verification fetch('/api/paypal/capture', { method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify({orderID: data.orderID}) }); }); } }).render('#paypal-button-container'); </script> ``` ```python # Backend - Verify and capture order from paypalrestsdk import Payment import paypalrestsdk paypalrestsdk.configure({ "mode": "sandbox", # or "live" "client_id": "YOUR_CLIENT_ID", "client_secret": "YOUR_CLIENT_SECRET" }) def capture_paypal_order(order_id): """Capture a PayPal order.""" payment = Payment.find(order_id) if payment.execute({"payer_id": payment.payer.payer_info.payer_id}): # Payment successful return { 'status': 'success', 'transaction_id': payment.id, 'amount': payment.transactions[0].amount.total } else: # Payment failed return { 'status': 'failed', 'error': payment.error } ``` ## Express Checkout Implementation ### Server-Side Order Creation ```python import requests import json class PayPalClient: def __init__(self, client_id, client_secret, mode='sandbox'): self.client_id = client_id self.client_secret = client_secret self.base_url = 'https://api-m.sandbox.paypal.com' if mode == 'sandbox' else 'https://api-m.paypal.com' self.access_token = self.get_access_token() def get_access_token(self): """Get OAuth access token.""" url = f"{self.base_url}/v1/oauth2/token" headers = {"Accept": "application/json", "Accept-Language": "en_US"} response = requests.post( url, headers=headers, data={"grant_type": "client_credentials"}, auth=(self.client_id, self.client_secret) ) return response.json()['access_token'] def create_order(self, amount, currency='USD'): """Create a PayPal order.""" url = f"{self.base_url}/v2/checkout/orders" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.access_token}" } payload = { "intent": "CAPTURE", "purchase_units": [{ "amount": { "currency_code": currency, "value": str(amount) } }] } response = requests.post(url, headers=headers, json=payload) return response.json() def capture_order(self, order_id): """Capture payment for an order.""" url = f"{self.base_url}/v2/checkout/orders/{order_id}/capture" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.access_token}" } response = requests.post(url, headers=headers) return response.json() def get_order_details(self, order_id): """Get order details.""" url = f"{self.base_url}/v2/checkout/orders/{order_id}" headers = { "Authorization": f"Bearer {self.access_token}" } response = requests.get(url, headers=headers) return response.json() ``` ## IPN (Instant Payment Notification) Handling ### IPN Verification and Processing ```python from flask import Flask, request import requests from urllib.parse import parse_qs app = Flask(__name__) @app.route('/ipn', methods=['POST']) def handle_ipn(): """Handle PayPal IPN notifications.""" # Get IPN message ipn_data = request.form.to_dict() # Verify IPN with PayPal if not verify_ipn(ipn_data): return 'IPN verification failed', 400 # Process IPN based on transaction type payment_status = ipn_data.get('payment_status') txn_type = ipn_data.get('txn_type') if payment_status == 'Completed': handle_payment_completed(ipn_data) elif payment_status == 'Refunded': handle_refund(ipn_data) elif payment_status == 'Reversed': handle_chargeback(ipn_data) return 'IPN processed', 200 def verify_ipn(ipn_data): """Verify IPN message authenticity.""" # Add 'cmd' parameter verify_data = ipn_data.copy() verify_data['cmd'] = '_notify-validate' # Send back to PayPal for verification paypal_url = 'https://ipnpb.sandbox.paypal.com/cgi-bin/webscr' # or production URL response = requests.post(paypal_url, data=verify_data) return response.text == 'VERIFIED' def handle_payment_completed(ipn_data): """Process completed payment.""" txn_id = ipn_data.get('txn_id') payer_email = ipn_data.get('payer_email') mc_gross = ipn_data.get('mc_gross') item_name = ipn_data.get('item_name') # Check if already processed (prevent duplicates) if is_transaction_processed(txn_id): return # Update database # Send confirmation email # Fulfill order print(f"Payment completed: {txn_id}, Amount: ${mc_gross}") def handle_refund(ipn_data): """Handle refund.""" parent_txn_id = ipn_data.get('parent_txn_id') mc_gross = ipn_data.get('mc_gross') # Process refund in your system print(f"Refund processed: {parent_txn_id}, Amount: ${mc_gross}") def handle_chargeback(ipn_data): """Handle payment reversal/chargeback.""" txn_id = ipn_data.get('txn_id') reason_code = ipn_data.get('reason_code') # Handle chargeback print(f"Chargeback: {txn_id}, Reason: {reason_code}") ``` ## Subscription/Recurring Billing ### Create Subscription Plan ```python def create_subscription_plan(name, amount, interval='MONTH'): """Create a subscription plan.""" client = PayPalClient(CLIENT_ID, CLIENT_SECRET) url = f"{client.base_url}/v1/billing/plans" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {client.access_token}" } payload = { "product_id": "PRODUCT_ID", # Create product first "name": name, "billing_cycles": [{ "frequency": { "interval_unit": interval, "interval_count": 1 }, "tenure_type": "REGULAR", "sequence": 1, "total_cycles": 0, # Infinite "pricing_scheme": { "fixed_price": { "value": str(amount), "currency_code": "USD" } } }], "payment_preferences": { "auto_bill_outstanding": True, "setup_fee": { "value": "0", "currency_code": "USD" }, "setup_fee_failure_action": "CONTINUE", "payment_failure_threshold": 3 } } response = requests.post(url, headers=headers, json=payload) return response.json() def create_subscription(plan_id, subscriber_email): """Create a subscription for a customer.""" client = PayPalClient(CLIENT_ID, CLIENT_SECRET) url = f"{client.base_url}/v1/billing/subscriptions" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {client.access_token}" } payload = { "plan_id": plan_id, "subscriber": { "email_address": subscriber_email }, "application_context": { "return_url": "https://yourdomain.com/subscription/success", "cancel_url": "https://yourdomain.com/subscription/cancel" } } response = requests.post(url, headers=headers, json=payload) subscription = response.json() # Get approval URL for link in subscription.get('links', []): if link['rel'] == 'approve': return { 'subscription_id': subscription['id'], 'approval_url': link['href'] } ``` ## Refund Workflows ```python def create_refund(capture_id, amount=None, note=None): """Create a refund for a captured payment.""" client = PayPalClient(CLIENT_ID, CLIENT_SECRET) url = f"{client.base_url}/v2/payments/captures/{capture_id}/refund" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {client.access_token}" } payload = {} if amount: payload["amount"] = { "value": str(amount), "currency_code": "USD" } if note: payload["note_to_payer"] = note response = requests.post(url, headers=headers, json=payload) return response.json() def get_refund_details(refund_id): """Get refund details.""" client = PayPalClient(CLIENT_ID, CLIENT_SECRET) url = f"{client.base_url}/v2/payments/refunds/{refund_id}" headers = { "Authorization": f"Bearer {client.access_token}" } response = requests.get(url, headers=headers) return response.json() ``` ## Error Handling ```python class PayPalError(Exception): """Custom PayPal error.""" pass def handle_paypal_api_call(api_function): """Wrapper for PayPal API calls with error handling.""" try: result = api_function() return result except requests.exceptions.RequestException as e: # Network error raise PayPalError(f"Network error: {str(e)}") except Exception as e: # Other errors raise PayPalError(f"PayPal API error: {str(e)}") # Usage try: order = handle_paypal_api_call(lambda: client.create_order(25.00)) except PayPalError as e: # Handle error appropriately log_error(e) ``` ## Testing ```python # Use sandbox credentials SANDBOX_CLIENT_ID = "..." SANDBOX_SECRET = "..." # Test accounts # Create test buyer and seller accounts at developer.paypal.com def test_payment_flow(): """Test complete payment flow.""" client = PayPalClient(SANDBOX_CLIENT_ID, SANDBOX_SECRET, mode='sandbox') # Create order order = client.create_order(10.00) assert 'id' in order # Get approval URL approval_url = next((link['href'] for link in order['links'] if link['rel'] == 'approve'), None) assert approval_url is not None # After approval (manual step with test account) # Capture order # captured = client.capture_order(order['id']) # assert captured['status'] == 'COMPLETED' ``` ## Resources - **references/express-checkout.md**: Express Checkout implementation guide - **references/ipn-handling.md**: IPN verification and processing - **references/refund-workflows.md**: Refund handling patterns - **references/billing-agreements.md**: Recurring billing setup - **assets/paypal-client.py**: Production PayPal client - **assets/ipn-processor.py**: IPN webhook processor - **assets/recurring-billing.py**: Subscription management ## Best Practices 1. **Always Verify IPN**: Never trust IPN without verification 2. **Idempotent Processing**: Handle duplicate IPN notifications 3. **Error Handling**: Implement robust error handling 4. **Logging**: Log all transactions and errors 5. **Test Thoroughly**: Use sandbox extensively 6. **Webhook Backup**: Don't rely solely on client-side callbacks 7. **Currency Handling**: Always specify currency explicitly ## Common Pitfalls - **Not Verifying IPN**: Accepting IPN without verification - **Duplicate Processing**: Not checking for duplicate transactions - **Wrong Environment**: Mixing sandbox and production URLs/credentials - **Missing Webhooks**: Not handling all payment states - **Hardcoded Values**: Not making configurable for different environments
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

saga-orchestration

Implement saga patterns for distributed transactions and

coding
⭐1
# Saga Orchestration Patterns for managing distributed transactions and long-running business processes. ## When to Use This Skill - Coordinating multi-service transactions - Implementing compensating transactions - Managing long-running business workflows - Handling failures in distributed systems - Building order fulfillment processes - Implementing approval workflows ## Core Concepts ### 1. Saga Types ``` Choreography Orchestration β”Œβ”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚Svc A│─►│Svc B│─►│Svc Cβ”‚ β”‚ Orchestratorβ”‚ β””β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”˜ β”‚ β”‚ β”‚ β”‚ β–Ό β–Ό β–Ό β”Œβ”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β” Event Event Event β–Ό β–Ό β–Ό β”Œβ”€β”€β”€β”€β”β”Œβ”€β”€β”€β”€β”β”Œβ”€β”€β”€β”€β” β”‚Svc1β”‚β”‚Svc2β”‚β”‚Svc3β”‚ β””β”€β”€β”€β”€β”˜β””β”€β”€β”€β”€β”˜β””β”€β”€β”€β”€β”˜ ``` ### 2. Saga Execution States | State | Description | | ---------------- | ------------------------------ | | **Started** | Saga initiated | | **Pending** | Waiting for step completion | | **Compensating** | Rolling back due to failure | | **Completed** | All steps succeeded | | **Failed** | Saga failed after compensation | ## Templates ### Template 1: Saga Orchestrator Base ```python from abc import ABC, abstractmethod from dataclasses import dataclass, field from enum import Enum from typing import List, Dict, Any, Optional from datetime import datetime import uuid class SagaState(Enum): STARTED = "started" PENDING = "pending" COMPENSATING = "compensating" COMPLETED = "completed" FAILED = "failed" @dataclass class SagaStep: name: str action: str compensation: str status: str = "pending" result: Optional[Dict] = None error: Optional[str] = None executed_at: Optional[datetime] = None compensated_at: Optional[datetime] = None @dataclass class Saga: saga_id: str saga_type: str state: SagaState data: Dict[str, Any] steps: List[SagaStep] current_step: int = 0 created_at: datetime = field(default_factory=datetime.utcnow) updated_at: datetime = field(default_factory=datetime.utcnow) class SagaOrchestrator(ABC): """Base class for saga orchestrators.""" def __init__(self, saga_store, event_publisher): self.saga_store = saga_store self.event_publisher = event_publisher @abstractmethod def define_steps(self, data: Dict) -> List[SagaStep]: """Define the saga steps.""" pass @property @abstractmethod def saga_type(self) -> str: """Unique saga type identifier.""" pass async def start(self, data: Dict) -> Saga: """Start a new saga.""" saga = Saga( saga_id=str(uuid.uuid4()), saga_type=self.saga_type, state=SagaState.STARTED, data=data, steps=self.define_steps(data) ) await self.saga_store.save(saga) await self._execute_next_step(saga) return saga async def handle_step_completed(self, saga_id: str, step_name: str, result: Dict): """Handle successful step completion.""" saga = await self.saga_store.get(saga_id) # Update step for step in saga.steps: if step.name == step_name: step.status = "completed" step.result = result step.executed_at = datetime.utcnow() break saga.current_step += 1 saga.updated_at = datetime.utcnow() # Check if saga is complete if saga.current_step >= len(saga.steps): saga.state = SagaState.COMPLETED await self.saga_store.save(saga) await self._on_saga_completed(saga) else: saga.state = SagaState.PENDING await self.saga_store.save(saga) await self._execute_next_step(saga) async def handle_step_failed(self, saga_id: str, step_name: str, error: str): """Handle step failure - start compensation.""" saga = await self.saga_store.get(saga_id) # Mark step as failed for step in saga.steps: if step.name == step_name: step.status = "failed" step.error = error break saga.state = SagaState.COMPENSATING saga.updated_at = datetime.utcnow() await self.saga_store.save(saga) # Start compensation from current step backwards await self._compensate(saga) async def _execute_next_step(self, saga: Saga): """Execute the next step in the saga.""" if saga.current_step >= len(saga.steps): return step = saga.steps[saga.current_step] step.status = "executing" await self.saga_store.save(saga) # Publish command to execute step await self.event_publisher.publish( step.action, { "saga_id": saga.saga_id, "step_name": step.name, **saga.data } ) async def _compensate(self, saga: Saga): """Execute compensation for completed steps.""" # Compensate in reverse order for i in range(saga.current_step - 1, -1, -1): step = saga.steps[i] if step.status == "completed": step.status = "compensating" await self.saga_store.save(saga) await self.event_publisher.publish( step.compensation, { "saga_id": saga.saga_id, "step_name": step.name, "original_result": step.result, **saga.data } ) async def handle_compensation_completed(self, saga_id: str, step_name: str): """Handle compensation completion.""" saga = await self.saga_store.get(saga_id) for step in saga.steps: if step.name == step_name: step.status = "compensated" step.compensated_at = datetime.utcnow() break # Check if all compensations complete all_compensated = all( s.status in ("compensated", "pending", "failed") for s in saga.steps ) if all_compensated: saga.state = SagaState.FAILED await self._on_saga_failed(saga) await self.saga_store.save(saga) async def _on_saga_completed(self, saga: Saga): """Called when saga completes successfully.""" await self.event_publisher.publish( f"{self.saga_type}Completed", {"saga_id": saga.saga_id, **saga.data} ) async def _on_saga_failed(self, saga: Saga): """Called when saga fails after compensation.""" await self.event_publisher.publish( f"{self.saga_type}Failed", {"saga_id": saga.saga_id, "error": "Saga failed", **saga.data} ) ``` ### Template 2: Order Fulfillment Saga ```python class OrderFulfillmentSaga(SagaOrchestrator): """Orchestrates order fulfillment across services.""" @property def saga_type(self) -> str: return "OrderFulfillment" def define_steps(self, data: Dict) -> List[SagaStep]: return [ SagaStep( name="reserve_inventory", action="InventoryService.ReserveItems", compensation="InventoryService.ReleaseReservation" ), SagaStep( name="process_payment", action="PaymentService.ProcessPayment", compensation="PaymentService.RefundPayment" ), SagaStep( name="create_shipment", action="ShippingService.CreateShipment", compensation="ShippingService.CancelShipment" ), SagaStep( name="send_confirmation", action="NotificationService.SendOrderConfirmation", compensation="NotificationService.SendCancellationNotice" ) ] # Usage async def create_order(order_data: Dict): saga = OrderFulfillmentSaga(saga_store, event_publisher) return await saga.start({ "order_id": order_data["order_id"], "customer_id": order_data["customer_id"], "items": order_data["items"], "payment_method": order_data["payment_method"], "shipping_address": order_data["shipping_address"] }) # Event handlers in each service class InventoryService: async def handle_reserve_items(self, command: Dict): try: # Reserve inventory reservation = await self.reserve( command["items"], command["order_id"] ) # Report success await self.event_publisher.publish( "SagaStepCompleted", { "saga_id": command["saga_id"], "step_name": "reserve_inventory", "result": {"reservation_id": reservation.id} } ) except InsufficientInventoryError as e: await self.event_publisher.publish( "SagaStepFailed", { "saga_id": command["saga_id"], "step_name": "reserve_inventory", "error": str(e) } ) async def handle_release_reservation(self, command: Dict): # Compensating action await self.release_reservation( command["original_result"]["reservation_id"] ) await self.event_publisher.publish( "SagaCompensationCompleted", { "saga_id": command["saga_id"], "step_name": "reserve_inventory" } ) ``` ### Template 3: Choreography-Based Saga ```python from dataclasses import dataclass from typing import Dict, Any import asyncio @dataclass class SagaContext: """Passed through choreographed saga events.""" saga_id: str step: int data: Dict[str, Any] completed_steps: list class OrderChoreographySaga: """Choreography-based saga using events.""" def __init__(self, event_bus): self.event_bus = event_bus self._register_handlers() def _register_handlers(self): self.event_bus.subscribe("OrderCreated", self._on_order_created) self.event_bus.subscribe("InventoryReserved", self._on_inventory_reserved) self.event_bus.subscribe("PaymentProcessed", self._on_payment_processed) self.event_bus.subscribe("ShipmentCreated", self._on_shipment_created) # Compensation handlers self.event_bus.subscribe("PaymentFailed", self._on_payment_failed) self.event_bus.subscribe("ShipmentFailed", self._on_shipment_failed) async def _on_order_created(self, event: Dict): """Step 1: Order created, reserve inventory.""" await self.event_bus.publish("ReserveInventory", { "saga_id": event["order_id"], "order_id": event["order_id"], "items": event["items"] }) async def _on_inventory_reserved(self, event: Dict): """Step 2: Inventory reserved, process payment.""" await self.event_bus.publish("ProcessPayment", { "saga_id": event["saga_id"], "order_id": event["order_id"], "amount": event["total_amount"], "reservation_id": event["reservation_id"] }) async def _on_payment_processed(self, event: Dict): """Step 3: Payment done, create shipment.""" await self.event_bus.publish("CreateShipment", { "saga_id": event["saga_id"], "order_id": event["order_id"], "payment_id": event["payment_id"] }) async def _on_shipment_created(self, event: Dict): """Step 4: Complete - send confirmation.""" await self.event_bus.publish("OrderFulfilled", { "saga_id": event["saga_id"], "order_id": event["order_id"], "tracking_number": event["tracking_number"] }) # Compensation handlers async def _on_payment_failed(self, event: Dict): """Payment failed - release inventory.""" await self.event_bus.publish("ReleaseInventory", { "saga_id": event["saga_id"], "reservation_id": event["reservation_id"] }) await self.event_bus.publish("OrderFailed", { "order_id": event["order_id"], "reason": "Payment failed" }) async def _on_shipment_failed(self, event: Dict): """Shipment failed - refund payment and release inventory.""" await self.event_bus.publish("RefundPayment", { "saga_id": event["saga_id"], "payment_id": event["payment_id"] }) await self.event_bus.publish("ReleaseInventory", { "saga_id": event["saga_id"], "reservation_id": event["reservation_id"] }) ``` ### Template 4: Saga with Timeouts ```python class TimeoutSagaOrchestrator(SagaOrchestrator): """Saga orchestrator with step timeouts.""" def __init__(self, saga_store, event_publisher, scheduler): super().__init__(saga_store, event_publisher) self.scheduler = scheduler async def _execute_next_step(self, saga: Saga): if saga.current_step >= len(saga.steps): return step = saga.steps[saga.current_step] step.status = "executing" step.timeout_at = datetime.utcnow() + timedelta(minutes=5) await self.saga_store.save(saga) # Schedule timeout check await self.scheduler.schedule( f"saga_timeout_{saga.saga_id}_{step.name}", self._check_timeout, {"saga_id": saga.saga_id, "step_name": step.name}, run_at=step.timeout_at ) await self.event_publisher.publish( step.action, {"saga_id": saga.saga_id, "step_name": step.name, **saga.data} ) async def _check_timeout(self, data: Dict): """Check if step has timed out.""" saga = await self.saga_store.get(data["saga_id"]) step = next(s for s in saga.steps if s.name == data["step_name"]) if step.status == "executing": # Step timed out - fail it await self.handle_step_failed( data["saga_id"], data["step_name"], "Step timed out" ) ``` ## Best Practices ### Do's - **Make steps idempotent** - Safe to retry - **Design compensations carefully** - They must work - **Use correlation IDs** - For tracing across services - **Implement timeouts** - Don't wait forever - **Log everything** - For debugging failures ### Don'ts - **Don't assume instant completion** - Sagas take time - **Don't skip compensation testing** - Most critical part - **Don't couple services** - Use async messaging - **Don't ignore partial failures** - Handle gracefully ## Resources - [Saga Pattern](https://microservices.io/patterns/data/saga.html) - [Designing Data-Intensive Applications](https://dataintensive.net/)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered