Skip to main content
EVOKORE// BROWSE / SECURITY & AUDITING / SYSTEM PROMPTS

Security & Auditing / System prompts

System prompts for Security & Auditing

Browse curated system prompts for assistants, agents, tools, and durable AI workflows for security & auditing teams.

11 indexed promptsISR: 300s
>

./browse/prompts

11 NODES
πŸ€–system promptβ€’7 months ago

solidity-security

Master smart contract security best practices to prevent common

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

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

memory-forensics

Master memory forensics techniques including memory acquisition,

security
⭐1
# Memory Forensics Comprehensive techniques for acquiring, analyzing, and extracting artifacts from memory dumps for incident response and malware analysis. ## Memory Acquisition ### Live Acquisition Tools #### Windows ```powershell # WinPmem (Recommended) winpmem_mini_x64.exe memory.raw # DumpIt DumpIt.exe # Belkasoft RAM Capturer # GUI-based, outputs raw format # Magnet RAM Capture # GUI-based, outputs raw format ``` #### Linux ```bash # LiME (Linux Memory Extractor) sudo insmod lime.ko "path=/tmp/memory.lime format=lime" # /dev/mem (limited, requires permissions) sudo dd if=/dev/mem of=memory.raw bs=1M # /proc/kcore (ELF format) sudo cp /proc/kcore memory.elf ``` #### macOS ```bash # osxpmem sudo ./osxpmem -o memory.raw # MacQuisition (commercial) ``` ### Virtual Machine Memory ```bash # VMware: .vmem file is raw memory cp vm.vmem memory.raw # VirtualBox: Use debug console vboxmanage debugvm "VMName" dumpvmcore --filename memory.elf # QEMU virsh dump <domain> memory.raw --memory-only # Hyper-V # Checkpoint contains memory state ``` ## Volatility 3 Framework ### Installation and Setup ```bash # Install Volatility 3 pip install volatility3 # Install symbol tables (Windows) # Download from https://downloads.volatilityfoundation.org/volatility3/symbols/ # Basic usage vol -f memory.raw <plugin> # With symbol path vol -f memory.raw -s /path/to/symbols windows.pslist ``` ### Essential Plugins #### Process Analysis ```bash # List processes vol -f memory.raw windows.pslist # Process tree (parent-child relationships) vol -f memory.raw windows.pstree # Hidden process detection vol -f memory.raw windows.psscan # Process memory dumps vol -f memory.raw windows.memmap --pid <PID> --dump # Process environment variables vol -f memory.raw windows.envars --pid <PID> # Command line arguments vol -f memory.raw windows.cmdline ``` #### Network Analysis ```bash # Network connections vol -f memory.raw windows.netscan # Network connection state vol -f memory.raw windows.netstat ``` #### DLL and Module Analysis ```bash # Loaded DLLs per process vol -f memory.raw windows.dlllist --pid <PID> # Find hidden/injected DLLs vol -f memory.raw windows.ldrmodules # Kernel modules vol -f memory.raw windows.modules # Module dumps vol -f memory.raw windows.moddump --pid <PID> ``` #### Memory Injection Detection ```bash # Detect code injection vol -f memory.raw windows.malfind # VAD (Virtual Address Descriptor) analysis vol -f memory.raw windows.vadinfo --pid <PID> # Dump suspicious memory regions vol -f memory.raw windows.vadyarascan --yara-rules rules.yar ``` #### Registry Analysis ```bash # List registry hives vol -f memory.raw windows.registry.hivelist # Print registry key vol -f memory.raw windows.registry.printkey --key "Software\Microsoft\Windows\CurrentVersion\Run" # Dump registry hive vol -f memory.raw windows.registry.hivescan --dump ``` #### File System Artifacts ```bash # Scan for file objects vol -f memory.raw windows.filescan # Dump files from memory vol -f memory.raw windows.dumpfiles --pid <PID> # MFT analysis vol -f memory.raw windows.mftscan ``` ### Linux Analysis ```bash # Process listing vol -f memory.raw linux.pslist # Process tree vol -f memory.raw linux.pstree # Bash history vol -f memory.raw linux.bash # Network connections vol -f memory.raw linux.sockstat # Loaded kernel modules vol -f memory.raw linux.lsmod # Mount points vol -f memory.raw linux.mount # Environment variables vol -f memory.raw linux.envars ``` ### macOS Analysis ```bash # Process listing vol -f memory.raw mac.pslist # Process tree vol -f memory.raw mac.pstree # Network connections vol -f memory.raw mac.netstat # Kernel extensions vol -f memory.raw mac.lsmod ``` ## Analysis Workflows ### Malware Analysis Workflow ```bash # 1. Initial process survey vol -f memory.raw windows.pstree > processes.txt vol -f memory.raw windows.pslist > pslist.txt # 2. Network connections vol -f memory.raw windows.netscan > network.txt # 3. Detect injection vol -f memory.raw windows.malfind > malfind.txt # 4. Analyze suspicious processes vol -f memory.raw windows.dlllist --pid <PID> vol -f memory.raw windows.handles --pid <PID> # 5. Dump suspicious executables vol -f memory.raw windows.pslist --pid <PID> --dump # 6. Extract strings from dumps strings -a pid.<PID>.exe > strings.txt # 7. YARA scanning vol -f memory.raw windows.yarascan --yara-rules malware.yar ``` ### Incident Response Workflow ```bash # 1. Timeline of events vol -f memory.raw windows.timeliner > timeline.csv # 2. User activity vol -f memory.raw windows.cmdline vol -f memory.raw windows.consoles # 3. Persistence mechanisms vol -f memory.raw windows.registry.printkey \ --key "Software\Microsoft\Windows\CurrentVersion\Run" # 4. Services vol -f memory.raw windows.svcscan # 5. Scheduled tasks vol -f memory.raw windows.scheduled_tasks # 6. Recent files vol -f memory.raw windows.filescan | grep -i "recent" ``` ## Data Structures ### Windows Process Structures ```c // EPROCESS (Executive Process) typedef struct _EPROCESS { KPROCESS Pcb; // Kernel process block EX_PUSH_LOCK ProcessLock; LARGE_INTEGER CreateTime; LARGE_INTEGER ExitTime; // ... LIST_ENTRY ActiveProcessLinks; // Doubly-linked list ULONG_PTR UniqueProcessId; // PID // ... PEB* Peb; // Process Environment Block // ... } EPROCESS; // PEB (Process Environment Block) typedef struct _PEB { BOOLEAN InheritedAddressSpace; BOOLEAN ReadImageFileExecOptions; BOOLEAN BeingDebugged; // Anti-debug check // ... PVOID ImageBaseAddress; // Base address of executable PPEB_LDR_DATA Ldr; // Loader data (DLL list) PRTL_USER_PROCESS_PARAMETERS ProcessParameters; // ... } PEB; ``` ### VAD (Virtual Address Descriptor) ```c typedef struct _MMVAD { MMVAD_SHORT Core; union { ULONG LongFlags; MMVAD_FLAGS VadFlags; } u; // ... PVOID FirstPrototypePte; PVOID LastContiguousPte; // ... PFILE_OBJECT FileObject; } MMVAD; // Memory protection flags #define PAGE_EXECUTE 0x10 #define PAGE_EXECUTE_READ 0x20 #define PAGE_EXECUTE_READWRITE 0x40 #define PAGE_EXECUTE_WRITECOPY 0x80 ``` ## Detection Patterns ### Process Injection Indicators ```python # Malfind indicators # - PAGE_EXECUTE_READWRITE protection (suspicious) # - MZ header in non-image VAD region # - Shellcode patterns at allocation start # Common injection techniques # 1. Classic DLL Injection # - VirtualAllocEx + WriteProcessMemory + CreateRemoteThread # 2. Process Hollowing # - CreateProcess (SUSPENDED) + NtUnmapViewOfSection + WriteProcessMemory # 3. APC Injection # - QueueUserAPC targeting alertable threads # 4. Thread Execution Hijacking # - SuspendThread + SetThreadContext + ResumeThread ``` ### Rootkit Detection ```bash # Compare process lists vol -f memory.raw windows.pslist > pslist.txt vol -f memory.raw windows.psscan > psscan.txt diff pslist.txt psscan.txt # Hidden processes # Check for DKOM (Direct Kernel Object Manipulation) vol -f memory.raw windows.callbacks # Detect hooked functions vol -f memory.raw windows.ssdt # System Service Descriptor Table # Driver analysis vol -f memory.raw windows.driverscan vol -f memory.raw windows.driverirp ``` ### Credential Extraction ```bash # Dump hashes (requires hivelist first) vol -f memory.raw windows.hashdump # LSA secrets vol -f memory.raw windows.lsadump # Cached domain credentials vol -f memory.raw windows.cachedump # Mimikatz-style extraction # Requires specific plugins/tools ``` ## YARA Integration ### Writing Memory YARA Rules ```yara rule Suspicious_Injection { meta: description = "Detects common injection shellcode" strings: // Common shellcode patterns $mz = { 4D 5A } $shellcode1 = { 55 8B EC 83 EC } // Function prologue $api_hash = { 68 ?? ?? ?? ?? 68 ?? ?? ?? ?? E8 } // Push hash, call condition: $mz at 0 or any of ($shellcode*) } rule Cobalt_Strike_Beacon { meta: description = "Detects Cobalt Strike beacon in memory" strings: $config = { 00 01 00 01 00 02 } $sleep = "sleeptime" $beacon = "%s (admin)" wide condition: 2 of them } ``` ### Scanning Memory ```bash # Scan all process memory vol -f memory.raw windows.yarascan --yara-rules rules.yar # Scan specific process vol -f memory.raw windows.yarascan --yara-rules rules.yar --pid 1234 # Scan kernel memory vol -f memory.raw windows.yarascan --yara-rules rules.yar --kernel ``` ## String Analysis ### Extracting Strings ```bash # Basic string extraction strings -a memory.raw > all_strings.txt # Unicode strings strings -el memory.raw >> all_strings.txt # Targeted extraction from process dump vol -f memory.raw windows.memmap --pid 1234 --dump strings -a pid.1234.dmp > process_strings.txt # Pattern matching grep -E "(https?://|[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})" all_strings.txt ``` ### FLOSS for Obfuscated Strings ```bash # FLOSS extracts obfuscated strings floss malware.exe > floss_output.txt # From memory dump floss pid.1234.dmp ``` ## Best Practices ### Acquisition Best Practices 1. **Minimize footprint**: Use lightweight acquisition tools 2. **Document everything**: Record time, tool, and hash of capture 3. **Verify integrity**: Hash memory dump immediately after capture 4. **Chain of custody**: Maintain proper forensic handling ### Analysis Best Practices 1. **Start broad**: Get overview before deep diving 2. **Cross-reference**: Use multiple plugins for same data 3. **Timeline correlation**: Correlate memory findings with disk/network 4. **Document findings**: Keep detailed notes and screenshots 5. **Validate results**: Verify findings through multiple methods ### Common Pitfalls - **Stale data**: Memory is volatile, analyze promptly - **Incomplete dumps**: Verify dump size matches expected RAM - **Symbol issues**: Ensure correct symbol files for OS version - **Smear**: Memory may change during acquisition - **Encryption**: Some data may be encrypted in memory
πŸ‘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

attack-tree-construction

Build comprehensive attack trees to visualize threat paths. Use

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

sast-configuration

Configure Static Application Security Testing (SAST) tools for

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

security-requirement-extraction

Derive security requirements from threat models and business

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

stride-analysis-patterns

Apply STRIDE methodology to systematically identify threats. Use

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

threat-mitigation-mapping

Map identified threats to appropriate security controls and

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

Human-in-the-Loop Approval Token Workflow

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

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