Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

19 NODES
šŸ¤–system prompt•7 months ago

parallel-debugging

Debug complex issues using competing hypotheses with parallel

coding
⭐1
# Parallel Debugging Framework for debugging complex issues using the Analysis of Competing Hypotheses (ACH) methodology with parallel agent investigation. ## When to Use This Skill - Bug has multiple plausible root causes - Initial debugging attempts haven't identified the issue - Issue spans multiple modules or components - Need systematic root cause analysis with evidence - Want to avoid confirmation bias in debugging ## Hypothesis Generation Framework Generate hypotheses across 6 failure mode categories: ### 1. Logic Error - Incorrect conditional logic (wrong operator, missing case) - Off-by-one errors in loops or array access - Missing edge case handling - Incorrect algorithm implementation ### 2. Data Issue - Invalid or unexpected input data - Type mismatch or coercion error - Null/undefined/None where value expected - Encoding or serialization problem - Data truncation or overflow ### 3. State Problem - Race condition between concurrent operations - Stale cache returning outdated data - Incorrect initialization or default values - Unintended mutation of shared state - State machine transition error ### 4. Integration Failure - API contract violation (request/response mismatch) - Version incompatibility between components - Configuration mismatch between environments - Missing or incorrect environment variables - Network timeout or connection failure ### 5. Resource Issue - Memory leak causing gradual degradation - Connection pool exhaustion - File descriptor or handle leak - Disk space or quota exceeded - CPU saturation from inefficient processing ### 6. Environment - Missing runtime dependency - Wrong library or framework version - Platform-specific behavior difference - Permission or access control issue - Timezone or locale-related behavior ## Evidence Collection Standards ### What Constitutes Evidence | Evidence Type | Strength | Example | | ----------------- | -------- | --------------------------------------------------------------- | | **Direct** | Strong | Code at `file.ts:42` shows `if (x > 0)` should be `if (x >= 0)` | | **Correlational** | Medium | Error rate increased after commit `abc123` | | **Testimonial** | Weak | "It works on my machine" | | **Absence** | Variable | No null check found in the code path | ### Citation Format Always cite evidence with file:line references: ``` **Evidence**: The validation function at `src/validators/user.ts:87` does not check for empty strings, only null/undefined. This allows empty email addresses to pass validation. ``` ### Confidence Levels | Level | Criteria | | ------------------- | ----------------------------------------------------------------------------------- | | **High (>80%)** | Multiple direct evidence pieces, clear causal chain, no contradicting evidence | | **Medium (50-80%)** | Some direct evidence, plausible causal chain, minor ambiguities | | **Low (<50%)** | Mostly correlational evidence, incomplete causal chain, some contradicting evidence | ## Result Arbitration Protocol After all investigators report: ### Step 1: Categorize Results - **Confirmed**: High confidence, strong evidence, clear causal chain - **Plausible**: Medium confidence, some evidence, reasonable causal chain - **Falsified**: Evidence contradicts the hypothesis - **Inconclusive**: Insufficient evidence to confirm or falsify ### Step 2: Compare Confirmed Hypotheses If multiple hypotheses are confirmed, rank by: 1. Confidence level 2. Number of supporting evidence pieces 3. Strength of causal chain 4. Absence of contradicting evidence ### Step 3: Determine Root Cause - If one hypothesis clearly dominates: declare as root cause - If multiple hypotheses are equally likely: may be compound issue (multiple contributing causes) - If no hypotheses confirmed: generate new hypotheses based on evidence gathered ### Step 4: Validate Fix Before declaring the bug fixed: - [ ] Fix addresses the identified root cause - [ ] Fix doesn't introduce new issues - [ ] Original reproduction case no longer fails - [ ] Related edge cases are covered - [ ] Relevant tests are added or updated
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

team-communication-protocols

Structured messaging protocols for agent team communication

coding
⭐1
# Team Communication Protocols Protocols for effective communication between agent teammates, including message type selection, plan approval workflows, shutdown procedures, and common anti-patterns to avoid. ## When to Use This Skill - Establishing communication norms for a new team - Choosing between message types (message, broadcast, shutdown_request) - Handling plan approval workflows - Managing graceful team shutdown - Discovering teammate identities and capabilities ## Message Type Selection ### `message` (Direct Message) — Default Choice Send to a single specific teammate: ```json { "type": "message", "recipient": "implementer-1", "content": "Your API endpoint is ready. You can now build the frontend form.", "summary": "API endpoint ready for frontend" } ``` **Use for**: Task updates, coordination, questions, integration notifications. ### `broadcast` — Use Sparingly Send to ALL teammates simultaneously: ```json { "type": "broadcast", "content": "Critical: shared types file has been updated. Pull latest before continuing.", "summary": "Shared types updated" } ``` **Use ONLY for**: Critical blockers affecting everyone, major changes to shared resources. **Why sparingly?**: Each broadcast sends N separate messages (one per teammate), consuming API resources proportional to team size. ### `shutdown_request` — Graceful Termination Request a teammate to shut down: ```json { "type": "shutdown_request", "recipient": "reviewer-1", "content": "Review complete, shutting down team." } ``` The teammate responds with `shutdown_response` (approve or reject with reason). ## Communication Anti-Patterns | Anti-Pattern | Problem | Better Approach | | --------------------------------------- | ---------------------------------------- | -------------------------------------- | | Broadcasting routine updates | Wastes resources, noise | Direct message to affected teammate | | Sending JSON status messages | Not designed for structured data | Use TaskUpdate to update task status | | Not communicating at integration points | Teammates build against stale interfaces | Message when your interface is ready | | Micromanaging via messages | Overwhelms teammates, slows work | Check in at milestones, not every step | | Using UUIDs instead of names | Hard to read, error-prone | Always use teammate names | | Ignoring idle teammates | Wasted capacity | Assign new work or shut down | ## Plan Approval Workflow When a teammate is spawned with `plan_mode_required`: 1. Teammate creates a plan using read-only exploration tools 2. Teammate calls `ExitPlanMode` which sends a `plan_approval_request` to the lead 3. Lead reviews the plan 4. Lead responds with `plan_approval_response`: **Approve**: ```json { "type": "plan_approval_response", "request_id": "abc-123", "recipient": "implementer-1", "approve": true } ``` **Reject with feedback**: ```json { "type": "plan_approval_response", "request_id": "abc-123", "recipient": "implementer-1", "approve": false, "content": "Please add error handling for the API calls" } ``` ## Shutdown Protocol ### Graceful Shutdown Sequence 1. **Lead sends shutdown_request** to each teammate 2. **Teammate receives request** as a JSON message with `type: "shutdown_request"` 3. **Teammate responds** with `shutdown_response`: - `approve: true` — Teammate saves state and exits - `approve: false` + reason — Teammate continues working 4. **Lead handles rejections** — Wait for teammate to finish, then retry 5. **After all teammates shut down** — Call `Teammate` cleanup ### Handling Rejections If a teammate rejects shutdown: - Check their reason (usually "still working on task") - Wait for their current task to complete - Retry shutdown request - If urgent, user can force shutdown ## Teammate Discovery Find team members by reading the config file: **Location**: `~/.claude/teams/{team-name}/config.json` **Structure**: ```json { "members": [ { "name": "security-reviewer", "agentId": "uuid-here", "agentType": "team-reviewer" }, { "name": "perf-reviewer", "agentId": "uuid-here", "agentType": "team-reviewer" } ] } ``` **Always use `name`** for messaging and task assignment. Never use `agentId` directly.
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

defi-protocol-templates

Implement DeFi protocols with production-ready templates for

coding
⭐1
# DeFi Protocol Templates Production-ready templates for common DeFi protocols including staking, AMMs, governance, lending, and flash loans. ## When to Use This Skill - Building staking platforms with reward distribution - Implementing AMM (Automated Market Maker) protocols - Creating governance token systems - Developing lending/borrowing protocols - Integrating flash loan functionality - Launching yield farming platforms ## Staking Contract ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC20/IERC20.sol"; import "@openzeppelin/contracts/security/ReentrancyGuard.sol"; import "@openzeppelin/contracts/access/Ownable.sol"; contract StakingRewards is ReentrancyGuard, Ownable { IERC20 public stakingToken; IERC20 public rewardsToken; uint256 public rewardRate = 100; // Rewards per second uint256 public lastUpdateTime; uint256 public rewardPerTokenStored; mapping(address => uint256) public userRewardPerTokenPaid; mapping(address => uint256) public rewards; mapping(address => uint256) public balances; uint256 private _totalSupply; event Staked(address indexed user, uint256 amount); event Withdrawn(address indexed user, uint256 amount); event RewardPaid(address indexed user, uint256 reward); constructor(address _stakingToken, address _rewardsToken) { stakingToken = IERC20(_stakingToken); rewardsToken = IERC20(_rewardsToken); } modifier updateReward(address account) { rewardPerTokenStored = rewardPerToken(); lastUpdateTime = block.timestamp; if (account != address(0)) { rewards[account] = earned(account); userRewardPerTokenPaid[account] = rewardPerTokenStored; } _; } function rewardPerToken() public view returns (uint256) { if (_totalSupply == 0) { return rewardPerTokenStored; } return rewardPerTokenStored + ((block.timestamp - lastUpdateTime) * rewardRate * 1e18) / _totalSupply; } function earned(address account) public view returns (uint256) { return (balances[account] * (rewardPerToken() - userRewardPerTokenPaid[account])) / 1e18 + rewards[account]; } function stake(uint256 amount) external nonReentrant updateReward(msg.sender) { require(amount > 0, "Cannot stake 0"); _totalSupply += amount; balances[msg.sender] += amount; stakingToken.transferFrom(msg.sender, address(this), amount); emit Staked(msg.sender, amount); } function withdraw(uint256 amount) public nonReentrant updateReward(msg.sender) { require(amount > 0, "Cannot withdraw 0"); _totalSupply -= amount; balances[msg.sender] -= amount; stakingToken.transfer(msg.sender, amount); emit Withdrawn(msg.sender, amount); } function getReward() public nonReentrant updateReward(msg.sender) { uint256 reward = rewards[msg.sender]; if (reward > 0) { rewards[msg.sender] = 0; rewardsToken.transfer(msg.sender, reward); emit RewardPaid(msg.sender, reward); } } function exit() external { withdraw(balances[msg.sender]); getReward(); } } ``` ## AMM (Automated Market Maker) ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC20/IERC20.sol"; contract SimpleAMM { IERC20 public token0; IERC20 public token1; uint256 public reserve0; uint256 public reserve1; uint256 public totalSupply; mapping(address => uint256) public balanceOf; event Mint(address indexed to, uint256 amount); event Burn(address indexed from, uint256 amount); event Swap(address indexed trader, uint256 amount0In, uint256 amount1In, uint256 amount0Out, uint256 amount1Out); constructor(address _token0, address _token1) { token0 = IERC20(_token0); token1 = IERC20(_token1); } function addLiquidity(uint256 amount0, uint256 amount1) external returns (uint256 shares) { token0.transferFrom(msg.sender, address(this), amount0); token1.transferFrom(msg.sender, address(this), amount1); if (totalSupply == 0) { shares = sqrt(amount0 * amount1); } else { shares = min( (amount0 * totalSupply) / reserve0, (amount1 * totalSupply) / reserve1 ); } require(shares > 0, "Shares = 0"); _mint(msg.sender, shares); _update( token0.balanceOf(address(this)), token1.balanceOf(address(this)) ); emit Mint(msg.sender, shares); } function removeLiquidity(uint256 shares) external returns (uint256 amount0, uint256 amount1) { uint256 bal0 = token0.balanceOf(address(this)); uint256 bal1 = token1.balanceOf(address(this)); amount0 = (shares * bal0) / totalSupply; amount1 = (shares * bal1) / totalSupply; require(amount0 > 0 && amount1 > 0, "Amount0 or amount1 = 0"); _burn(msg.sender, shares); _update(bal0 - amount0, bal1 - amount1); token0.transfer(msg.sender, amount0); token1.transfer(msg.sender, amount1); emit Burn(msg.sender, shares); } function swap(address tokenIn, uint256 amountIn) external returns (uint256 amountOut) { require(tokenIn == address(token0) || tokenIn == address(token1), "Invalid token"); bool isToken0 = tokenIn == address(token0); (IERC20 tokenIn_, IERC20 tokenOut, uint256 resIn, uint256 resOut) = isToken0 ? (token0, token1, reserve0, reserve1) : (token1, token0, reserve1, reserve0); tokenIn_.transferFrom(msg.sender, address(this), amountIn); // 0.3% fee uint256 amountInWithFee = (amountIn * 997) / 1000; amountOut = (resOut * amountInWithFee) / (resIn + amountInWithFee); tokenOut.transfer(msg.sender, amountOut); _update( token0.balanceOf(address(this)), token1.balanceOf(address(this)) ); emit Swap(msg.sender, isToken0 ? amountIn : 0, isToken0 ? 0 : amountIn, isToken0 ? 0 : amountOut, isToken0 ? amountOut : 0); } function _mint(address to, uint256 amount) private { balanceOf[to] += amount; totalSupply += amount; } function _burn(address from, uint256 amount) private { balanceOf[from] -= amount; totalSupply -= amount; } function _update(uint256 res0, uint256 res1) private { reserve0 = res0; reserve1 = res1; } function sqrt(uint256 y) private pure returns (uint256 z) { if (y > 3) { z = y; uint256 x = y / 2 + 1; while (x < z) { z = x; x = (y / x + x) / 2; } } else if (y != 0) { z = 1; } } function min(uint256 x, uint256 y) private pure returns (uint256) { return x <= y ? x : y; } } ``` ## Governance Token ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC20/extensions/ERC20Votes.sol"; import "@openzeppelin/contracts/access/Ownable.sol"; contract GovernanceToken is ERC20Votes, Ownable { constructor() ERC20("Governance Token", "GOV") ERC20Permit("Governance Token") { _mint(msg.sender, 1000000 * 10**decimals()); } function _afterTokenTransfer( address from, address to, uint256 amount ) internal override(ERC20Votes) { super._afterTokenTransfer(from, to, amount); } function _mint(address to, uint256 amount) internal override(ERC20Votes) { super._mint(to, amount); } function _burn(address account, uint256 amount) internal override(ERC20Votes) { super._burn(account, amount); } } contract Governor is Ownable { GovernanceToken public governanceToken; struct Proposal { uint256 id; address proposer; string description; uint256 forVotes; uint256 againstVotes; uint256 startBlock; uint256 endBlock; bool executed; mapping(address => bool) hasVoted; } uint256 public proposalCount; mapping(uint256 => Proposal) public proposals; uint256 public votingPeriod = 17280; // ~3 days in blocks uint256 public proposalThreshold = 100000 * 10**18; event ProposalCreated(uint256 indexed proposalId, address proposer, string description); event VoteCast(address indexed voter, uint256 indexed proposalId, bool support, uint256 weight); event ProposalExecuted(uint256 indexed proposalId); constructor(address _governanceToken) { governanceToken = GovernanceToken(_governanceToken); } function propose(string memory description) external returns (uint256) { require( governanceToken.getPastVotes(msg.sender, block.number - 1) >= proposalThreshold, "Proposer votes below threshold" ); proposalCount++; Proposal storage newProposal = proposals[proposalCount]; newProposal.id = proposalCount; newProposal.proposer = msg.sender; newProposal.description = description; newProposal.startBlock = block.number; newProposal.endBlock = block.number + votingPeriod; emit ProposalCreated(proposalCount, msg.sender, description); return proposalCount; } function vote(uint256 proposalId, bool support) external { Proposal storage proposal = proposals[proposalId]; require(block.number >= proposal.startBlock, "Voting not started"); require(block.number <= proposal.endBlock, "Voting ended"); require(!proposal.hasVoted[msg.sender], "Already voted"); uint256 weight = governanceToken.getPastVotes(msg.sender, proposal.startBlock); require(weight > 0, "No voting power"); proposal.hasVoted[msg.sender] = true; if (support) { proposal.forVotes += weight; } else { proposal.againstVotes += weight; } emit VoteCast(msg.sender, proposalId, support, weight); } function execute(uint256 proposalId) external { Proposal storage proposal = proposals[proposalId]; require(block.number > proposal.endBlock, "Voting not ended"); require(!proposal.executed, "Already executed"); require(proposal.forVotes > proposal.againstVotes, "Proposal failed"); proposal.executed = true; // Execute proposal logic here emit ProposalExecuted(proposalId); } } ``` ## Flash Loan ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC20/IERC20.sol"; interface IFlashLoanReceiver { function executeOperation( address asset, uint256 amount, uint256 fee, bytes calldata params ) external returns (bool); } contract FlashLoanProvider { IERC20 public token; uint256 public feePercentage = 9; // 0.09% fee event FlashLoan(address indexed borrower, uint256 amount, uint256 fee); constructor(address _token) { token = IERC20(_token); } function flashLoan( address receiver, uint256 amount, bytes calldata params ) external { uint256 balanceBefore = token.balanceOf(address(this)); require(balanceBefore >= amount, "Insufficient liquidity"); uint256 fee = (amount * feePercentage) / 10000; // Send tokens to receiver token.transfer(receiver, amount); // Execute callback require( IFlashLoanReceiver(receiver).executeOperation( address(token), amount, fee, params ), "Flash loan failed" ); // Verify repayment uint256 balanceAfter = token.balanceOf(address(this)); require(balanceAfter >= balanceBefore + fee, "Flash loan not repaid"); emit FlashLoan(receiver, amount, fee); } } // Example flash loan receiver contract FlashLoanReceiver is IFlashLoanReceiver { function executeOperation( address asset, uint256 amount, uint256 fee, bytes calldata params ) external override returns (bool) { // Decode params and execute arbitrage, liquidation, etc. // ... // Approve repayment IERC20(asset).approve(msg.sender, amount + fee); return true; } } ``` ## Resources - **references/staking.md**: Staking mechanics and reward distribution - **references/liquidity-pools.md**: AMM mathematics and pricing - **references/governance-tokens.md**: Governance and voting systems - **references/lending-protocols.md**: Lending/borrowing implementation - **references/flash-loans.md**: Flash loan security and use cases - **assets/staking-contract.sol**: Production staking template - **assets/amm-contract.sol**: Full AMM implementation - **assets/governance-token.sol**: Governance system - **assets/lending-protocol.sol**: Lending platform template ## Best Practices 1. **Use Established Libraries**: OpenZeppelin, Solmate 2. **Test Thoroughly**: Unit tests, integration tests, fuzzing 3. **Audit Before Launch**: Professional security audits 4. **Start Simple**: MVP first, add features incrementally 5. **Monitor**: Track contract health and user activity 6. **Upgradability**: Consider proxy patterns for upgrades 7. **Emergency Controls**: Pause mechanisms for critical issues ## Common DeFi Patterns - **Time-Weighted Average Price (TWAP)**: Price oracle resistance - **Liquidity Mining**: Incentivize liquidity provision - **Vesting**: Lock tokens with gradual release - **Multisig**: Require multiple signatures for critical operations - **Timelocks**: Delay execution of governance decisions
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

nft-standards

Implement NFT standards (ERC-721, ERC-1155) with proper metadata

coding
⭐1
# NFT Standards Master ERC-721 and ERC-1155 NFT standards, metadata best practices, and advanced NFT features. ## When to Use This Skill - Creating NFT collections (art, gaming, collectibles) - Implementing marketplace functionality - Building on-chain or off-chain metadata - Creating soulbound tokens (non-transferable) - Implementing royalties and revenue sharing - Developing dynamic/evolving NFTs ## ERC-721 (Non-Fungible Token Standard) ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC721/extensions/ERC721URIStorage.sol"; import "@openzeppelin/contracts/token/ERC721/extensions/ERC721Enumerable.sol"; import "@openzeppelin/contracts/access/Ownable.sol"; import "@openzeppelin/contracts/utils/Counters.sol"; contract MyNFT is ERC721URIStorage, ERC721Enumerable, Ownable { using Counters for Counters.Counter; Counters.Counter private _tokenIds; uint256 public constant MAX_SUPPLY = 10000; uint256 public constant MINT_PRICE = 0.08 ether; uint256 public constant MAX_PER_MINT = 20; constructor() ERC721("MyNFT", "MNFT") {} function mint(uint256 quantity) external payable { require(quantity > 0 && quantity <= MAX_PER_MINT, "Invalid quantity"); require(_tokenIds.current() + quantity <= MAX_SUPPLY, "Exceeds max supply"); require(msg.value >= MINT_PRICE * quantity, "Insufficient payment"); for (uint256 i = 0; i < quantity; i++) { _tokenIds.increment(); uint256 newTokenId = _tokenIds.current(); _safeMint(msg.sender, newTokenId); _setTokenURI(newTokenId, generateTokenURI(newTokenId)); } } function generateTokenURI(uint256 tokenId) internal pure returns (string memory) { // Return IPFS URI or on-chain metadata return string(abi.encodePacked("ipfs://QmHash/", Strings.toString(tokenId), ".json")); } // Required overrides function _beforeTokenTransfer( address from, address to, uint256 tokenId, uint256 batchSize ) internal override(ERC721, ERC721Enumerable) { super._beforeTokenTransfer(from, to, tokenId, batchSize); } function _burn(uint256 tokenId) internal override(ERC721, ERC721URIStorage) { super._burn(tokenId); } function tokenURI(uint256 tokenId) public view override(ERC721, ERC721URIStorage) returns (string memory) { return super.tokenURI(tokenId); } function supportsInterface(bytes4 interfaceId) public view override(ERC721, ERC721Enumerable) returns (bool) { return super.supportsInterface(interfaceId); } function withdraw() external onlyOwner { payable(owner()).transfer(address(this).balance); } } ``` ## ERC-1155 (Multi-Token Standard) ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC1155/ERC1155.sol"; import "@openzeppelin/contracts/access/Ownable.sol"; contract GameItems is ERC1155, Ownable { uint256 public constant SWORD = 1; uint256 public constant SHIELD = 2; uint256 public constant POTION = 3; mapping(uint256 => uint256) public tokenSupply; mapping(uint256 => uint256) public maxSupply; constructor() ERC1155("ipfs://QmBaseHash/{id}.json") { maxSupply[SWORD] = 1000; maxSupply[SHIELD] = 500; maxSupply[POTION] = 10000; } function mint( address to, uint256 id, uint256 amount ) external onlyOwner { require(tokenSupply[id] + amount <= maxSupply[id], "Exceeds max supply"); _mint(to, id, amount, ""); tokenSupply[id] += amount; } function mintBatch( address to, uint256[] memory ids, uint256[] memory amounts ) external onlyOwner { for (uint256 i = 0; i < ids.length; i++) { require(tokenSupply[ids[i]] + amounts[i] <= maxSupply[ids[i]], "Exceeds max supply"); tokenSupply[ids[i]] += amounts[i]; } _mintBatch(to, ids, amounts, ""); } function burn( address from, uint256 id, uint256 amount ) external { require(from == msg.sender || isApprovedForAll(from, msg.sender), "Not authorized"); _burn(from, id, amount); tokenSupply[id] -= amount; } } ``` ## Metadata Standards ### Off-Chain Metadata (IPFS) ```json { "name": "NFT #1", "description": "Description of the NFT", "image": "ipfs://QmImageHash", "attributes": [ { "trait_type": "Background", "value": "Blue" }, { "trait_type": "Rarity", "value": "Legendary" }, { "trait_type": "Power", "value": 95, "display_type": "number", "max_value": 100 } ] } ``` ### On-Chain Metadata ```solidity contract OnChainNFT is ERC721 { struct Traits { uint8 background; uint8 body; uint8 head; uint8 rarity; } mapping(uint256 => Traits) public tokenTraits; function tokenURI(uint256 tokenId) public view override returns (string memory) { Traits memory traits = tokenTraits[tokenId]; string memory json = Base64.encode( bytes( string( abi.encodePacked( '{"name": "NFT #', Strings.toString(tokenId), '",', '"description": "On-chain NFT",', '"image": "data:image/svg+xml;base64,', generateSVG(traits), '",', '"attributes": [', '{"trait_type": "Background", "value": "', Strings.toString(traits.background), '"},', '{"trait_type": "Rarity", "value": "', getRarityName(traits.rarity), '"}', ']}' ) ) ) ); return string(abi.encodePacked("data:application/json;base64,", json)); } function generateSVG(Traits memory traits) internal pure returns (string memory) { // Generate SVG based on traits return "..."; } } ``` ## Royalties (EIP-2981) ```solidity import "@openzeppelin/contracts/interfaces/IERC2981.sol"; contract NFTWithRoyalties is ERC721, IERC2981 { address public royaltyRecipient; uint96 public royaltyFee = 500; // 5% constructor() ERC721("Royalty NFT", "RNFT") { royaltyRecipient = msg.sender; } function royaltyInfo(uint256 tokenId, uint256 salePrice) external view override returns (address receiver, uint256 royaltyAmount) { return (royaltyRecipient, (salePrice * royaltyFee) / 10000); } function setRoyalty(address recipient, uint96 fee) external onlyOwner { require(fee <= 1000, "Royalty fee too high"); // Max 10% royaltyRecipient = recipient; royaltyFee = fee; } function supportsInterface(bytes4 interfaceId) public view override(ERC721, IERC165) returns (bool) { return interfaceId == type(IERC2981).interfaceId || super.supportsInterface(interfaceId); } } ``` ## Soulbound Tokens (Non-Transferable) ```solidity contract SoulboundToken is ERC721 { constructor() ERC721("Soulbound", "SBT") {} function _beforeTokenTransfer( address from, address to, uint256 tokenId, uint256 batchSize ) internal virtual override { require(from == address(0) || to == address(0), "Token is soulbound"); super._beforeTokenTransfer(from, to, tokenId, batchSize); } function mint(address to) external { uint256 tokenId = totalSupply() + 1; _safeMint(to, tokenId); } // Burn is allowed (user can destroy their SBT) function burn(uint256 tokenId) external { require(ownerOf(tokenId) == msg.sender, "Not token owner"); _burn(tokenId); } } ``` ## Dynamic NFTs ```solidity contract DynamicNFT is ERC721 { struct TokenState { uint256 level; uint256 experience; uint256 lastUpdated; } mapping(uint256 => TokenState) public tokenStates; function gainExperience(uint256 tokenId, uint256 exp) external { require(ownerOf(tokenId) == msg.sender, "Not token owner"); TokenState storage state = tokenStates[tokenId]; state.experience += exp; // Level up logic if (state.experience >= state.level * 100) { state.level++; } state.lastUpdated = block.timestamp; } function tokenURI(uint256 tokenId) public view override returns (string memory) { TokenState memory state = tokenStates[tokenId]; // Generate metadata based on current state return generateMetadata(tokenId, state); } function generateMetadata(uint256 tokenId, TokenState memory state) internal pure returns (string memory) { // Dynamic metadata generation return ""; } } ``` ## Gas-Optimized Minting (ERC721A) ```solidity import "erc721a/contracts/ERC721A.sol"; contract OptimizedNFT is ERC721A { uint256 public constant MAX_SUPPLY = 10000; uint256 public constant MINT_PRICE = 0.05 ether; constructor() ERC721A("Optimized NFT", "ONFT") {} function mint(uint256 quantity) external payable { require(_totalMinted() + quantity <= MAX_SUPPLY, "Exceeds max supply"); require(msg.value >= MINT_PRICE * quantity, "Insufficient payment"); _mint(msg.sender, quantity); } function _baseURI() internal pure override returns (string memory) { return "ipfs://QmBaseHash/"; } } ``` ## Resources - **references/erc721.md**: ERC-721 specification details - **references/erc1155.md**: ERC-1155 multi-token standard - **references/metadata-standards.md**: Metadata best practices - **references/enumeration.md**: Token enumeration patterns - **assets/erc721-contract.sol**: Production ERC-721 template - **assets/erc1155-contract.sol**: Production ERC-1155 template - **assets/metadata-schema.json**: Standard metadata format - **assets/metadata-uploader.py**: IPFS upload utility ## Best Practices 1. **Use OpenZeppelin**: Battle-tested implementations 2. **Pin Metadata**: Use IPFS with pinning service 3. **Implement Royalties**: EIP-2981 for marketplace compatibility 4. **Gas Optimization**: Use ERC721A for batch minting 5. **Reveal Mechanism**: Placeholder → reveal pattern 6. **Enumeration**: Support walletOfOwner for marketplaces 7. **Whitelist**: Merkle trees for efficient whitelisting ## Marketplace Integration - OpenSea: ERC-721/1155, metadata standards - LooksRare: Royalty enforcement - Rarible: Protocol fees, lazy minting - Blur: Gas-optimized trading
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

solidity-security

Master smart contract security best practices to prevent common

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

istio-traffic-management

Configure Istio traffic management including routing, load

architecture
⭐1
# Istio Traffic Management Comprehensive guide to Istio traffic management for production service mesh deployments. ## When to Use This Skill - Configuring service-to-service routing - Implementing canary or blue-green deployments - Setting up circuit breakers and retries - Load balancing configuration - Traffic mirroring for testing - Fault injection for chaos engineering ## Core Concepts ### 1. Traffic Management Resources | Resource | Purpose | Scope | | ------------------- | ----------------------------- | ------------- | | **VirtualService** | Route traffic to destinations | Host-based | | **DestinationRule** | Define policies after routing | Service-based | | **Gateway** | Configure ingress/egress | Cluster edge | | **ServiceEntry** | Add external services | Mesh-wide | ### 2. Traffic Flow ``` Client → Gateway → VirtualService → DestinationRule → Service (routing) (policies) (pods) ``` ## Templates ### Template 1: Basic Routing ```yaml apiVersion: networking.istio.io/v1beta1 kind: VirtualService metadata: name: reviews-route namespace: bookinfo spec: hosts: - reviews http: - match: - headers: end-user: exact: jason route: - destination: host: reviews subset: v2 - route: - destination: host: reviews subset: v1 --- apiVersion: networking.istio.io/v1beta1 kind: DestinationRule metadata: name: reviews-destination namespace: bookinfo spec: host: reviews subsets: - name: v1 labels: version: v1 - name: v2 labels: version: v2 - name: v3 labels: version: v3 ``` ### Template 2: Canary Deployment ```yaml apiVersion: networking.istio.io/v1beta1 kind: VirtualService metadata: name: my-service-canary spec: hosts: - my-service http: - route: - destination: host: my-service subset: stable weight: 90 - destination: host: my-service subset: canary weight: 10 --- apiVersion: networking.istio.io/v1beta1 kind: DestinationRule metadata: name: my-service-dr spec: host: my-service trafficPolicy: connectionPool: tcp: maxConnections: 100 http: h2UpgradePolicy: UPGRADE http1MaxPendingRequests: 100 http2MaxRequests: 1000 subsets: - name: stable labels: version: stable - name: canary labels: version: canary ``` ### Template 3: Circuit Breaker ```yaml apiVersion: networking.istio.io/v1beta1 kind: DestinationRule metadata: name: circuit-breaker spec: host: my-service trafficPolicy: connectionPool: tcp: maxConnections: 100 http: http1MaxPendingRequests: 100 http2MaxRequests: 1000 maxRequestsPerConnection: 10 maxRetries: 3 outlierDetection: consecutive5xxErrors: 5 interval: 30s baseEjectionTime: 30s maxEjectionPercent: 50 minHealthPercent: 30 ``` ### Template 4: Retry and Timeout ```yaml apiVersion: networking.istio.io/v1beta1 kind: VirtualService metadata: name: ratings-retry spec: hosts: - ratings http: - route: - destination: host: ratings timeout: 10s retries: attempts: 3 perTryTimeout: 3s retryOn: connect-failure,refused-stream,unavailable,cancelled,retriable-4xx,503 retryRemoteLocalities: true ``` ### Template 5: Traffic Mirroring ```yaml apiVersion: networking.istio.io/v1beta1 kind: VirtualService metadata: name: mirror-traffic spec: hosts: - my-service http: - route: - destination: host: my-service subset: v1 mirror: host: my-service subset: v2 mirrorPercentage: value: 100.0 ``` ### Template 6: Fault Injection ```yaml apiVersion: networking.istio.io/v1beta1 kind: VirtualService metadata: name: fault-injection spec: hosts: - ratings http: - fault: delay: percentage: value: 10 fixedDelay: 5s abort: percentage: value: 5 httpStatus: 503 route: - destination: host: ratings ``` ### Template 7: Ingress Gateway ```yaml apiVersion: networking.istio.io/v1beta1 kind: Gateway metadata: name: my-gateway spec: selector: istio: ingressgateway servers: - port: number: 443 name: https protocol: HTTPS tls: mode: SIMPLE credentialName: my-tls-secret hosts: - "*.example.com" --- apiVersion: networking.istio.io/v1beta1 kind: VirtualService metadata: name: my-vs spec: hosts: - "api.example.com" gateways: - my-gateway http: - match: - uri: prefix: /api/v1 route: - destination: host: api-service port: number: 8080 ``` ## Load Balancing Strategies ```yaml apiVersion: networking.istio.io/v1beta1 kind: DestinationRule metadata: name: load-balancing spec: host: my-service trafficPolicy: loadBalancer: simple: ROUND_ROBIN # or LEAST_CONN, RANDOM, PASSTHROUGH --- # Consistent hashing for sticky sessions apiVersion: networking.istio.io/v1beta1 kind: DestinationRule metadata: name: sticky-sessions spec: host: my-service trafficPolicy: loadBalancer: consistentHash: httpHeaderName: x-user-id # or: httpCookie, useSourceIp, httpQueryParameterName ``` ## Best Practices ### Do's - **Start simple** - Add complexity incrementally - **Use subsets** - Version your services clearly - **Set timeouts** - Always configure reasonable timeouts - **Enable retries** - But with backoff and limits - **Monitor** - Use Kiali and Jaeger for visibility ### Don'ts - **Don't over-retry** - Can cause cascading failures - **Don't ignore outlier detection** - Enable circuit breakers - **Don't mirror to production** - Mirror to test environments - **Don't skip canary** - Test with small traffic percentage first ## Debugging Commands ```bash # Check VirtualService configuration istioctl analyze # View effective routes istioctl proxy-config routes deploy/my-app -o json # Check endpoint discovery istioctl proxy-config endpoints deploy/my-app # Debug traffic istioctl proxy-config log deploy/my-app --level debug ``` ## Resources - [Istio Traffic Management](https://istio.io/latest/docs/concepts/traffic-management/) - [Virtual Service Reference](https://istio.io/latest/docs/reference/config/networking/virtual-service/) - [Destination Rule Reference](https://istio.io/latest/docs/reference/config/networking/destination-rule/)
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

service-mesh-observability

Implement comprehensive observability for service meshes including

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

workflow-patterns

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

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

turborepo-caching

Configure Turborepo for efficient monorepo builds with local and

coding
⭐1
# Turborepo Caching Production patterns for Turborepo build optimization. ## When to Use This Skill - Setting up new Turborepo projects - Configuring build pipelines - Implementing remote caching - Optimizing CI/CD performance - Migrating from other monorepo tools - Debugging cache misses ## Core Concepts ### 1. Turborepo Architecture ``` Workspace Root/ ā”œā”€ā”€ apps/ │ ā”œā”€ā”€ web/ │ │ └── package.json │ └── docs/ │ └── package.json ā”œā”€ā”€ packages/ │ ā”œā”€ā”€ ui/ │ │ └── package.json │ └── config/ │ └── package.json ā”œā”€ā”€ turbo.json └── package.json ``` ### 2. Pipeline Concepts | Concept | Description | | -------------- | -------------------------------- | | **dependsOn** | Tasks that must complete first | | **cache** | Whether to cache outputs | | **outputs** | Files to cache | | **inputs** | Files that affect cache key | | **persistent** | Long-running tasks (dev servers) | ## Templates ### Template 1: turbo.json Configuration ```json { "$schema": "https://turbo.build/schema.json", "globalDependencies": [".env", ".env.local"], "globalEnv": ["NODE_ENV", "VERCEL_URL"], "pipeline": { "build": { "dependsOn": ["^build"], "outputs": ["dist/**", ".next/**", "!.next/cache/**"], "env": ["API_URL", "NEXT_PUBLIC_*"] }, "test": { "dependsOn": ["build"], "outputs": ["coverage/**"], "inputs": ["src/**/*.tsx", "src/**/*.ts", "test/**/*.ts"] }, "lint": { "outputs": [], "cache": true }, "typecheck": { "dependsOn": ["^build"], "outputs": [] }, "dev": { "cache": false, "persistent": true }, "clean": { "cache": false } } } ``` ### Template 2: Package-Specific Pipeline ```json // apps/web/turbo.json { "$schema": "https://turbo.build/schema.json", "extends": ["//"], "pipeline": { "build": { "outputs": [".next/**", "!.next/cache/**"], "env": ["NEXT_PUBLIC_API_URL", "NEXT_PUBLIC_ANALYTICS_ID"] }, "test": { "outputs": ["coverage/**"], "inputs": ["src/**", "tests/**", "jest.config.js"] } } } ``` ### Template 3: Remote Caching with Vercel ```bash # Login to Vercel npx turbo login # Link to Vercel project npx turbo link # Run with remote cache turbo build --remote-only # CI environment variables TURBO_TOKEN=your-token TURBO_TEAM=your-team ``` ```yaml # .github/workflows/ci.yml name: CI on: push: branches: [main] pull_request: env: TURBO_TOKEN: ${{ secrets.TURBO_TOKEN }} TURBO_TEAM: ${{ vars.TURBO_TEAM }} jobs: build: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: actions/setup-node@v4 with: node-version: 20 cache: "npm" - name: Install dependencies run: npm ci - name: Build run: npx turbo build --filter='...[origin/main]' - name: Test run: npx turbo test --filter='...[origin/main]' ``` ### Template 4: Self-Hosted Remote Cache ```typescript // Custom remote cache server (Express) import express from "express"; import { createReadStream, createWriteStream } from "fs"; import { mkdir } from "fs/promises"; import { join } from "path"; const app = express(); const CACHE_DIR = "./cache"; // Get artifact app.get("/v8/artifacts/:hash", async (req, res) => { const { hash } = req.params; const team = req.query.teamId || "default"; const filePath = join(CACHE_DIR, team, hash); try { const stream = createReadStream(filePath); stream.pipe(res); } catch { res.status(404).send("Not found"); } }); // Put artifact app.put("/v8/artifacts/:hash", async (req, res) => { const { hash } = req.params; const team = req.query.teamId || "default"; const dir = join(CACHE_DIR, team); const filePath = join(dir, hash); await mkdir(dir, { recursive: true }); const stream = createWriteStream(filePath); req.pipe(stream); stream.on("finish", () => { res.json({ urls: [`${req.protocol}://${req.get("host")}/v8/artifacts/${hash}`], }); }); }); // Check artifact exists app.head("/v8/artifacts/:hash", async (req, res) => { const { hash } = req.params; const team = req.query.teamId || "default"; const filePath = join(CACHE_DIR, team, hash); try { await fs.access(filePath); res.status(200).end(); } catch { res.status(404).end(); } }); app.listen(3000); ``` ```json // turbo.json for self-hosted cache { "remoteCache": { "signature": false } } ``` ```bash # Use self-hosted cache turbo build --api="http://localhost:3000" --token="my-token" --team="my-team" ``` ### Template 5: Filtering and Scoping ```bash # Build specific package turbo build --filter=@myorg/web # Build package and its dependencies turbo build --filter=@myorg/web... # Build package and its dependents turbo build --filter=...@myorg/ui # Build changed packages since main turbo build --filter='...[origin/main]' # Build packages in directory turbo build --filter='./apps/*' # Combine filters turbo build --filter=@myorg/web --filter=@myorg/docs # Exclude package turbo build --filter='!@myorg/docs' # Include dependencies of changed turbo build --filter='...[HEAD^1]...' ``` ### Template 6: Advanced Pipeline Configuration ```json { "$schema": "https://turbo.build/schema.json", "pipeline": { "build": { "dependsOn": ["^build"], "outputs": ["dist/**"], "inputs": ["$TURBO_DEFAULT$", "!**/*.md", "!**/*.test.*"] }, "test": { "dependsOn": ["^build"], "outputs": ["coverage/**"], "inputs": ["src/**", "tests/**", "*.config.*"], "env": ["CI", "NODE_ENV"] }, "test:e2e": { "dependsOn": ["build"], "outputs": [], "cache": false }, "deploy": { "dependsOn": ["build", "test", "lint"], "outputs": [], "cache": false }, "db:generate": { "cache": false }, "db:push": { "cache": false, "dependsOn": ["db:generate"] }, "@myorg/web#build": { "dependsOn": ["^build", "@myorg/db#db:generate"], "outputs": [".next/**"], "env": ["NEXT_PUBLIC_*"] } } } ``` ### Template 7: Root package.json Setup ```json { "name": "my-turborepo", "private": true, "workspaces": ["apps/*", "packages/*"], "scripts": { "build": "turbo build", "dev": "turbo dev", "lint": "turbo lint", "test": "turbo test", "clean": "turbo clean && rm -rf node_modules", "format": "prettier --write \"**/*.{ts,tsx,md}\"", "changeset": "changeset", "version-packages": "changeset version", "release": "turbo build --filter=./packages/* && changeset publish" }, "devDependencies": { "turbo": "^1.10.0", "prettier": "^3.0.0", "@changesets/cli": "^2.26.0" }, "packageManager": "npm@10.0.0" } ``` ## Debugging Cache ```bash # Dry run to see what would run turbo build --dry-run # Verbose output with hashes turbo build --verbosity=2 # Show task graph turbo build --graph # Force no cache turbo build --force # Show cache status turbo build --summarize # Debug specific task TURBO_LOG_VERBOSITY=debug turbo build --filter=@myorg/web ``` ## Best Practices ### Do's - **Define explicit inputs** - Avoid cache invalidation - **Use workspace protocol** - `"@myorg/ui": "workspace:*"` - **Enable remote caching** - Share across CI and local - **Filter in CI** - Build only affected packages - **Cache build outputs** - Not source files ### Don'ts - **Don't cache dev servers** - Use `persistent: true` - **Don't include secrets in env** - Use runtime env vars - **Don't ignore dependsOn** - Causes race conditions - **Don't over-filter** - May miss dependencies ## Resources - [Turborepo Documentation](https://turbo.build/repo/docs) - [Caching Guide](https://turbo.build/repo/docs/core-concepts/caching) - [Remote Caching](https://turbo.build/repo/docs/core-concepts/remote-caching)
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

k8s-manifest-generator

Create production-ready Kubernetes manifests for Deployments,

architecture
⭐1
# Kubernetes Manifest Generator Step-by-step guidance for creating production-ready Kubernetes manifests including Deployments, Services, ConfigMaps, Secrets, and PersistentVolumeClaims. ## Purpose This skill provides comprehensive guidance for generating well-structured, secure, and production-ready Kubernetes manifests following cloud-native best practices and Kubernetes conventions. ## When to Use This Skill Use this skill when you need to: - Create new Kubernetes Deployment manifests - Define Service resources for network connectivity - Generate ConfigMap and Secret resources for configuration management - Create PersistentVolumeClaim manifests for stateful workloads - Follow Kubernetes best practices and naming conventions - Implement resource limits, health checks, and security contexts - Design manifests for multi-environment deployments ## Step-by-Step Workflow ### 1. Gather Requirements **Understand the workload:** - Application type (stateless/stateful) - Container image and version - Environment variables and configuration needs - Storage requirements - Network exposure requirements (internal/external) - Resource requirements (CPU, memory) - Scaling requirements - Health check endpoints **Questions to ask:** - What is the application name and purpose? - What container image and tag will be used? - Does the application need persistent storage? - What ports does the application expose? - Are there any secrets or configuration files needed? - What are the CPU and memory requirements? - Does the application need to be exposed externally? ### 2. Create Deployment Manifest **Follow this structure:** ```yaml apiVersion: apps/v1 kind: Deployment metadata: name: <app-name> namespace: <namespace> labels: app: <app-name> version: <version> spec: replicas: 3 selector: matchLabels: app: <app-name> template: metadata: labels: app: <app-name> version: <version> spec: containers: - name: <container-name> image: <image>:<tag> ports: - containerPort: <port> name: http resources: requests: memory: "256Mi" cpu: "250m" limits: memory: "512Mi" cpu: "500m" livenessProbe: httpGet: path: /health port: http initialDelaySeconds: 30 periodSeconds: 10 readinessProbe: httpGet: path: /ready port: http initialDelaySeconds: 5 periodSeconds: 5 env: - name: ENV_VAR value: "value" envFrom: - configMapRef: name: <app-name>-config - secretRef: name: <app-name>-secret ``` **Best practices to apply:** - Always set resource requests and limits - Implement both liveness and readiness probes - Use specific image tags (never `:latest`) - Apply security context for non-root users - Use labels for organization and selection - Set appropriate replica count based on availability needs **Reference:** See `references/deployment-spec.md` for detailed deployment options ### 3. Create Service Manifest **Choose the appropriate Service type:** **ClusterIP (internal only):** ```yaml apiVersion: v1 kind: Service metadata: name: <app-name> namespace: <namespace> labels: app: <app-name> spec: type: ClusterIP selector: app: <app-name> ports: - name: http port: 80 targetPort: 8080 protocol: TCP ``` **LoadBalancer (external access):** ```yaml apiVersion: v1 kind: Service metadata: name: <app-name> namespace: <namespace> labels: app: <app-name> annotations: service.beta.kubernetes.io/aws-load-balancer-type: nlb spec: type: LoadBalancer selector: app: <app-name> ports: - name: http port: 80 targetPort: 8080 protocol: TCP ``` **Reference:** See `references/service-spec.md` for service types and networking ### 4. Create ConfigMap **For application configuration:** ```yaml apiVersion: v1 kind: ConfigMap metadata: name: <app-name>-config namespace: <namespace> data: APP_MODE: production LOG_LEVEL: info DATABASE_HOST: db.example.com # For config files app.properties: | server.port=8080 server.host=0.0.0.0 logging.level=INFO ``` **Best practices:** - Use ConfigMaps for non-sensitive data only - Organize related configuration together - Use meaningful names for keys - Consider using one ConfigMap per component - Version ConfigMaps when making changes **Reference:** See `assets/configmap-template.yaml` for examples ### 5. Create Secret **For sensitive data:** ```yaml apiVersion: v1 kind: Secret metadata: name: <app-name>-secret namespace: <namespace> type: Opaque stringData: DATABASE_PASSWORD: "changeme" API_KEY: "secret-api-key" # For certificate files tls.crt: | -----BEGIN CERTIFICATE----- ... -----END CERTIFICATE----- tls.key: | -----BEGIN PRIVATE KEY----- ... -----END PRIVATE KEY----- ``` **Security considerations:** - Never commit secrets to Git in plain text - Use Sealed Secrets, External Secrets Operator, or Vault - Rotate secrets regularly - Use RBAC to limit secret access - Consider using Secret type: `kubernetes.io/tls` for TLS secrets ### 6. Create PersistentVolumeClaim (if needed) **For stateful applications:** ```yaml apiVersion: v1 kind: PersistentVolumeClaim metadata: name: <app-name>-data namespace: <namespace> spec: accessModes: - ReadWriteOnce storageClassName: gp3 resources: requests: storage: 10Gi ``` **Mount in Deployment:** ```yaml spec: template: spec: containers: - name: app volumeMounts: - name: data mountPath: /var/lib/app volumes: - name: data persistentVolumeClaim: claimName: <app-name>-data ``` **Storage considerations:** - Choose appropriate StorageClass for performance needs - Use ReadWriteOnce for single-pod access - Use ReadWriteMany for multi-pod shared storage - Consider backup strategies - Set appropriate retention policies ### 7. Apply Security Best Practices **Add security context to Deployment:** ```yaml spec: template: spec: securityContext: runAsNonRoot: true runAsUser: 1000 fsGroup: 1000 seccompProfile: type: RuntimeDefault containers: - name: app securityContext: allowPrivilegeEscalation: false readOnlyRootFilesystem: true capabilities: drop: - ALL ``` **Security checklist:** - [ ] Run as non-root user - [ ] Drop all capabilities - [ ] Use read-only root filesystem - [ ] Disable privilege escalation - [ ] Set seccomp profile - [ ] Use Pod Security Standards ### 8. Add Labels and Annotations **Standard labels (recommended):** ```yaml metadata: labels: app.kubernetes.io/name: <app-name> app.kubernetes.io/instance: <instance-name> app.kubernetes.io/version: "1.0.0" app.kubernetes.io/component: backend app.kubernetes.io/part-of: <system-name> app.kubernetes.io/managed-by: kubectl ``` **Useful annotations:** ```yaml metadata: annotations: description: "Application description" contact: "team@example.com" prometheus.io/scrape: "true" prometheus.io/port: "9090" prometheus.io/path: "/metrics" ``` ### 9. Organize Multi-Resource Manifests **File organization options:** **Option 1: Single file with `---` separator** ```yaml # app-name.yaml --- apiVersion: v1 kind: ConfigMap ... --- apiVersion: v1 kind: Secret ... --- apiVersion: apps/v1 kind: Deployment ... --- apiVersion: v1 kind: Service ... ``` **Option 2: Separate files** ``` manifests/ ā”œā”€ā”€ configmap.yaml ā”œā”€ā”€ secret.yaml ā”œā”€ā”€ deployment.yaml ā”œā”€ā”€ service.yaml └── pvc.yaml ``` **Option 3: Kustomize structure** ``` base/ ā”œā”€ā”€ kustomization.yaml ā”œā”€ā”€ deployment.yaml ā”œā”€ā”€ service.yaml └── configmap.yaml overlays/ ā”œā”€ā”€ dev/ │ └── kustomization.yaml └── prod/ └── kustomization.yaml ``` ### 10. Validate and Test **Validation steps:** ```bash # Dry-run validation kubectl apply -f manifest.yaml --dry-run=client # Server-side validation kubectl apply -f manifest.yaml --dry-run=server # Validate with kubeval kubeval manifest.yaml # Validate with kube-score kube-score score manifest.yaml # Check with kube-linter kube-linter lint manifest.yaml ``` **Testing checklist:** - [ ] Manifest passes dry-run validation - [ ] All required fields are present - [ ] Resource limits are reasonable - [ ] Health checks are configured - [ ] Security context is set - [ ] Labels follow conventions - [ ] Namespace exists or is created ## Common Patterns ### Pattern 1: Simple Stateless Web Application **Use case:** Standard web API or microservice **Components needed:** - Deployment (3 replicas for HA) - ClusterIP Service - ConfigMap for configuration - Secret for API keys - HorizontalPodAutoscaler (optional) **Reference:** See `assets/deployment-template.yaml` ### Pattern 2: Stateful Database Application **Use case:** Database or persistent storage application **Components needed:** - StatefulSet (not Deployment) - Headless Service - PersistentVolumeClaim template - ConfigMap for DB configuration - Secret for credentials ### Pattern 3: Background Job or Cron **Use case:** Scheduled tasks or batch processing **Components needed:** - CronJob or Job - ConfigMap for job parameters - Secret for credentials - ServiceAccount with RBAC ### Pattern 4: Multi-Container Pod **Use case:** Application with sidecar containers **Components needed:** - Deployment with multiple containers - Shared volumes between containers - Init containers for setup - Service (if needed) ## Templates The following templates are available in the `assets/` directory: - `deployment-template.yaml` - Standard deployment with best practices - `service-template.yaml` - Service configurations (ClusterIP, LoadBalancer, NodePort) - `configmap-template.yaml` - ConfigMap examples with different data types - `secret-template.yaml` - Secret examples (to be generated, not committed) - `pvc-template.yaml` - PersistentVolumeClaim templates ## Reference Documentation - `references/deployment-spec.md` - Detailed Deployment specification - `references/service-spec.md` - Service types and networking details ## Best Practices Summary 1. **Always set resource requests and limits** - Prevents resource starvation 2. **Implement health checks** - Ensures Kubernetes can manage your application 3. **Use specific image tags** - Avoid unpredictable deployments 4. **Apply security contexts** - Run as non-root, drop capabilities 5. **Use ConfigMaps and Secrets** - Separate config from code 6. **Label everything** - Enables filtering and organization 7. **Follow naming conventions** - Use standard Kubernetes labels 8. **Validate before applying** - Use dry-run and validation tools 9. **Version your manifests** - Keep in Git with version control 10. **Document with annotations** - Add context for other developers ## Troubleshooting **Pods not starting:** - Check image pull errors: `kubectl describe pod <pod-name>` - Verify resource availability: `kubectl get nodes` - Check events: `kubectl get events --sort-by='.lastTimestamp'` **Service not accessible:** - Verify selector matches pod labels: `kubectl get endpoints <service-name>` - Check service type and port configuration - Test from within cluster: `kubectl run debug --rm -it --image=busybox -- sh` **ConfigMap/Secret not loading:** - Verify names match in Deployment - Check namespace - Ensure resources exist: `kubectl get configmap,secret` ## Next Steps After creating manifests: 1. Store in Git repository 2. Set up CI/CD pipeline for deployment 3. Consider using Helm or Kustomize for templating 4. Implement GitOps with ArgoCD or Flux 5. Add monitoring and observability ## Related Skills - `helm-chart-scaffolding` - For templating and packaging - `gitops-workflow` - For automated deployments - `k8s-security-policies` - For advanced security configurations
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

k8s-security-policies

Implement Kubernetes security policies including NetworkPolicy,

architecture
⭐1
# Kubernetes Security Policies Comprehensive guide for implementing NetworkPolicy, PodSecurityPolicy, RBAC, and Pod Security Standards in Kubernetes. ## Purpose Implement defense-in-depth security for Kubernetes clusters using network policies, pod security standards, and RBAC. ## When to Use This Skill - Implement network segmentation - Configure pod security standards - Set up RBAC for least-privilege access - Create security policies for compliance - Implement admission control - Secure multi-tenant clusters ## Pod Security Standards ### 1. Privileged (Unrestricted) ```yaml apiVersion: v1 kind: Namespace metadata: name: privileged-ns labels: pod-security.kubernetes.io/enforce: privileged pod-security.kubernetes.io/audit: privileged pod-security.kubernetes.io/warn: privileged ``` ### 2. Baseline (Minimally restrictive) ```yaml apiVersion: v1 kind: Namespace metadata: name: baseline-ns labels: pod-security.kubernetes.io/enforce: baseline pod-security.kubernetes.io/audit: baseline pod-security.kubernetes.io/warn: baseline ``` ### 3. Restricted (Most restrictive) ```yaml apiVersion: v1 kind: Namespace metadata: name: restricted-ns labels: pod-security.kubernetes.io/enforce: restricted pod-security.kubernetes.io/audit: restricted pod-security.kubernetes.io/warn: restricted ``` ## Network Policies ### Default Deny All ```yaml apiVersion: networking.k8s.io/v1 kind: NetworkPolicy metadata: name: default-deny-all namespace: production spec: podSelector: {} policyTypes: - Ingress - Egress ``` ### Allow Frontend to Backend ```yaml apiVersion: networking.k8s.io/v1 kind: NetworkPolicy metadata: name: allow-frontend-to-backend namespace: production spec: podSelector: matchLabels: app: backend policyTypes: - Ingress ingress: - from: - podSelector: matchLabels: app: frontend ports: - protocol: TCP port: 8080 ``` ### Allow DNS ```yaml apiVersion: networking.k8s.io/v1 kind: NetworkPolicy metadata: name: allow-dns namespace: production spec: podSelector: {} policyTypes: - Egress egress: - to: - namespaceSelector: matchLabels: name: kube-system ports: - protocol: UDP port: 53 ``` **Reference:** See `assets/network-policy-template.yaml` ## RBAC Configuration ### Role (Namespace-scoped) ```yaml apiVersion: rbac.authorization.k8s.io/v1 kind: Role metadata: name: pod-reader namespace: production rules: - apiGroups: [""] resources: ["pods"] verbs: ["get", "watch", "list"] ``` ### ClusterRole (Cluster-wide) ```yaml apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRole metadata: name: secret-reader rules: - apiGroups: [""] resources: ["secrets"] verbs: ["get", "watch", "list"] ``` ### RoleBinding ```yaml apiVersion: rbac.authorization.k8s.io/v1 kind: RoleBinding metadata: name: read-pods namespace: production subjects: - kind: User name: jane apiGroup: rbac.authorization.k8s.io - kind: ServiceAccount name: default namespace: production roleRef: kind: Role name: pod-reader apiGroup: rbac.authorization.k8s.io ``` **Reference:** See `references/rbac-patterns.md` ## Pod Security Context ### Restricted Pod ```yaml apiVersion: v1 kind: Pod metadata: name: secure-pod spec: securityContext: runAsNonRoot: true runAsUser: 1000 fsGroup: 1000 seccompProfile: type: RuntimeDefault containers: - name: app image: myapp:1.0 securityContext: allowPrivilegeEscalation: false readOnlyRootFilesystem: true capabilities: drop: - ALL ``` ## Policy Enforcement with OPA Gatekeeper ### ConstraintTemplate ```yaml apiVersion: templates.gatekeeper.sh/v1 kind: ConstraintTemplate metadata: name: k8srequiredlabels spec: crd: spec: names: kind: K8sRequiredLabels validation: openAPIV3Schema: type: object properties: labels: type: array items: type: string targets: - target: admission.k8s.gatekeeper.sh rego: | package k8srequiredlabels violation[{"msg": msg, "details": {"missing_labels": missing}}] { provided := {label | input.review.object.metadata.labels[label]} required := {label | label := input.parameters.labels[_]} missing := required - provided count(missing) > 0 msg := sprintf("missing required labels: %v", [missing]) } ``` ### Constraint ```yaml apiVersion: constraints.gatekeeper.sh/v1beta1 kind: K8sRequiredLabels metadata: name: require-app-label spec: match: kinds: - apiGroups: ["apps"] kinds: ["Deployment"] parameters: labels: ["app", "environment"] ``` ## Service Mesh Security (Istio) ### PeerAuthentication (mTLS) ```yaml apiVersion: security.istio.io/v1beta1 kind: PeerAuthentication metadata: name: default namespace: production spec: mtls: mode: STRICT ``` ### AuthorizationPolicy ```yaml apiVersion: security.istio.io/v1beta1 kind: AuthorizationPolicy metadata: name: allow-frontend namespace: production spec: selector: matchLabels: app: backend action: ALLOW rules: - from: - source: principals: ["cluster.local/ns/production/sa/frontend"] ``` ## Best Practices 1. **Implement Pod Security Standards** at namespace level 2. **Use Network Policies** for network segmentation 3. **Apply least-privilege RBAC** for all service accounts 4. **Enable admission control** (OPA Gatekeeper/Kyverno) 5. **Run containers as non-root** 6. **Use read-only root filesystem** 7. **Drop all capabilities** unless needed 8. **Implement resource quotas** and limit ranges 9. **Enable audit logging** for security events 10. **Regular security scanning** of images ## Compliance Frameworks ### CIS Kubernetes Benchmark - Use RBAC authorization - Enable audit logging - Use Pod Security Standards - Configure network policies - Implement secrets encryption at rest - Enable node authentication ### NIST Cybersecurity Framework - Implement defense in depth - Use network segmentation - Configure security monitoring - Implement access controls - Enable logging and monitoring ## Troubleshooting **NetworkPolicy not working:** ```bash # Check if CNI supports NetworkPolicy kubectl get nodes -o wide kubectl describe networkpolicy <name> ``` **RBAC permission denied:** ```bash # Check effective permissions kubectl auth can-i list pods --as system:serviceaccount:default:my-sa kubectl auth can-i '*' '*' --as system:serviceaccount:default:my-sa ``` ## Reference Files - `assets/network-policy-template.yaml` - Network policy examples - `assets/pod-security-template.yaml` - Pod security policies - `references/rbac-patterns.md` - RBAC configuration patterns ## Related Skills - `k8s-manifest-generator` - For creating secure manifests - `gitops-workflow` - For automated policy deployment
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

distributed-tracing

Implement distributed tracing with Jaeger and Tempo to track

coding
⭐1
# Distributed Tracing Implement distributed tracing with Jaeger and Tempo for request flow visibility across microservices. ## Purpose Track requests across distributed systems to understand latency, dependencies, and failure points. ## When to Use - Debug latency issues - Understand service dependencies - Identify bottlenecks - Trace error propagation - Analyze request paths ## Distributed Tracing Concepts ### Trace Structure ``` Trace (Request ID: abc123) ↓ Span (frontend) [100ms] ↓ Span (api-gateway) [80ms] ā”œā†’ Span (auth-service) [10ms] └→ Span (user-service) [60ms] └→ Span (database) [40ms] ``` ### Key Components - **Trace** - End-to-end request journey - **Span** - Single operation within a trace - **Context** - Metadata propagated between services - **Tags** - Key-value pairs for filtering - **Logs** - Timestamped events within a span ## Jaeger Setup ### Kubernetes Deployment ```bash # Deploy Jaeger Operator kubectl create namespace observability kubectl create -f https://github.com/jaegertracing/jaeger-operator/releases/download/v1.51.0/jaeger-operator.yaml -n observability # Deploy Jaeger instance kubectl apply -f - <<EOF apiVersion: jaegertracing.io/v1 kind: Jaeger metadata: name: jaeger namespace: observability spec: strategy: production storage: type: elasticsearch options: es: server-urls: http://elasticsearch:9200 ingress: enabled: true EOF ``` ### Docker Compose ```yaml version: "3.8" services: jaeger: image: jaegertracing/all-in-one:latest ports: - "5775:5775/udp" - "6831:6831/udp" - "6832:6832/udp" - "5778:5778" - "16686:16686" # UI - "14268:14268" # Collector - "14250:14250" # gRPC - "9411:9411" # Zipkin environment: - COLLECTOR_ZIPKIN_HOST_PORT=:9411 ``` **Reference:** See `references/jaeger-setup.md` ## Application Instrumentation ### OpenTelemetry (Recommended) #### Python (Flask) ```python from opentelemetry import trace from opentelemetry.exporter.jaeger.thrift import JaegerExporter from opentelemetry.sdk.resources import SERVICE_NAME, Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.instrumentation.flask import FlaskInstrumentor from flask import Flask # Initialize tracer resource = Resource(attributes={SERVICE_NAME: "my-service"}) provider = TracerProvider(resource=resource) processor = BatchSpanProcessor(JaegerExporter( agent_host_name="jaeger", agent_port=6831, )) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # Instrument Flask app = Flask(__name__) FlaskInstrumentor().instrument_app(app) @app.route('/api/users') def get_users(): tracer = trace.get_tracer(__name__) with tracer.start_as_current_span("get_users") as span: span.set_attribute("user.count", 100) # Business logic users = fetch_users_from_db() return {"users": users} def fetch_users_from_db(): tracer = trace.get_tracer(__name__) with tracer.start_as_current_span("database_query") as span: span.set_attribute("db.system", "postgresql") span.set_attribute("db.statement", "SELECT * FROM users") # Database query return query_database() ``` #### Node.js (Express) ```javascript const { NodeTracerProvider } = require("@opentelemetry/sdk-trace-node"); const { JaegerExporter } = require("@opentelemetry/exporter-jaeger"); const { BatchSpanProcessor } = require("@opentelemetry/sdk-trace-base"); const { registerInstrumentations } = require("@opentelemetry/instrumentation"); const { HttpInstrumentation } = require("@opentelemetry/instrumentation-http"); const { ExpressInstrumentation, } = require("@opentelemetry/instrumentation-express"); // Initialize tracer const provider = new NodeTracerProvider({ resource: { attributes: { "service.name": "my-service" } }, }); const exporter = new JaegerExporter({ endpoint: "http://jaeger:14268/api/traces", }); provider.addSpanProcessor(new BatchSpanProcessor(exporter)); provider.register(); // Instrument libraries registerInstrumentations({ instrumentations: [new HttpInstrumentation(), new ExpressInstrumentation()], }); const express = require("express"); const app = express(); app.get("/api/users", async (req, res) => { const tracer = trace.getTracer("my-service"); const span = tracer.startSpan("get_users"); try { const users = await fetchUsers(); span.setAttributes({ "user.count": users.length }); res.json({ users }); } finally { span.end(); } }); ``` #### Go ```go package main import ( "context" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/exporters/jaeger" "go.opentelemetry.io/otel/sdk/resource" sdktrace "go.opentelemetry.io/otel/sdk/trace" semconv "go.opentelemetry.io/otel/semconv/v1.4.0" ) func initTracer() (*sdktrace.TracerProvider, error) { exporter, err := jaeger.New(jaeger.WithCollectorEndpoint( jaeger.WithEndpoint("http://jaeger:14268/api/traces"), )) if err != nil { return nil, err } tp := sdktrace.NewTracerProvider( sdktrace.WithBatcher(exporter), sdktrace.WithResource(resource.NewWithAttributes( semconv.SchemaURL, semconv.ServiceNameKey.String("my-service"), )), ) otel.SetTracerProvider(tp) return tp, nil } func getUsers(ctx context.Context) ([]User, error) { tracer := otel.Tracer("my-service") ctx, span := tracer.Start(ctx, "get_users") defer span.End() span.SetAttributes(attribute.String("user.filter", "active")) users, err := fetchUsersFromDB(ctx) if err != nil { span.RecordError(err) return nil, err } span.SetAttributes(attribute.Int("user.count", len(users))) return users, nil } ``` **Reference:** See `references/instrumentation.md` ## Context Propagation ### HTTP Headers ``` traceparent: 00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01 tracestate: congo=t61rcWkgMzE ``` ### Propagation in HTTP Requests #### Python ```python from opentelemetry.propagate import inject headers = {} inject(headers) # Injects trace context response = requests.get('http://downstream-service/api', headers=headers) ``` #### Node.js ```javascript const { propagation } = require("@opentelemetry/api"); const headers = {}; propagation.inject(context.active(), headers); axios.get("http://downstream-service/api", { headers }); ``` ## Tempo Setup (Grafana) ### Kubernetes Deployment ```yaml apiVersion: v1 kind: ConfigMap metadata: name: tempo-config data: tempo.yaml: | server: http_listen_port: 3200 distributor: receivers: jaeger: protocols: thrift_http: grpc: otlp: protocols: http: grpc: storage: trace: backend: s3 s3: bucket: tempo-traces endpoint: s3.amazonaws.com querier: frontend_worker: frontend_address: tempo-query-frontend:9095 --- apiVersion: apps/v1 kind: Deployment metadata: name: tempo spec: replicas: 1 template: spec: containers: - name: tempo image: grafana/tempo:latest args: - -config.file=/etc/tempo/tempo.yaml volumeMounts: - name: config mountPath: /etc/tempo volumes: - name: config configMap: name: tempo-config ``` **Reference:** See `assets/jaeger-config.yaml.template` ## Sampling Strategies ### Probabilistic Sampling ```yaml # Sample 1% of traces sampler: type: probabilistic param: 0.01 ``` ### Rate Limiting Sampling ```yaml # Sample max 100 traces per second sampler: type: ratelimiting param: 100 ``` ### Adaptive Sampling ```python from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased # Sample based on trace ID (deterministic) sampler = ParentBased(root=TraceIdRatioBased(0.01)) ``` ## Trace Analysis ### Finding Slow Requests **Jaeger Query:** ``` service=my-service duration > 1s ``` ### Finding Errors **Jaeger Query:** ``` service=my-service error=true tags.http.status_code >= 500 ``` ### Service Dependency Graph Jaeger automatically generates service dependency graphs showing: - Service relationships - Request rates - Error rates - Average latencies ## Best Practices 1. **Sample appropriately** (1-10% in production) 2. **Add meaningful tags** (user_id, request_id) 3. **Propagate context** across all service boundaries 4. **Log exceptions** in spans 5. **Use consistent naming** for operations 6. **Monitor tracing overhead** (<1% CPU impact) 7. **Set up alerts** for trace errors 8. **Implement distributed context** (baggage) 9. **Use span events** for important milestones 10. **Document instrumentation** standards ## Integration with Logging ### Correlated Logs ```python import logging from opentelemetry import trace logger = logging.getLogger(__name__) def process_request(): span = trace.get_current_span() trace_id = span.get_span_context().trace_id logger.info( "Processing request", extra={"trace_id": format(trace_id, '032x')} ) ``` ## Troubleshooting **No traces appearing:** - Check collector endpoint - Verify network connectivity - Check sampling configuration - Review application logs **High latency overhead:** - Reduce sampling rate - Use batch span processor - Check exporter configuration ## Reference Files - `references/jaeger-setup.md` - Jaeger installation - `references/instrumentation.md` - Instrumentation patterns - `assets/jaeger-config.yaml.template` - Jaeger configuration ## Related Skills - `prometheus-configuration` - For metrics - `grafana-dashboards` - For visualization - `slo-implementation` - For latency SLOs
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

python-design-patterns

Python design patterns including KISS, Separation of Concerns,

coding
⭐1
# Python Design Patterns Write maintainable Python code using fundamental design principles. These patterns help you build systems that are easy to understand, test, and modify. ## When to Use This Skill - Designing new components or services - Refactoring complex or tangled code - Deciding whether to create an abstraction - Choosing between inheritance and composition - Evaluating code complexity and coupling - Planning modular architectures ## Core Concepts ### 1. KISS (Keep It Simple) Choose the simplest solution that works. Complexity must be justified by concrete requirements. ### 2. Single Responsibility (SRP) Each unit should have one reason to change. Separate concerns into focused components. ### 3. Composition Over Inheritance Build behavior by combining objects, not extending classes. ### 4. Rule of Three Wait until you have three instances before abstracting. Duplication is often better than premature abstraction. ## Quick Start ```python # Simple beats clever # Instead of a factory/registry pattern: FORMATTERS = {"json": JsonFormatter, "csv": CsvFormatter} def get_formatter(name: str) -> Formatter: return FORMATTERS[name]() ``` ## Fundamental Patterns ### Pattern 1: KISS - Keep It Simple Before adding complexity, ask: does a simpler solution work? ```python # Over-engineered: Factory with registration class OutputFormatterFactory: _formatters: dict[str, type[Formatter]] = {} @classmethod def register(cls, name: str): def decorator(formatter_cls): cls._formatters[name] = formatter_cls return formatter_cls return decorator @classmethod def create(cls, name: str) -> Formatter: return cls._formatters[name]() @OutputFormatterFactory.register("json") class JsonFormatter(Formatter): ... # Simple: Just use a dictionary FORMATTERS = { "json": JsonFormatter, "csv": CsvFormatter, "xml": XmlFormatter, } def get_formatter(name: str) -> Formatter: """Get formatter by name.""" if name not in FORMATTERS: raise ValueError(f"Unknown format: {name}") return FORMATTERS[name]() ``` The factory pattern adds code without adding value here. Save patterns for when they solve real problems. ### Pattern 2: Single Responsibility Principle Each class or function should have one reason to change. ```python # BAD: Handler does everything class UserHandler: async def create_user(self, request: Request) -> Response: # HTTP parsing data = await request.json() # Validation if not data.get("email"): return Response({"error": "email required"}, status=400) # Database access user = await db.execute( "INSERT INTO users (email, name) VALUES ($1, $2) RETURNING *", data["email"], data["name"] ) # Response formatting return Response({"id": user.id, "email": user.email}, status=201) # GOOD: Separated concerns class UserService: """Business logic only.""" def __init__(self, repo: UserRepository) -> None: self._repo = repo async def create_user(self, data: CreateUserInput) -> User: # Only business rules here user = User(email=data.email, name=data.name) return await self._repo.save(user) class UserHandler: """HTTP concerns only.""" def __init__(self, service: UserService) -> None: self._service = service async def create_user(self, request: Request) -> Response: data = CreateUserInput(**(await request.json())) user = await self._service.create_user(data) return Response(user.to_dict(), status=201) ``` Now HTTP changes don't affect business logic, and vice versa. ### Pattern 3: Separation of Concerns Organize code into distinct layers with clear responsibilities. ``` ā”Œā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā” │ API Layer (handlers) │ │ - Parse requests │ │ - Call services │ │ - Format responses │ ā””ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”˜ │ ā–¼ ā”Œā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā” │ Service Layer (business logic) │ │ - Domain rules and validation │ │ - Orchestrate operations │ │ - Pure functions where possible │ ā””ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”˜ │ ā–¼ ā”Œā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā” │ Repository Layer (data access) │ │ - SQL queries │ │ - External API calls │ │ - Cache operations │ ā””ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”˜ ``` Each layer depends only on layers below it: ```python # Repository: Data access class UserRepository: async def get_by_id(self, user_id: str) -> User | None: row = await self._db.fetchrow( "SELECT * FROM users WHERE id = $1", user_id ) return User(**row) if row else None # Service: Business logic class UserService: def __init__(self, repo: UserRepository) -> None: self._repo = repo async def get_user(self, user_id: str) -> User: user = await self._repo.get_by_id(user_id) if user is None: raise UserNotFoundError(user_id) return user # Handler: HTTP concerns @app.get("/users/{user_id}") async def get_user(user_id: str) -> UserResponse: user = await user_service.get_user(user_id) return UserResponse.from_user(user) ``` ### Pattern 4: Composition Over Inheritance Build behavior by combining objects rather than inheriting. ```python # Inheritance: Rigid and hard to test class EmailNotificationService(NotificationService): def __init__(self): super().__init__() self._smtp = SmtpClient() # Hard to mock def notify(self, user: User, message: str) -> None: self._smtp.send(user.email, message) # Composition: Flexible and testable class NotificationService: """Send notifications via multiple channels.""" def __init__( self, email_sender: EmailSender, sms_sender: SmsSender | None = None, push_sender: PushSender | None = None, ) -> None: self._email = email_sender self._sms = sms_sender self._push = push_sender async def notify( self, user: User, message: str, channels: set[str] | None = None, ) -> None: channels = channels or {"email"} if "email" in channels: await self._email.send(user.email, message) if "sms" in channels and self._sms and user.phone: await self._sms.send(user.phone, message) if "push" in channels and self._push and user.device_token: await self._push.send(user.device_token, message) # Easy to test with fakes service = NotificationService( email_sender=FakeEmailSender(), sms_sender=FakeSmsSender(), ) ``` ## Advanced Patterns ### Pattern 5: Rule of Three Wait until you have three instances before abstracting. ```python # Two similar functions? Don't abstract yet def process_orders(orders: list[Order]) -> list[Result]: results = [] for order in orders: validated = validate_order(order) result = process_validated_order(validated) results.append(result) return results def process_returns(returns: list[Return]) -> list[Result]: results = [] for ret in returns: validated = validate_return(ret) result = process_validated_return(validated) results.append(result) return results # These look similar, but wait! Are they actually the same? # Different validation, different processing, different errors... # Duplication is often better than the wrong abstraction # Only after a third case, consider if there's a real pattern # But even then, sometimes explicit is better than abstract ``` ### Pattern 6: Function Size Guidelines Keep functions focused. Extract when a function: - Exceeds 20-50 lines (varies by complexity) - Serves multiple distinct purposes - Has deeply nested logic (3+ levels) ```python # Too long, multiple concerns mixed def process_order(order: Order) -> Result: # 50 lines of validation... # 30 lines of inventory check... # 40 lines of payment processing... # 20 lines of notification... pass # Better: Composed from focused functions def process_order(order: Order) -> Result: """Process a customer order through the complete workflow.""" validate_order(order) reserve_inventory(order) payment_result = charge_payment(order) send_confirmation(order, payment_result) return Result(success=True, order_id=order.id) ``` ### Pattern 7: Dependency Injection Pass dependencies through constructors for testability. ```python from typing import Protocol class Logger(Protocol): def info(self, msg: str, **kwargs) -> None: ... def error(self, msg: str, **kwargs) -> None: ... class Cache(Protocol): async def get(self, key: str) -> str | None: ... async def set(self, key: str, value: str, ttl: int) -> None: ... class UserService: """Service with injected dependencies.""" def __init__( self, repository: UserRepository, cache: Cache, logger: Logger, ) -> None: self._repo = repository self._cache = cache self._logger = logger async def get_user(self, user_id: str) -> User: # Check cache first cached = await self._cache.get(f"user:{user_id}") if cached: self._logger.info("Cache hit", user_id=user_id) return User.from_json(cached) # Fetch from database user = await self._repo.get_by_id(user_id) if user: await self._cache.set(f"user:{user_id}", user.to_json(), ttl=300) return user # Production service = UserService( repository=PostgresUserRepository(db), cache=RedisCache(redis), logger=StructlogLogger(), ) # Testing service = UserService( repository=InMemoryUserRepository(), cache=FakeCache(), logger=NullLogger(), ) ``` ### Pattern 8: Avoiding Common Anti-Patterns **Don't expose internal types:** ```python # BAD: Leaking ORM model to API @app.get("/users/{id}") def get_user(id: str) -> UserModel: # SQLAlchemy model return db.query(UserModel).get(id) # GOOD: Use response schemas @app.get("/users/{id}") def get_user(id: str) -> UserResponse: user = db.query(UserModel).get(id) return UserResponse.from_orm(user) ``` **Don't mix I/O with business logic:** ```python # BAD: SQL embedded in business logic def calculate_discount(user_id: str) -> float: user = db.query("SELECT * FROM users WHERE id = ?", user_id) orders = db.query("SELECT * FROM orders WHERE user_id = ?", user_id) # Business logic mixed with data access # GOOD: Repository pattern def calculate_discount(user: User, order_history: list[Order]) -> float: # Pure business logic, easily testable if len(order_history) > 10: return 0.15 return 0.0 ``` ## Best Practices Summary 1. **Keep it simple** - Choose the simplest solution that works 2. **Single responsibility** - Each unit has one reason to change 3. **Separate concerns** - Distinct layers with clear purposes 4. **Compose, don't inherit** - Combine objects for flexibility 5. **Rule of three** - Wait before abstracting 6. **Keep functions small** - 20-50 lines (varies by complexity), one purpose 7. **Inject dependencies** - Constructor injection for testability 8. **Delete before abstracting** - Remove dead code, then consider patterns 9. **Test each layer** - Isolated tests for each concern 10. **Explicit over clever** - Readable code beats elegant code
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

python-resilience

Python resilience patterns including automatic retries, exponential

coding
⭐1
# Python Resilience Patterns Build fault-tolerant Python applications that gracefully handle transient failures, network issues, and service outages. Resilience patterns keep systems running when dependencies are unreliable. ## When to Use This Skill - Adding retry logic to external service calls - Implementing timeouts for network operations - Building fault-tolerant microservices - Handling rate limiting and backpressure - Creating infrastructure decorators - Designing circuit breakers ## Core Concepts ### 1. Transient vs Permanent Failures Retry transient errors (network timeouts, temporary service issues). Don't retry permanent errors (invalid credentials, bad requests). ### 2. Exponential Backoff Increase wait time between retries to avoid overwhelming recovering services. ### 3. Jitter Add randomness to backoff to prevent thundering herd when many clients retry simultaneously. ### 4. Bounded Retries Cap both attempt count and total duration to prevent infinite retry loops. ## Quick Start ```python from tenacity import retry, stop_after_attempt, wait_exponential_jitter @retry( stop=stop_after_attempt(3), wait=wait_exponential_jitter(initial=1, max=10), ) def call_external_service(request: dict) -> dict: return httpx.post("https://api.example.com", json=request).json() ``` ## Fundamental Patterns ### Pattern 1: Basic Retry with Tenacity Use the `tenacity` library for production-grade retry logic. For simpler cases, consider built-in retry functionality or a lightweight custom implementation. ```python from tenacity import ( retry, stop_after_attempt, stop_after_delay, wait_exponential_jitter, retry_if_exception_type, ) TRANSIENT_ERRORS = (ConnectionError, TimeoutError, OSError) @retry( retry=retry_if_exception_type(TRANSIENT_ERRORS), stop=stop_after_attempt(5) | stop_after_delay(60), wait=wait_exponential_jitter(initial=1, max=30), ) def fetch_data(url: str) -> dict: """Fetch data with automatic retry on transient failures.""" response = httpx.get(url, timeout=30) response.raise_for_status() return response.json() ``` ### Pattern 2: Retry Only Appropriate Errors Whitelist specific transient exceptions. Never retry: - `ValueError`, `TypeError` - These are bugs, not transient issues - `AuthenticationError` - Invalid credentials won't become valid - HTTP 4xx errors (except 429) - Client errors are permanent ```python from tenacity import retry, retry_if_exception_type import httpx # Define what's retryable RETRYABLE_EXCEPTIONS = ( ConnectionError, TimeoutError, httpx.ConnectTimeout, httpx.ReadTimeout, ) @retry( retry=retry_if_exception_type(RETRYABLE_EXCEPTIONS), stop=stop_after_attempt(3), wait=wait_exponential_jitter(initial=1, max=10), ) def resilient_api_call(endpoint: str) -> dict: """Make API call with retry on network issues.""" return httpx.get(endpoint, timeout=10).json() ``` ### Pattern 3: HTTP Status Code Retries Retry specific HTTP status codes that indicate transient issues. ```python from tenacity import retry, retry_if_result, stop_after_attempt import httpx RETRY_STATUS_CODES = {429, 502, 503, 504} def should_retry_response(response: httpx.Response) -> bool: """Check if response indicates a retryable error.""" return response.status_code in RETRY_STATUS_CODES @retry( retry=retry_if_result(should_retry_response), stop=stop_after_attempt(3), wait=wait_exponential_jitter(initial=1, max=10), ) def http_request(method: str, url: str, **kwargs) -> httpx.Response: """Make HTTP request with retry on transient status codes.""" return httpx.request(method, url, timeout=30, **kwargs) ``` ### Pattern 4: Combined Exception and Status Retry Handle both network exceptions and HTTP status codes. ```python from tenacity import ( retry, retry_if_exception_type, retry_if_result, stop_after_attempt, wait_exponential_jitter, before_sleep_log, ) import logging import httpx logger = logging.getLogger(__name__) TRANSIENT_EXCEPTIONS = ( ConnectionError, TimeoutError, httpx.ConnectError, httpx.ReadTimeout, ) RETRY_STATUS_CODES = {429, 500, 502, 503, 504} def is_retryable_response(response: httpx.Response) -> bool: return response.status_code in RETRY_STATUS_CODES @retry( retry=( retry_if_exception_type(TRANSIENT_EXCEPTIONS) | retry_if_result(is_retryable_response) ), stop=stop_after_attempt(5), wait=wait_exponential_jitter(initial=1, max=30), before_sleep=before_sleep_log(logger, logging.WARNING), ) def robust_http_call( method: str, url: str, **kwargs, ) -> httpx.Response: """HTTP call with comprehensive retry handling.""" return httpx.request(method, url, timeout=30, **kwargs) ``` ## Advanced Patterns ### Pattern 5: Logging Retry Attempts Track retry behavior for debugging and alerting. ```python from tenacity import retry, stop_after_attempt, wait_exponential import structlog logger = structlog.get_logger() def log_retry_attempt(retry_state): """Log detailed retry information.""" exception = retry_state.outcome.exception() logger.warning( "Retrying operation", attempt=retry_state.attempt_number, exception_type=type(exception).__name__, exception_message=str(exception), next_wait_seconds=retry_state.next_action.sleep if retry_state.next_action else None, ) @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10), before_sleep=log_retry_attempt, ) def call_with_logging(request: dict) -> dict: """External call with retry logging.""" ... ``` ### Pattern 6: Timeout Decorator Create reusable timeout decorators for consistent timeout handling. ```python import asyncio from functools import wraps from typing import TypeVar, Callable T = TypeVar("T") def with_timeout(seconds: float): """Decorator to add timeout to async functions.""" def decorator(func: Callable[..., T]) -> Callable[..., T]: @wraps(func) async def wrapper(*args, **kwargs) -> T: return await asyncio.wait_for( func(*args, **kwargs), timeout=seconds, ) return wrapper return decorator @with_timeout(30) async def fetch_with_timeout(url: str) -> dict: """Fetch URL with 30 second timeout.""" async with httpx.AsyncClient() as client: response = await client.get(url) return response.json() ``` ### Pattern 7: Cross-Cutting Concerns via Decorators Stack decorators to separate infrastructure from business logic. ```python from functools import wraps from typing import TypeVar, Callable import structlog logger = structlog.get_logger() T = TypeVar("T") def traced(name: str | None = None): """Add tracing to function calls.""" def decorator(func: Callable[..., T]) -> Callable[..., T]: span_name = name or func.__name__ @wraps(func) async def wrapper(*args, **kwargs) -> T: logger.info("Operation started", operation=span_name) try: result = await func(*args, **kwargs) logger.info("Operation completed", operation=span_name) return result except Exception as e: logger.error("Operation failed", operation=span_name, error=str(e)) raise return wrapper return decorator # Stack multiple concerns @traced("fetch_user_data") @with_timeout(30) @retry(stop=stop_after_attempt(3), wait=wait_exponential_jitter()) async def fetch_user_data(user_id: str) -> dict: """Fetch user with tracing, timeout, and retry.""" ... ``` ### Pattern 8: Dependency Injection for Testability Pass infrastructure components through constructors for easy testing. ```python from dataclasses import dataclass from typing import Protocol class Logger(Protocol): def info(self, msg: str, **kwargs) -> None: ... def error(self, msg: str, **kwargs) -> None: ... class MetricsClient(Protocol): def increment(self, metric: str, tags: dict | None = None) -> None: ... def timing(self, metric: str, value: float) -> None: ... @dataclass class UserService: """Service with injected infrastructure.""" repository: UserRepository logger: Logger metrics: MetricsClient async def get_user(self, user_id: str) -> User: self.logger.info("Fetching user", user_id=user_id) start = time.perf_counter() try: user = await self.repository.get(user_id) self.metrics.increment("user.fetch.success") return user except Exception as e: self.metrics.increment("user.fetch.error") self.logger.error("Failed to fetch user", user_id=user_id, error=str(e)) raise finally: elapsed = time.perf_counter() - start self.metrics.timing("user.fetch.duration", elapsed) # Easy to test with fakes service = UserService( repository=FakeRepository(), logger=FakeLogger(), metrics=FakeMetrics(), ) ``` ### Pattern 9: Fail-Safe Defaults Degrade gracefully when non-critical operations fail. ```python from typing import TypeVar from collections.abc import Callable T = TypeVar("T") def fail_safe(default: T, log_failure: bool = True): """Return default value on failure instead of raising.""" def decorator(func: Callable[..., T]) -> Callable[..., T]: @wraps(func) async def wrapper(*args, **kwargs) -> T: try: return await func(*args, **kwargs) except Exception as e: if log_failure: logger.warning( "Operation failed, using default", function=func.__name__, error=str(e), ) return default return wrapper return decorator @fail_safe(default=[]) async def get_recommendations(user_id: str) -> list[str]: """Get recommendations, return empty list on failure.""" ... ``` ## Best Practices Summary 1. **Retry only transient errors** - Don't retry bugs or authentication failures 2. **Use exponential backoff** - Give services time to recover 3. **Add jitter** - Prevent thundering herd from synchronized retries 4. **Cap total duration** - `stop_after_attempt(5) | stop_after_delay(60)` 5. **Log every retry** - Silent retries hide systemic problems 6. **Use decorators** - Keep retry logic separate from business logic 7. **Inject dependencies** - Make infrastructure testable 8. **Set timeouts everywhere** - Every network call needs a timeout 9. **Fail gracefully** - Return cached/default values for non-critical paths 10. **Monitor retry rates** - High retry rates indicate underlying issues
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

python-resource-management

Python resource management with context managers, cleanup patterns,

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

python-type-safety

Python type safety with type hints, generics, protocols, and strict

coding
⭐1
# Python Type Safety Leverage Python's type system to catch errors at static analysis time. Type annotations serve as enforced documentation that tooling validates automatically. ## When to Use This Skill - Adding type hints to existing code - Creating generic, reusable classes - Defining structural interfaces with protocols - Configuring mypy or pyright for strict checking - Understanding type narrowing and guards - Building type-safe APIs and libraries ## Core Concepts ### 1. Type Annotations Declare expected types for function parameters, return values, and variables. ### 2. Generics Write reusable code that preserves type information across different types. ### 3. Protocols Define structural interfaces without inheritance (duck typing with type safety). ### 4. Type Narrowing Use guards and conditionals to narrow types within code blocks. ## Quick Start ```python def get_user(user_id: str) -> User | None: """Return type makes 'might not exist' explicit.""" ... # Type checker enforces handling None case user = get_user("123") if user is None: raise UserNotFoundError("123") print(user.name) # Type checker knows user is User here ``` ## Fundamental Patterns ### Pattern 1: Annotate All Public Signatures Every public function, method, and class should have type annotations. ```python def get_user(user_id: str) -> User: """Retrieve user by ID.""" ... def process_batch( items: list[Item], max_workers: int = 4, ) -> BatchResult[ProcessedItem]: """Process items concurrently.""" ... class UserRepository: def __init__(self, db: Database) -> None: self._db = db async def find_by_id(self, user_id: str) -> User | None: """Return User if found, None otherwise.""" ... async def find_by_email(self, email: str) -> User | None: ... async def save(self, user: User) -> User: """Save and return user with generated ID.""" ... ``` Use `mypy --strict` or `pyright` in CI to catch type errors early. For existing projects, enable strict mode incrementally using per-module overrides. ### Pattern 2: Use Modern Union Syntax Python 3.10+ provides cleaner union syntax. ```python # Preferred (3.10+) def find_user(user_id: str) -> User | None: ... def parse_value(v: str) -> int | float | str: ... # Older style (still valid, needed for 3.9) from typing import Optional, Union def find_user(user_id: str) -> Optional[User]: ... ``` ### Pattern 3: Type Narrowing with Guards Use conditionals to narrow types for the type checker. ```python def process_user(user_id: str) -> UserData: user = find_user(user_id) if user is None: raise UserNotFoundError(f"User {user_id} not found") # Type checker knows user is User here, not User | None return UserData( name=user.name, email=user.email, ) def process_items(items: list[Item | None]) -> list[ProcessedItem]: # Filter and narrow types valid_items = [item for item in items if item is not None] # valid_items is now list[Item] return [process(item) for item in valid_items] ``` ### Pattern 4: Generic Classes Create type-safe reusable containers. ```python from typing import TypeVar, Generic T = TypeVar("T") E = TypeVar("E", bound=Exception) class Result(Generic[T, E]): """Represents either a success value or an error.""" def __init__( self, value: T | None = None, error: E | None = None, ) -> None: if (value is None) == (error is None): raise ValueError("Exactly one of value or error must be set") self._value = value self._error = error @property def is_success(self) -> bool: return self._error is None @property def is_failure(self) -> bool: return self._error is not None def unwrap(self) -> T: """Get value or raise the error.""" if self._error is not None: raise self._error return self._value # type: ignore[return-value] def unwrap_or(self, default: T) -> T: """Get value or return default.""" if self._error is not None: return default return self._value # type: ignore[return-value] # Usage preserves types def parse_config(path: str) -> Result[Config, ConfigError]: try: return Result(value=Config.from_file(path)) except ConfigError as e: return Result(error=e) result = parse_config("config.yaml") if result.is_success: config = result.unwrap() # Type: Config ``` ## Advanced Patterns ### Pattern 5: Generic Repository Create type-safe data access patterns. ```python from typing import TypeVar, Generic from abc import ABC, abstractmethod T = TypeVar("T") ID = TypeVar("ID") class Repository(ABC, Generic[T, ID]): """Generic repository interface.""" @abstractmethod async def get(self, id: ID) -> T | None: """Get entity by ID.""" ... @abstractmethod async def save(self, entity: T) -> T: """Save and return entity.""" ... @abstractmethod async def delete(self, id: ID) -> bool: """Delete entity, return True if existed.""" ... class UserRepository(Repository[User, str]): """Concrete repository for Users with string IDs.""" async def get(self, id: str) -> User | None: row = await self._db.fetchrow( "SELECT * FROM users WHERE id = $1", id ) return User(**row) if row else None async def save(self, entity: User) -> User: ... async def delete(self, id: str) -> bool: ... ``` ### Pattern 6: TypeVar with Bounds Restrict generic parameters to specific types. ```python from typing import TypeVar from pydantic import BaseModel ModelT = TypeVar("ModelT", bound=BaseModel) def validate_and_create(model_cls: type[ModelT], data: dict) -> ModelT: """Create a validated Pydantic model from dict.""" return model_cls.model_validate(data) # Works with any BaseModel subclass class User(BaseModel): name: str email: str user = validate_and_create(User, {"name": "Alice", "email": "a@b.com"}) # user is typed as User # Type error: str is not a BaseModel subclass result = validate_and_create(str, {"name": "Alice"}) # Error! ``` ### Pattern 7: Protocols for Structural Typing Define interfaces without requiring inheritance. ```python from typing import Protocol, runtime_checkable @runtime_checkable class Serializable(Protocol): """Any class that can be serialized to/from dict.""" def to_dict(self) -> dict: ... @classmethod def from_dict(cls, data: dict) -> "Serializable": ... # User satisfies Serializable without inheriting from it class User: def __init__(self, id: str, name: str) -> None: self.id = id self.name = name def to_dict(self) -> dict: return {"id": self.id, "name": self.name} @classmethod def from_dict(cls, data: dict) -> "User": return cls(id=data["id"], name=data["name"]) def serialize(obj: Serializable) -> str: """Works with any Serializable object.""" return json.dumps(obj.to_dict()) # Works - User matches the protocol serialize(User("1", "Alice")) # Runtime checking with @runtime_checkable isinstance(User("1", "Alice"), Serializable) # True ``` ### Pattern 8: Common Protocol Patterns Define reusable structural interfaces. ```python from typing import Protocol class Closeable(Protocol): """Resource that can be closed.""" def close(self) -> None: ... class AsyncCloseable(Protocol): """Async resource that can be closed.""" async def close(self) -> None: ... class Readable(Protocol): """Object that can be read from.""" def read(self, n: int = -1) -> bytes: ... class HasId(Protocol): """Object with an ID property.""" @property def id(self) -> str: ... class Comparable(Protocol): """Object that supports comparison.""" def __lt__(self, other: "Comparable") -> bool: ... def __le__(self, other: "Comparable") -> bool: ... ``` ### Pattern 9: Type Aliases Create meaningful type names. **Note:** The `type` statement was introduced in Python 3.10 for simple aliases. Generic type statements require Python 3.12+. ```python # Python 3.10+ type statement for simple aliases type UserId = str type UserDict = dict[str, Any] # Python 3.12+ type statement with generics type Handler[T] = Callable[[Request], T] type AsyncHandler[T] = Callable[[Request], Awaitable[T]] # Python 3.9-3.11 style (needed for broader compatibility) from typing import TypeAlias from collections.abc import Callable, Awaitable UserId: TypeAlias = str Handler: TypeAlias = Callable[[Request], Response] # Usage def register_handler(path: str, handler: Handler[Response]) -> None: ... ``` ### Pattern 10: Callable Types Type function parameters and callbacks. ```python from collections.abc import Callable, Awaitable # Sync callback ProgressCallback = Callable[[int, int], None] # (current, total) # Async callback AsyncHandler = Callable[[Request], Awaitable[Response]] # With named parameters (using Protocol) class OnProgress(Protocol): def __call__( self, current: int, total: int, *, message: str = "", ) -> None: ... def process_items( items: list[Item], on_progress: ProgressCallback | None = None, ) -> list[Result]: for i, item in enumerate(items): if on_progress: on_progress(i, len(items)) ... ``` ## Configuration ### Strict Mode Checklist For `mypy --strict` compliance: ```toml # pyproject.toml [tool.mypy] python_version = "3.12" strict = true warn_return_any = true warn_unused_ignores = true disallow_untyped_defs = true disallow_incomplete_defs = true no_implicit_optional = true ``` Incremental adoption goals: - All function parameters annotated - All return types annotated - Class attributes annotated - Minimize `Any` usage (acceptable for truly dynamic data) - Generic collections use type parameters (`list[str]` not `list`) For existing codebases, enable strict mode per-module using `# mypy: strict` or configure per-module overrides in `pyproject.toml`. ## Best Practices Summary 1. **Annotate all public APIs** - Functions, methods, class attributes 2. **Use `T | None`** - Modern union syntax over `Optional[T]` 3. **Run strict type checking** - `mypy --strict` in CI 4. **Use generics** - Preserve type info in reusable code 5. **Define protocols** - Structural typing for interfaces 6. **Narrow types** - Use guards to help the type checker 7. **Bound type vars** - Restrict generics to meaningful types 8. **Create type aliases** - Meaningful names for complex types 9. **Minimize `Any`** - Use specific types or generics. `Any` is acceptable for truly dynamic data or when interfacing with untyped third-party code 10. **Document with types** - Types are enforceable documentation
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

protocol-reverse-engineering

Master network protocol reverse engineering including packet

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

stride-analysis-patterns

Apply STRIDE methodology to systematically identify threats. Use

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

bats-testing-patterns

Master Bash Automated Testing System (Bats) for comprehensive shell

coding
⭐1
# Bats Testing Patterns Comprehensive guidance for writing comprehensive unit tests for shell scripts using Bats (Bash Automated Testing System), including test patterns, fixtures, and best practices for production-grade shell testing. ## When to Use This Skill - Writing unit tests for shell scripts - Implementing test-driven development (TDD) for scripts - Setting up automated testing in CI/CD pipelines - Testing edge cases and error conditions - Validating behavior across different shell environments - Building maintainable test suites for scripts - Creating fixtures for complex test scenarios - Testing multiple shell dialects (bash, sh, dash) ## Bats Fundamentals ### What is Bats? Bats (Bash Automated Testing System) is a TAP (Test Anything Protocol) compliant testing framework for shell scripts that provides: - Simple, natural test syntax - TAP output format compatible with CI systems - Fixtures and setup/teardown support - Assertion helpers - Parallel test execution ### Installation ```bash # macOS with Homebrew brew install bats-core # Ubuntu/Debian git clone https://github.com/bats-core/bats-core.git cd bats-core ./install.sh /usr/local # From npm (Node.js) npm install --global bats # Verify installation bats --version ``` ### File Structure ``` project/ ā”œā”€ā”€ bin/ │ ā”œā”€ā”€ script.sh │ └── helper.sh ā”œā”€ā”€ tests/ │ ā”œā”€ā”€ test_script.bats │ ā”œā”€ā”€ test_helper.sh │ ā”œā”€ā”€ fixtures/ │ │ ā”œā”€ā”€ input.txt │ │ └── expected_output.txt │ └── helpers/ │ └── mocks.bash └── README.md ``` ## Basic Test Structure ### Simple Test File ```bash #!/usr/bin/env bats # Load test helper if present load test_helper # Setup runs before each test setup() { export TMPDIR=$(mktemp -d) } # Teardown runs after each test teardown() { rm -rf "$TMPDIR" } # Test: simple assertion @test "Function returns 0 on success" { run my_function "input" [ "$status" -eq 0 ] } # Test: output verification @test "Function outputs correct result" { run my_function "test" [ "$output" = "expected output" ] } # Test: error handling @test "Function returns 1 on missing argument" { run my_function [ "$status" -eq 1 ] } ``` ## Assertion Patterns ### Exit Code Assertions ```bash #!/usr/bin/env bats @test "Command succeeds" { run true [ "$status" -eq 0 ] } @test "Command fails as expected" { run false [ "$status" -ne 0 ] } @test "Command returns specific exit code" { run my_function --invalid [ "$status" -eq 127 ] } @test "Can capture command result" { run echo "hello" [ $status -eq 0 ] [ "$output" = "hello" ] } ``` ### Output Assertions ```bash #!/usr/bin/env bats @test "Output matches string" { result=$(echo "hello world") [ "$result" = "hello world" ] } @test "Output contains substring" { result=$(echo "hello world") [[ "$result" == *"world"* ]] } @test "Output matches pattern" { result=$(date +%Y) [[ "$result" =~ ^[0-9]{4}$ ]] } @test "Multi-line output" { run printf "line1\nline2\nline3" [ "$output" = "line1 line2 line3" ] } @test "Lines variable contains output" { run printf "line1\nline2\nline3" [ "${lines[0]}" = "line1" ] [ "${lines[1]}" = "line2" ] [ "${lines[2]}" = "line3" ] } ``` ### File Assertions ```bash #!/usr/bin/env bats @test "File is created" { [ ! -f "$TMPDIR/output.txt" ] my_function > "$TMPDIR/output.txt" [ -f "$TMPDIR/output.txt" ] } @test "File contents match expected" { my_function > "$TMPDIR/output.txt" [ "$(cat "$TMPDIR/output.txt")" = "expected content" ] } @test "File is readable" { touch "$TMPDIR/test.txt" [ -r "$TMPDIR/test.txt" ] } @test "File has correct permissions" { touch "$TMPDIR/test.txt" chmod 644 "$TMPDIR/test.txt" [ "$(stat -f %OLp "$TMPDIR/test.txt")" = "644" ] } @test "File size is correct" { echo -n "12345" > "$TMPDIR/test.txt" [ "$(wc -c < "$TMPDIR/test.txt")" -eq 5 ] } ``` ## Setup and Teardown Patterns ### Basic Setup and Teardown ```bash #!/usr/bin/env bats setup() { # Create test directory TEST_DIR=$(mktemp -d) export TEST_DIR # Source script under test source "${BATS_TEST_DIRNAME}/../bin/script.sh" } teardown() { # Clean up temporary directory rm -rf "$TEST_DIR" } @test "Test using TEST_DIR" { touch "$TEST_DIR/file.txt" [ -f "$TEST_DIR/file.txt" ] } ``` ### Setup with Resources ```bash #!/usr/bin/env bats setup() { # Create directory structure mkdir -p "$TMPDIR/data/input" mkdir -p "$TMPDIR/data/output" # Create test fixtures echo "line1" > "$TMPDIR/data/input/file1.txt" echo "line2" > "$TMPDIR/data/input/file2.txt" # Initialize environment export DATA_DIR="$TMPDIR/data" export INPUT_DIR="$DATA_DIR/input" export OUTPUT_DIR="$DATA_DIR/output" } teardown() { rm -rf "$TMPDIR/data" } @test "Processes input files" { run my_process_script "$INPUT_DIR" "$OUTPUT_DIR" [ "$status" -eq 0 ] [ -f "$OUTPUT_DIR/file1.txt" ] } ``` ### Global Setup/Teardown ```bash #!/usr/bin/env bats # Load shared setup from test_helper.sh load test_helper # setup_file runs once before all tests setup_file() { export SHARED_RESOURCE=$(mktemp -d) echo "Expensive setup" > "$SHARED_RESOURCE/data.txt" } # teardown_file runs once after all tests teardown_file() { rm -rf "$SHARED_RESOURCE" } @test "First test uses shared resource" { [ -f "$SHARED_RESOURCE/data.txt" ] } @test "Second test uses shared resource" { [ -d "$SHARED_RESOURCE" ] } ``` ## Mocking and Stubbing Patterns ### Function Mocking ```bash #!/usr/bin/env bats # Mock external command my_external_tool() { echo "mocked output" return 0 } @test "Function uses mocked tool" { export -f my_external_tool run my_function [[ "$output" == *"mocked output"* ]] } ``` ### Command Stubbing ```bash #!/usr/bin/env bats setup() { # Create stub directory STUBS_DIR="$TMPDIR/stubs" mkdir -p "$STUBS_DIR" # Add to PATH export PATH="$STUBS_DIR:$PATH" } create_stub() { local cmd="$1" local output="$2" local code="${3:-0}" cat > "$STUBS_DIR/$cmd" <<EOF #!/bin/bash echo "$output" exit $code EOF chmod +x "$STUBS_DIR/$cmd" } @test "Function works with stubbed curl" { create_stub curl "{ \"status\": \"ok\" }" 0 run my_api_function [ "$status" -eq 0 ] } ``` ### Variable Stubbing ```bash #!/usr/bin/env bats @test "Function handles environment override" { export MY_SETTING="override_value" run my_function [ "$status" -eq 0 ] [[ "$output" == *"override_value"* ]] } @test "Function uses default when var unset" { unset MY_SETTING run my_function [ "$status" -eq 0 ] [[ "$output" == *"default"* ]] } ``` ## Fixture Management ### Using Fixture Files ```bash #!/usr/bin/env bats # Fixture directory: tests/fixtures/ setup() { FIXTURES_DIR="${BATS_TEST_DIRNAME}/fixtures" WORK_DIR=$(mktemp -d) export WORK_DIR } teardown() { rm -rf "$WORK_DIR" } @test "Process fixture file" { # Copy fixture to work directory cp "$FIXTURES_DIR/input.txt" "$WORK_DIR/input.txt" # Run function run my_process_function "$WORK_DIR/input.txt" # Compare output diff "$WORK_DIR/output.txt" "$FIXTURES_DIR/expected_output.txt" } ``` ### Dynamic Fixture Generation ```bash #!/usr/bin/env bats generate_fixture() { local lines="$1" local file="$2" for i in $(seq 1 "$lines"); do echo "Line $i content" >> "$file" done } @test "Handle large input file" { generate_fixture 1000 "$TMPDIR/large.txt" run my_function "$TMPDIR/large.txt" [ "$status" -eq 0 ] [ "$(wc -l < "$TMPDIR/large.txt")" -eq 1000 ] } ``` ## Advanced Patterns ### Testing Error Conditions ```bash #!/usr/bin/env bats @test "Function fails with missing file" { run my_function "/nonexistent/file.txt" [ "$status" -ne 0 ] [[ "$output" == *"not found"* ]] } @test "Function fails with invalid input" { run my_function "" [ "$status" -ne 0 ] } @test "Function fails with permission denied" { touch "$TMPDIR/readonly.txt" chmod 000 "$TMPDIR/readonly.txt" run my_function "$TMPDIR/readonly.txt" [ "$status" -ne 0 ] chmod 644 "$TMPDIR/readonly.txt" # Cleanup } @test "Function provides helpful error message" { run my_function --invalid-option [ "$status" -ne 0 ] [[ "$output" == *"Usage:"* ]] } ``` ### Testing with Dependencies ```bash #!/usr/bin/env bats setup() { # Check for required tools if ! command -v jq &>/dev/null; then skip "jq is not installed" fi export SCRIPT="${BATS_TEST_DIRNAME}/../bin/script.sh" } @test "JSON parsing works" { skip_if ! command -v jq &>/dev/null run my_json_parser '{"key": "value"}' [ "$status" -eq 0 ] } ``` ### Testing Shell Compatibility ```bash #!/usr/bin/env bats @test "Script works in bash" { bash "${BATS_TEST_DIRNAME}/../bin/script.sh" arg1 } @test "Script works in sh (POSIX)" { sh "${BATS_TEST_DIRNAME}/../bin/script.sh" arg1 } @test "Script works in dash" { if command -v dash &>/dev/null; then dash "${BATS_TEST_DIRNAME}/../bin/script.sh" arg1 else skip "dash not installed" fi } ``` ### Parallel Execution ```bash #!/usr/bin/env bats @test "Multiple independent operations" { run bash -c 'for i in {1..10}; do my_operation "$i" & done wait' [ "$status" -eq 0 ] } @test "Concurrent file operations" { for i in {1..5}; do my_function "$TMPDIR/file$i" & done wait [ -f "$TMPDIR/file1" ] [ -f "$TMPDIR/file5" ] } ``` ## Test Helper Pattern ### test_helper.sh ```bash #!/usr/bin/env bash # Source script under test export SCRIPT_DIR="${BATS_TEST_DIRNAME%/*}/bin" # Common test utilities assert_file_exists() { if [ ! -f "$1" ]; then echo "Expected file to exist: $1" return 1 fi } assert_file_equals() { local file="$1" local expected="$2" if [ ! -f "$file" ]; then echo "File does not exist: $file" return 1 fi local actual=$(cat "$file") if [ "$actual" != "$expected" ]; then echo "File contents do not match" echo "Expected: $expected" echo "Actual: $actual" return 1 fi } # Create temporary test directory setup_test_dir() { export TEST_DIR=$(mktemp -d) } cleanup_test_dir() { rm -rf "$TEST_DIR" } ``` ## Integration with CI/CD ### GitHub Actions Workflow ```yaml name: Tests on: [push, pull_request] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Install Bats run: | npm install --global bats - name: Run Tests run: | bats tests/*.bats - name: Run Tests with Tap Reporter run: | bats tests/*.bats --tap | tee test_output.tap ``` ### Makefile Integration ```makefile .PHONY: test test-verbose test-tap test: bats tests/*.bats test-verbose: bats tests/*.bats --verbose test-tap: bats tests/*.bats --tap test-parallel: bats tests/*.bats --parallel 4 coverage: test # Optional: Generate coverage reports ``` ## Best Practices 1. **Test one thing per test** - Single responsibility principle 2. **Use descriptive test names** - Clearly states what is being tested 3. **Clean up after tests** - Always remove temporary files in teardown 4. **Test both success and failure paths** - Don't just test happy path 5. **Mock external dependencies** - Isolate unit under test 6. **Use fixtures for complex data** - Makes tests more readable 7. **Run tests in CI/CD** - Catch regressions early 8. **Test across shell dialects** - Ensure portability 9. **Keep tests fast** - Run in parallel when possible 10. **Document complex test setup** - Explain unusual patterns ## Resources - **Bats GitHub**: https://github.com/bats-core/bats-core - **Bats Documentation**: https://bats-core.readthedocs.io/ - **TAP Protocol**: https://testanything.org/ - **Test-Driven Development**: https://en.wikipedia.org/wiki/Test-driven_development
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered