From 49ac1a39376bc2ff577393f4f61dd28cf5a8fd8d Mon Sep 17 00:00:00 2001 From: spalen0 Date: Sun, 4 Oct 2026 21:00:49 +0200 Subject: [PATCH 1/2] Add PendleSwap context to upgrade explanations --- protocols/pendle/README.md | 10 + tests/fixtures/pendle_upgrade/new.json | 58 ++++++ tests/fixtures/pendle_upgrade/old.json | 74 +++++++ tests/test_pendle_context.py | 254 +++++++++++++++++++++++++ tests/test_protocol_context.py | 2 +- utils/llm/README.md | 26 +++ utils/llm/ai_explainer.py | 12 +- utils/llm/pendle_context.py | 247 ++++++++++++++++++++++++ utils/llm/protocol_context.py | 6 + 9 files changed, 684 insertions(+), 5 deletions(-) create mode 100644 tests/fixtures/pendle_upgrade/new.json create mode 100644 tests/fixtures/pendle_upgrade/old.json create mode 100644 tests/test_pendle_context.py create mode 100644 utils/llm/pendle_context.py diff --git a/protocols/pendle/README.md b/protocols/pendle/README.md index 1cf6410c..37ca1d89 100644 --- a/protocols/pendle/README.md +++ b/protocols/pendle/README.md @@ -10,3 +10,13 @@ Additionally, other contracts like vePENDLE, PENDLE, RewardDistributor, and Voti Arbitrum Safe Multisig: 0x7877AdFaDEd756f3248a0EBfe8Ac2E2eF87b75Ac The owner of SY contracts was changed to [governance proxy contract](https://etherscan.io/address/0x2aD631F72fB16d91c4953A7f4260A97C2fE2f31e) with an additional guardian role that can only pause SY contracts. The governance proxy contract owner is multisig defined above. + +## AI governance context + +PendleSwap upgrade alerts on Ethereum and Arbitrum include the proxy's live owner, +current/proposed verified swap and authorization code, and a pinned reference for +the standard router integration. This distinguishes the optional aggregator leg +from market, PT/YT and SY contracts, and makes swap payload/enum changes visible +even when the external ABI is unchanged. The context is included in the AI prompt +and full report; it does not impose a risk rating. See +[the LLM context documentation](../../utils/llm/README.md#5f-4-pendleswap-upgrade-context-utilsllmpendle_contextpy). diff --git a/tests/fixtures/pendle_upgrade/new.json b/tests/fixtures/pendle_upgrade/new.json new file mode 100644 index 00000000..058c4340 --- /dev/null +++ b/tests/fixtures/pendle_upgrade/new.json @@ -0,0 +1,58 @@ +{ + "schema": 2, + "contract_name": "PendleSwap", + "compiler_version": "v0.8.30+commit.73712a01", + "language": "Solidity", + "sources": { + "lib/pendle-core-v2/contracts/router/swap-aggregator/PendleSwap.sol": "// SPDX-License-Identifier: GPL-3.0-or-later\npragma solidity ^0.8.17;\n\nimport \"../../core/libraries/TokenHelper.sol\";\nimport \"./IPSwapAggregator.sol\";\nimport \"./OKXScaleHelper.sol\";\nimport \"./ParaswapScaleHelper.sol\";\nimport \"@openzeppelin/contracts/token/ERC20/utils/SafeERC20.sol\";\n\nimport \"../../core/libraries/BoringOwnableUpgradeableV2.sol\";\nimport \"@openzeppelin/contracts/proxy/utils/UUPSUpgradeable.sol\";\n\ncontract PendleSwap is\n IPSwapAggregator,\n TokenHelper,\n OKXScaleHelper,\n ParaswapScaleHelper,\n BoringOwnableUpgradeableV2,\n UUPSUpgradeable\n{\n using Address for address;\n using SafeERC20 for IERC20;\n\n address private constant KYBER_SCALING_HELPER = 0x2f577A41BeC1BE1152AeEA12e73b7391d15f655D;\n\n constructor(bool okx_allowUnsupportedChain) OKXScaleHelper(okx_allowUnsupportedChain) {\n _disableInitializers();\n }\n\n function initialize(address _owner) external initializer {\n __BoringOwnableV2_init(_owner);\n }\n\n function swap(address tokenIn, uint256 amountIn, SwapData calldata data) external payable {\n _approveForExtRouter(tokenIn, data);\n if (data.swapType == SwapType.ZEROX) {\n _zeroExSwap(tokenIn, amountIn, data);\n } else {\n data.extRouter\n .functionCallWithValue(\n data.needScale ? _getScaledInputData(data.swapType, data.extCalldata, amountIn) : data.extCalldata,\n tokenIn == NATIVE ? amountIn : 0\n );\n }\n\n emit SwapSingle(data.swapType, tokenIn, amountIn);\n }\n\n /// @dev extCalldata = abi.encode(tokenOut, quotedAmountIn, quotedMinOut, execCalldata), where execCalldata is the\n /// 0x AllowanceHolder.exec calldata, quoted with sellEntireBalance and the taker as recipient. 0x's own minBuyAmount\n /// is a fixed number, so it is only a backstop. The binding minimum is enforced here, scaled to the actual amountIn\n /// if needScale. The output is forwarded to msg.sender.\n function _zeroExSwap(address tokenIn, uint256 amountIn, SwapData calldata data) internal {\n (address tokenOut, uint256 quotedAmountIn, uint256 quotedMinOut, bytes memory execCalldata) =\n abi.decode(data.extCalldata, (address, uint256, uint256, bytes));\n\n uint256 balBefore = _selfBalance(tokenOut);\n data.extRouter.functionCallWithValue(execCalldata, tokenIn == NATIVE ? amountIn : 0);\n uint256 netOut = _selfBalance(tokenOut) - balBefore;\n\n uint256 minOut = data.needScale ? (quotedMinOut * amountIn) / quotedAmountIn : quotedMinOut;\n require(netOut >= minOut, \"PendleSwap: 0x insufficient out\");\n\n _transferOut(tokenOut, msg.sender, netOut);\n }\n\n function _approveForExtRouter(address token, SwapData calldata data) internal {\n if (token == NATIVE) return;\n\n if (data.swapType == SwapType.OKX) {\n _safeApproveInfV2(IERC20(token), _okx_getTokenApprove());\n } else {\n _safeApproveInfV2(IERC20(token), data.extRouter);\n }\n }\n\n function _safeApproveInfV2(IERC20 token, address spender) internal {\n if (token.allowance(address(this), spender) < type(uint256).max) {\n token.forceApprove(spender, type(uint256).max);\n }\n }\n\n function _getScaledInputData(SwapType swapType, bytes calldata rawCallData, uint256 amountIn)\n internal\n view\n returns (bytes memory scaledCallData)\n {\n if (swapType == SwapType.KYBERSWAP) {\n bool isSuccess;\n (isSuccess, scaledCallData) =\n IKyberScalingHelper(KYBER_SCALING_HELPER).getScaledInputData(rawCallData, amountIn);\n\n require(isSuccess, \"PendleSwap: Kyber scaling failed\");\n } else if (swapType == SwapType.PARASWAP) {\n scaledCallData = _paraswapScaling(rawCallData, amountIn);\n } else if (swapType == SwapType.OKX) {\n scaledCallData = _okxScaling(rawCallData, amountIn);\n } else {\n assert(false);\n }\n }\n\n receive() external payable {}\n\n function _authorizeUpgrade(address) internal virtual override onlyOwner {}\n}\n\ninterface IKyberScalingHelper {\n function getScaledInputData(bytes calldata inputData, uint256 newAmount)\n external\n view\n returns (bool isSuccess, bytes memory data);\n}\n", + "lib/pendle-core-v2/contracts/router/swap-aggregator/IPSwapAggregator.sol": "// SPDX-License-Identifier: GPL-3.0-or-later\npragma solidity ^0.8.0;\n\nstruct SwapData {\n SwapType swapType;\n address extRouter;\n bytes extCalldata;\n bool needScale;\n}\n\nstruct SwapDataExtra {\n address tokenIn;\n address tokenOut;\n uint256 minOut;\n SwapData swapData;\n}\n\nenum SwapType {\n NONE,\n KYBERSWAP,\n RESERVE_1,\n // ETH_WETH not used in Aggregator\n ETH_WETH,\n OKX,\n ONE_INCH,\n PARASWAP,\n ZEROX,\n RESERVE_3,\n RESERVE_4,\n RESERVE_5\n}\n\ninterface IPSwapAggregator {\n event SwapSingle(SwapType indexed swapType, address indexed tokenIn, uint256 amountIn);\n\n function swap(address tokenIn, uint256 amountIn, SwapData calldata swapData) external payable;\n}\n", + "lib/pendle-core-v2/contracts/core/libraries/BoringOwnableUpgradeableV2.sol": "// SPDX-License-Identifier: GPL-3.0-or-later\npragma solidity ^0.8.0;\n\nimport \"@openzeppelin/contracts-upgradeable/proxy/utils/Initializable.sol\";\n\ncontract BoringOwnableUpgradeableData {\n address public owner;\n address public pendingOwner;\n}\n\nabstract contract BoringOwnableUpgradeableV2 is BoringOwnableUpgradeableData, Initializable {\n event OwnershipTransferred(address indexed previousOwner, address indexed newOwner);\n\n function __BoringOwnableV2_init(address _owner) internal onlyInitializing {\n owner = _owner;\n }\n\n /// @notice Transfers ownership to `newOwner`. Either directly or claimable by the new pending owner.\n /// Can only be invoked by the current `owner`.\n /// @param newOwner Address of the new owner.\n /// @param direct True if `newOwner` should be set immediately. False if `newOwner` needs to use `claimOwnership`.\n /// @param renounce Allows the `newOwner` to be `address(0)` if `direct` and `renounce` is True. Has no effect\n /// otherwise.\n function transferOwnership(address newOwner, bool direct, bool renounce) public onlyOwner {\n if (direct) {\n // Checks\n require(newOwner != address(0) || renounce, \"Ownable: zero address\");\n\n // Effects\n emit OwnershipTransferred(owner, newOwner);\n owner = newOwner;\n pendingOwner = address(0);\n } else {\n // Effects\n pendingOwner = newOwner;\n }\n }\n\n /// @notice Needs to be called by `pendingOwner` to claim ownership.\n function claimOwnership() public {\n address _pendingOwner = pendingOwner;\n\n // Checks\n require(msg.sender == _pendingOwner, \"Ownable: caller != pending owner\");\n\n // Effects\n emit OwnershipTransferred(owner, _pendingOwner);\n owner = _pendingOwner;\n pendingOwner = address(0);\n }\n\n /// @notice Only allows the `owner` to execute the function.\n modifier onlyOwner() {\n require(msg.sender == owner, \"Ownable: caller is not the owner\");\n _;\n }\n\n uint256[48] private __gap;\n}\n" + }, + "settings": { + "remappings": [ + "@pendle/core-v2/=lib/pendle-core-v2/", + "@pendle/sy/=lib/pendle-sy/", + "forge-std/=lib/forge-std/", + "vendor/=vendor/", + "vendor-locked/=vendor-locked/", + "@chainlink/=node_modules/@chainlink/", + "@eth-optimism/=node_modules/@eth-optimism/", + "@layerzerolabs/=node_modules/@layerzerolabs/", + "@openzeppelin/=node_modules/@openzeppelin/", + "hardhat/=node_modules/hardhat/", + "solidity-bytes-utils/=node_modules/solidity-bytes-utils/" + ], + "optimizer": { + "enabled": true, + "runs": 1000000 + }, + "metadata": { + "useLiteralContent": false, + "bytecodeHash": "ipfs", + "appendCBOR": true + }, + "outputSelection": { + "*": { + "*": [ + "evm.bytecode", + "evm.deployedBytecode", + "devdoc", + "userdoc", + "metadata", + "abi" + ] + } + }, + "evmVersion": "cancun", + "viaIR": true + }, + "abi": [], + "contract_file": "lib/pendle-core-v2/contracts/router/swap-aggregator/PendleSwap.sol", + "provenance": { + "chain_id": 1, + "address": "0xd14feb6aaf8650bbfcc8abec299b249a80fe7c78", + "retrieved": "2026-10-04", + "source": "Etherscan v2 getsourcecode", + "scope": "Target source, swap interface and ownership base only; not a complete compilation bundle." + } +} diff --git a/tests/fixtures/pendle_upgrade/old.json b/tests/fixtures/pendle_upgrade/old.json new file mode 100644 index 00000000..e5a80091 --- /dev/null +++ b/tests/fixtures/pendle_upgrade/old.json @@ -0,0 +1,74 @@ +{ + "schema": 2, + "contract_name": "PendleSwap", + "compiler_version": "v0.8.30+commit.73712a01", + "language": "Solidity", + "sources": { + "lib/pendle-core-v2/contracts/router/swap-aggregator/PendleSwap.sol": "// SPDX-License-Identifier: GPL-3.0-or-later\npragma solidity ^0.8.17;\n\nimport \"../../core/libraries/TokenHelper.sol\";\nimport \"./IPSwapAggregator.sol\";\nimport \"./OKXScaleHelper.sol\";\nimport \"./ParaswapScaleHelper.sol\";\nimport \"@openzeppelin/contracts/token/ERC20/utils/SafeERC20.sol\";\n\nimport \"../../core/libraries/BoringOwnableUpgradeableV2.sol\";\nimport \"@openzeppelin/contracts/proxy/utils/UUPSUpgradeable.sol\";\n\ncontract PendleSwap is\n IPSwapAggregator,\n TokenHelper,\n OKXScaleHelper,\n ParaswapScaleHelper,\n BoringOwnableUpgradeableV2,\n UUPSUpgradeable\n{\n using Address for address;\n using SafeERC20 for IERC20;\n\n address private constant KYBER_SCALING_HELPER = 0x2f577A41BeC1BE1152AeEA12e73b7391d15f655D;\n\n constructor(bool okx_allowUnsupportedChain) OKXScaleHelper(okx_allowUnsupportedChain) {\n _disableInitializers();\n }\n\n function initialize(address _owner) external initializer {\n __BoringOwnableV2_init(_owner);\n }\n\n function swap(address tokenIn, uint256 amountIn, SwapData calldata data) external payable {\n _approveForExtRouter(tokenIn, data);\n data.extRouter\n .functionCallWithValue(\n data.needScale ? _getScaledInputData(data.swapType, data.extCalldata, amountIn) : data.extCalldata,\n tokenIn == NATIVE ? amountIn : 0\n );\n\n emit SwapSingle(data.swapType, tokenIn, amountIn);\n }\n\n function _approveForExtRouter(address token, SwapData calldata data) internal {\n if (token == NATIVE) return;\n\n if (data.swapType == SwapType.OKX) {\n _safeApproveInfV2(IERC20(token), _okx_getTokenApprove());\n } else {\n _safeApproveInfV2(IERC20(token), data.extRouter);\n }\n }\n\n function _safeApproveInfV2(IERC20 token, address spender) internal {\n if (token.allowance(address(this), spender) < type(uint256).max) {\n token.forceApprove(spender, type(uint256).max);\n }\n }\n\n function _getScaledInputData(SwapType swapType, bytes calldata rawCallData, uint256 amountIn)\n internal\n view\n returns (bytes memory scaledCallData)\n {\n if (swapType == SwapType.KYBERSWAP) {\n bool isSuccess;\n (isSuccess, scaledCallData) =\n IKyberScalingHelper(KYBER_SCALING_HELPER).getScaledInputData(rawCallData, amountIn);\n\n require(isSuccess, \"PendleSwap: Kyber scaling failed\");\n } else if (swapType == SwapType.ODOS) {\n scaledCallData = _odosScaling(rawCallData, amountIn);\n } else if (swapType == SwapType.PARASWAP) {\n scaledCallData = _paraswapScaling(rawCallData, amountIn);\n } else if (swapType == SwapType.OKX) {\n scaledCallData = _okxScaling(rawCallData, amountIn);\n } else {\n assert(false);\n }\n }\n\n function _odosScaling(bytes calldata rawCallData, uint256 amountIn)\n internal\n pure\n returns (bytes memory scaledCallData)\n {\n bytes4 selector = bytes4(rawCallData[:4]);\n bytes calldata dataToDecode = rawCallData[4:];\n\n assert(selector == IOdosRouterV2.swap.selector);\n (\n IOdosRouterV2.swapTokenInfo memory tokenInfo,\n bytes memory pathDefinition,\n address executor,\n uint32 referralCode\n ) = abi.decode(dataToDecode, (IOdosRouterV2.swapTokenInfo, bytes, address, uint32));\n\n tokenInfo.outputQuote = (tokenInfo.outputQuote * amountIn) / tokenInfo.inputAmount;\n tokenInfo.outputMin = (tokenInfo.outputMin * amountIn) / tokenInfo.inputAmount;\n tokenInfo.inputAmount = amountIn;\n\n return abi.encodeWithSelector(selector, tokenInfo, pathDefinition, executor, referralCode);\n }\n\n receive() external payable {}\n\n function _authorizeUpgrade(address) internal virtual override onlyOwner {}\n}\n\ninterface IKyberScalingHelper {\n function getScaledInputData(bytes calldata inputData, uint256 newAmount)\n external\n view\n returns (bool isSuccess, bytes memory data);\n}\n\ninterface IOdosRouterV2 {\n struct swapTokenInfo {\n address inputToken;\n uint256 inputAmount;\n address inputReceiver;\n address outputToken;\n uint256 outputQuote;\n uint256 outputMin;\n address outputReceiver;\n }\n\n function swap(swapTokenInfo memory tokenInfo, bytes calldata pathDefinition, address executor, uint32 referralCode)\n external\n payable\n returns (uint256 amountOut);\n}\n", + "lib/pendle-core-v2/contracts/router/swap-aggregator/IPSwapAggregator.sol": "// SPDX-License-Identifier: GPL-3.0-or-later\npragma solidity ^0.8.0;\n\nstruct SwapData {\n SwapType swapType;\n address extRouter;\n bytes extCalldata;\n bool needScale;\n}\n\nstruct SwapDataExtra {\n address tokenIn;\n address tokenOut;\n uint256 minOut;\n SwapData swapData;\n}\n\nenum SwapType {\n NONE,\n KYBERSWAP,\n ODOS,\n // ETH_WETH not used in Aggregator\n ETH_WETH,\n OKX,\n ONE_INCH,\n PARASWAP,\n RESERVE_2,\n RESERVE_3,\n RESERVE_4,\n RESERVE_5\n}\n\ninterface IPSwapAggregator {\n event SwapSingle(SwapType indexed swapType, address indexed tokenIn, uint256 amountIn);\n\n function swap(address tokenIn, uint256 amountIn, SwapData calldata swapData) external payable;\n}\n", + "lib/pendle-core-v2/contracts/core/libraries/BoringOwnableUpgradeableV2.sol": "// SPDX-License-Identifier: GPL-3.0-or-later\npragma solidity ^0.8.0;\n\nimport \"@openzeppelin/contracts-upgradeable/proxy/utils/Initializable.sol\";\n\ncontract BoringOwnableUpgradeableData {\n address public owner;\n address public pendingOwner;\n}\n\nabstract contract BoringOwnableUpgradeableV2 is BoringOwnableUpgradeableData, Initializable {\n event OwnershipTransferred(address indexed previousOwner, address indexed newOwner);\n\n function __BoringOwnableV2_init(address _owner) internal onlyInitializing {\n owner = _owner;\n }\n\n /// @notice Transfers ownership to `newOwner`. Either directly or claimable by the new pending owner.\n /// Can only be invoked by the current `owner`.\n /// @param newOwner Address of the new owner.\n /// @param direct True if `newOwner` should be set immediately. False if `newOwner` needs to use `claimOwnership`.\n /// @param renounce Allows the `newOwner` to be `address(0)` if `direct` and `renounce` is True. Has no effect\n /// otherwise.\n function transferOwnership(address newOwner, bool direct, bool renounce) public onlyOwner {\n if (direct) {\n // Checks\n require(newOwner != address(0) || renounce, \"Ownable: zero address\");\n\n // Effects\n emit OwnershipTransferred(owner, newOwner);\n owner = newOwner;\n pendingOwner = address(0);\n } else {\n // Effects\n pendingOwner = newOwner;\n }\n }\n\n /// @notice Needs to be called by `pendingOwner` to claim ownership.\n function claimOwnership() public {\n address _pendingOwner = pendingOwner;\n\n // Checks\n require(msg.sender == _pendingOwner, \"Ownable: caller != pending owner\");\n\n // Effects\n emit OwnershipTransferred(owner, _pendingOwner);\n owner = _pendingOwner;\n pendingOwner = address(0);\n }\n\n /// @notice Only allows the `owner` to execute the function.\n modifier onlyOwner() {\n require(msg.sender == owner, \"Ownable: caller is not the owner\");\n _;\n }\n\n uint256[48] private __gap;\n}\n" + }, + "settings": { + "remappings": [ + "lib/pendle-core-v3/:@openzeppelin/contracts/=node_modules/@openzeppelin-v5/contracts/", + "lib/pendle-core-v3/:@openzeppelin/contracts-upgradeable/=node_modules/@openzeppelin-v5/contracts-upgradeable/", + "node_modules/@layerzerolabs/test-devtools-evm-foundry/:forge-std/=node_modules/forge-std/src/", + "node_modules/@layerzerolabs/oapp-evm/:@openzeppelin/=node_modules/@openzeppelin/", + "node_modules/@layerzerolabs/oft-evm/:@openzeppelin/=node_modules/@openzeppelin/", + "node_modules/@layerzerolabs/:@openzeppelin/=node_modules/@openzeppelin-v5/", + "@axelar-network/=node_modules/@axelar-network/", + "@chainlink/=node_modules/@chainlink/", + "@openzeppelin/contracts/=node_modules/@openzeppelin/contracts/", + "@openzeppelin/contracts-upgradeable/=node_modules/@openzeppelin/contracts-upgradeable/", + "@openzeppelin-v5/contracts/=node_modules/@openzeppelin-v5/contracts/", + "@openzeppelin-v5/contracts-upgradeable/=node_modules/@openzeppelin-v5/contracts-upgradeable/", + "@eth-optimism/=node_modules/@eth-optimism/", + "@prb/test/=node_modules/@prb/test/", + "forge-std/=node_modules/forge-std/", + "@pendle/core-v2/=lib/pendle-core-v2/", + "@pendle/core-v3/=lib/pendle-core-v3/", + "@pendle/sy/=lib/pendle-sy/", + "pendle-sy/=lib/pendle-sy/contracts/", + "hardhat-deploy/=node_modules/hardhat-deploy/", + "ds-test/=lib/surl/lib/forge-std/lib/ds-test/src/", + "solidity-bytes-utils/=node_modules/solidity-bytes-utils/", + "solidity-stringutils/=lib/surl/lib/solidity-stringutils/src/", + "surl/=lib/surl/src/", + "@layerzerolabs/=node_modules/@layerzerolabs/", + "pendle-core-v2/=lib/pendle-core-v2/contracts/", + "pendle-core-v3/=lib/pendle-core-v3/contracts/" + ], + "optimizer": { + "enabled": true, + "runs": 1000000 + }, + "metadata": { + "useLiteralContent": false, + "bytecodeHash": "ipfs", + "appendCBOR": true + }, + "outputSelection": { + "*": { + "*": [ + "evm.bytecode", + "evm.deployedBytecode", + "devdoc", + "userdoc", + "metadata", + "abi" + ] + } + }, + "evmVersion": "shanghai", + "viaIR": true + }, + "abi": [], + "contract_file": "lib/pendle-core-v2/contracts/router/swap-aggregator/PendleSwap.sol", + "provenance": { + "chain_id": 1, + "address": "0xbc17404b7bb500051c75c83e4aa5ae447d967811", + "retrieved": "2026-10-04", + "source": "Etherscan v2 getsourcecode", + "scope": "Target source, swap interface and ownership base only; not a complete compilation bundle." + } +} diff --git a/tests/test_pendle_context.py b/tests/test_pendle_context.py new file mode 100644 index 00000000..0defa2fc --- /dev/null +++ b/tests/test_pendle_context.py @@ -0,0 +1,254 @@ +"""PendleSwap upgrade context regressions using the 2026-10-04 verified sources.""" + +import json +from collections.abc import Iterator +from dataclasses import replace +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from utils.calldata.decoder import DecodedCall +from utils.llm import pendle_context +from utils.llm.ai_explainer import _build_prompt +from utils.llm.pendle_context import ( + PENDLE_SWAP, + _read_owner, + _source_evidence, + format_pendle_prompt, + format_pendle_report, + resolve_pendle_context, +) +from utils.llm.protocol_context import resolve_protocol_context +from utils.verified_contract import VerifiedContract + +OLD = "0xBC17404b7bb500051c75C83E4aA5aE447D967811" +NEW = "0xD14feb6Aaf8650BbfcC8aBEc299B249a80FE7C78" +OWNER = "0x8119EC16F0573B7dAc7C0CB94EB504FB32456ee1" +type Boundaries = tuple[MagicMock, MagicMock, MagicMock] + + +def _record(name: str) -> VerifiedContract: + """Load the scoped verified bundle fixture.""" + data = json.loads((Path(__file__).parent / "fixtures" / "pendle_upgrade" / f"{name}.json").read_text()) + record = VerifiedContract.from_cache_dict(data) + assert record is not None + return record + + +def _upgrade(migrate: bool = False) -> DecodedCall: + """A decoded upgrade with or without an initialization payload.""" + if migrate: + return DecodedCall( + "upgradeToAndCall", "upgradeToAndCall(address,bytes)", [("address", NEW), ("bytes", b"\x01")] + ) + return DecodedCall("upgradeTo", "upgradeTo(address)", [("address", NEW)]) + + +@pytest.fixture +def boundaries() -> Iterator[Boundaries]: + """Replace RPC and source fetches while retaining all context resolution logic.""" + with ( + patch.object(pendle_context, "get_current_implementation", return_value=OLD) as implementation, + patch.object(pendle_context, "fetch_verified_contract") as source, + patch.object(pendle_context.ChainManager, "get_client") as client, + ): + source.side_effect = lambda chain, address: _record("old" if address.lower() == OLD.lower() else "new") + client.return_value.eth.contract.return_value.functions.owner.return_value.call.return_value = OWNER + yield implementation, source, client + + +@pytest.mark.parametrize("chain_id", [1, 42161]) +def test_upgrade_includes_scope_controls_and_route_semantics(boundaries: Boundaries, chain_id: int) -> None: + """Unchanged ABI must not hide enum changes, unsupported scaling or existing authorization.""" + contexts = resolve_pendle_context("PENDLE", chain_id, [(PENDLE_SWAP.lower(), _upgrade())]) + assert len(contexts) == 1 + prompt = format_pendle_prompt(contexts) + assert "optional aggregator leg" in prompt + assert "NONE and ETH_WETH branches bypass" in prompt + assert "calling router for the next step" in prompt + assert "not a live trace" in prompt + assert "not proof every caller uses a nonzero minimum" in prompt + assert f"Current proxy owner() (live read): {OWNER}" in prompt + assert "_authorizeUpgrade(address) internal virtual override onlyOwner" in prompt + assert 'require(msg.sender == owner, "Ownable: caller is not the owner")' in prompt + assert "type(uint256).max" in prompt + before, after = prompt.split("Proposed implementation evidence:") + assert "SwapType.ODOS" in before + assert "RESERVE_2" in before + assert "RESERVE_1" in after + assert "ZEROX" in after + assert "SwapType.ODOS" not in after + assert "assert(false)" in after + assert "(quotedMinOut * amountIn) / quotedAmountIn" in after + assert "_transferOut(tokenOut, msg.sender, netOut)" in after + assert "No fixed risk rating" in prompt + assert "removes ODOS calldata scaling" in prompt + assert "does not by itself establish a higher likelihood of storage collisions" in prompt + + +def test_registry_publishes_context_and_full_address_links(boundaries: Boundaries) -> None: + """The adapter's facts and additional owner/implementation addresses reach the report.""" + resolved = resolve_protocol_context("pendle", 1, [(PENDLE_SWAP, _upgrade())]) + assert "optional aggregator leg" in resolved.prompt + assert "optional aggregator leg" in resolved.report + for address in (PENDLE_SWAP, OLD, NEW, OWNER): + assert address in resolved.addresses + assert f"https://etherscan.io/address/{address}" in resolved.report + assert resolved.labels[PENDLE_SWAP] == "PendleSwap" + + +def test_prompt_preserves_integration_reference_and_live_observation_distinction(boundaries: Boundaries) -> None: + """The explainer must not relabel documented router flow as a live execution trace.""" + contexts = resolve_pendle_context("pendle", 1, [(PENDLE_SWAP, _upgrade())]) + prompt = _build_prompt( + target=PENDLE_SWAP, + value=0, + decoded_calls=[_upgrade()], + simulation=None, + protocol_context=format_pendle_prompt(contexts), + ) + assert "Distinguish documented integration architecture from live observations" in prompt + assert "not a live trace" in prompt + + +def test_arbitrum_report_uses_arbitrum_links(boundaries: Boundaries) -> None: + """Same deployment addresses must link to the actual alert chain.""" + contexts = resolve_pendle_context("pendle", 42161, [(PENDLE_SWAP, _upgrade())]) + report = format_pendle_report(contexts, 42161, {}) + assert f"https://arbiscan.io/address/{PENDLE_SWAP}" in report + assert "etherscan.io/address" not in report + + +@pytest.mark.parametrize( + "protocol,chain,target,call", + [ + ("yearn", 1, PENDLE_SWAP, _upgrade()), + ("pendle", 10, PENDLE_SWAP, _upgrade()), + ("pendle", 1, OWNER, _upgrade()), + ("pendle", 1, PENDLE_SWAP, DecodedCall("swap", "swap(address,uint256)", [])), + ("pendle", 1, PENDLE_SWAP, DecodedCall("upgradeTo", "upgradeTo(address)", [])), + ("pendle", 1, PENDLE_SWAP, DecodedCall("upgradeTo", "upgradeTo(address)", [("uint256", 1)])), + ("pendle", 1, PENDLE_SWAP, DecodedCall("upgradeTo", "upgradeTo(address)", [("address", "bad")])), + ], +) +def test_unrelated_or_malformed_calls_make_no_network_requests( + boundaries: Boundaries, + protocol: str, + chain: int, + target: str, + call: DecodedCall, +) -> None: + """Scope checks run before all RPC/source work.""" + assert resolve_pendle_context(protocol, chain, [(target, call)]) == [] + for boundary in boundaries: + boundary.assert_not_called() + + +def test_empty_batch_makes_no_network_requests(boundaries: Boundaries) -> None: + """Empty alerts need no enrichment.""" + assert resolve_pendle_context("pendle", 1, []) == [] + for boundary in boundaries: + boundary.assert_not_called() + + +def test_duplicate_calls_and_migration_are_distinguished(boundaries: Boundaries) -> None: + """A Safe batch retains migration context while repeated identical upgrades are coalesced.""" + contexts = resolve_pendle_context( + "pendle", + 1, + [ + (PENDLE_SWAP, _upgrade()), + (PENDLE_SWAP, _upgrade()), + (PENDLE_SWAP, _upgrade(True)), + ], + ) + assert len(contexts) == 2 + prompt = format_pendle_prompt(contexts) + assert "upgradeTo has no initialization/migration payload" in prompt + assert "upgradeToAndCall includes a bytes payload" in prompt + + +def test_owner_read_failure_preserves_source_evidence(boundaries: Boundaries) -> None: + """An unavailable owner is explicit and cannot remove the code comparison.""" + boundaries[2].return_value.eth.contract.return_value.functions.owner.return_value.call.side_effect = RuntimeError( + "RPC failed" + ) + contexts = resolve_pendle_context("pendle", 1, [(PENDLE_SWAP, _upgrade())]) + prompt = format_pendle_prompt(contexts) + assert "Current proxy owner() (live read): unavailable" in prompt + assert "_zeroExSwap" in prompt + assert contexts[0].owner is None + + +def test_owner_is_read_from_proxy_not_implementation() -> None: + """Ownership state lives at the proxy address.""" + client = MagicMock() + client.eth.contract.return_value.functions.owner.return_value.call.return_value = OWNER + with patch.object(pendle_context.ChainManager, "get_client", return_value=client): + assert _read_owner(1, PENDLE_SWAP) == OWNER + assert client.eth.contract.call_args.kwargs["address"] == PENDLE_SWAP + + +def test_missing_source_and_implementation_are_explicit(boundaries: Boundaries) -> None: + """Missing evidence never turns into a guessed old implementation or safe-storage claim.""" + boundaries[0].return_value = None + boundaries[1].side_effect = None + boundaries[1].return_value = None + contexts = resolve_pendle_context("pendle", 1, [(PENDLE_SWAP, _upgrade())]) + prompt = format_pendle_prompt(contexts) + assert "Current implementation: unavailable" in prompt + assert prompt.count("target source unavailable") == 2 + assert "Storage UNKNOWN is a validation gap, not a proven collision" in prompt + + +def test_failed_batch_member_does_not_hide_next_upgrade(boundaries: Boundaries) -> None: + """Source/RPC errors remain local to a batch member.""" + boundaries[0].side_effect = [RuntimeError("RPC failed"), OLD] + contexts = resolve_pendle_context("pendle", 1, [(PENDLE_SWAP, _upgrade()), (PENDLE_SWAP, _upgrade(True))]) + assert len(contexts) == 1 + assert contexts[0].is_upgrade_and_call + + +def test_source_excerpts_are_scoped_to_the_deployed_contract() -> None: + """A same-name member in another bundled contract must not become PendleSwap evidence.""" + record = _record("new") + assert record.contract_file is not None + sources = dict(record.sources) + sources[record.contract_file] += '\ncontract Decoy { function swap() external { revert("DECOY"); } }' + assert "DECOY" not in _source_evidence(replace(record, sources=sources)) + + +def test_flattened_source_does_not_include_unrelated_contracts() -> None: + """Imported swap declarations must not cause an entire flattened bundle to enter the prompt.""" + record = _record("new") + flattened = "\n".join(record.sources.values()) + flattened += '\ncontract Decoy { function swap() external { revert("DECOY"); } }' + evidence = _source_evidence(replace(record, sources={"flat.sol": flattened}, contract_file="flat.sol")) + assert "DECOY" not in evidence + assert "enum SwapType" in evidence + assert "struct SwapData" in evidence + assert "_zeroExSwap" in evidence + assert 'require(msg.sender == owner, "Ownable: caller is not the owner")' in evidence + + +def test_other_replacement_and_unresolved_target_are_not_assumed_to_be_pendle() -> None: + """Labels and imported interfaces cannot establish replacement behavior.""" + record = _record("new") + for candidate in (None, replace(record, contract_name="OtherContract"), replace(record, contract_file=None)): + evidence = _source_evidence(candidate) + assert "unavailable" in evidence + assert "_zeroExSwap" not in evidence + + +def test_ambiguous_interfaces_and_owner_bases_are_not_guessed() -> None: + """Duplicate declarations in a bundle leave their facts unresolved.""" + record = _record("new") + sources = dict(record.sources) + for path, source in record.sources.items(): + if path.endswith(("IPSwapAggregator.sol", "BoringOwnableUpgradeableV2.sol")): + sources[f"decoy/{path}"] = source + evidence = _source_evidence(replace(record, sources=sources)) + assert "Verified swap payload/enum" not in evidence + assert "Verified ownership guard" not in evidence + assert "_zeroExSwap" in evidence diff --git a/tests/test_protocol_context.py b/tests/test_protocol_context.py index b50f6a8d..bc0986f6 100644 --- a/tests/test_protocol_context.py +++ b/tests/test_protocol_context.py @@ -81,7 +81,7 @@ def capture(contexts: list, chain_id: int, labels: dict[str, str]) -> str: def test_registered_adapters_cover_the_known_protocols(self) -> None: self.assertEqual( {adapter.name for adapter in protocol_context._ADAPTERS}, - {"infinifi", "infinifi-outland", "3jane", "yearn-v3", "control-transfer"}, + {"infinifi", "infinifi-outland", "3jane", "pendle", "yearn-v3", "control-transfer"}, ) diff --git a/utils/llm/README.md b/utils/llm/README.md index 50736530..564ec000 100644 --- a/utils/llm/README.md +++ b/utils/llm/README.md @@ -281,6 +281,31 @@ For any protocol, calls that hand over control (`set_management`, `transferOwner Involved Safes get `Safe m-of-n` labels. These are applied with `setdefault`, so curated names win. Failures are best-effort and never block the alert. +### 5f-4. PendleSwap Upgrade Context (`utils/llm/pendle_context.py`) + +For Pendle alerts on Ethereum and Arbitrum, direct `upgradeTo` and `upgradeToAndCall` +calls to the published PendleSwap proxy (including calls in Safe multisend batches) +receive adapter-specific context. This identifies the optional aggregator leg separately +from markets, PT/YT and SY contracts, and cites a pinned upstream `ActionBase` integration +reference for input pre-funding, output returned to the calling router, aggregator bypass +branches and caller-supplied output constraints. The architecture reference is explicitly +separate from a live execution trace or a claim about current balances or route usage. + +The adapter reads `owner()` at the proxy and includes complete, contract-scoped verified +members from both current and proposed implementations: swap dispatch, output handling, +scaling, approvals and the upgrade authorization hook. The verified swap payload/enum and +ownership guard are also included when uniquely resolved in the bundle. This exposes +semantic enum changes despite an unchanged ABI, unsupported scaling reverts, and unchanged +authorization/approval code that a diff alone omits. Missing source or owner reads remain +explicitly unavailable; a replacement with a different contract name is not assumed to +retain PendleSwap behavior. Context is rendered in both the prompt and gist and sets no +fixed risk tag or storage-safety verdict. Nested governance wrappers are not resolved by +this adapter. + +The summary critique also checks functional claims against supplied code: removing a +scaling branch does not establish removal of unscaled routes, and an UNKNOWN storage +verdict is not evidence that collisions are more likely. + ### 5g. Adapter Registry (`utils/llm/protocol_context.py`) Adapters register in `_ADAPTERS`; `resolve_protocol_context()` fans one call out to all of them and merges the rendered prompt text, report text, introduced addresses, and address labels. Each adapter guards itself, so registration order carries no meaning and one adapter raising is logged and skipped rather than dropping the alert. Most guard on protocol and chain. The Yearn V3 adapter guards on call shape and `apiVersion()`, and the control-transfer adapter on call shape alone. @@ -548,6 +573,7 @@ utils/llm/ ├── factory.py # Provider factory with env-based config + singleton ├── infinifi_context.py # Infinifi adapter: escrow → farm, custody, setRate APR, whitelist calls ├── openai_compat.py # OpenAI-compatible provider (Venice, OpenAI, etc.) +├── pendle_context.py # PendleSwap upgrades: router integration, owner, complete source members ├── protocol_context.py # Registry fanning one call out to every protocol adapter ├── report.py # Gist report: metadata header + deterministic call flow + analysis ├── threejane_abi.py # 3Jane checked-in ABIs + verified-ABI probes (proxy-aware) diff --git a/utils/llm/ai_explainer.py b/utils/llm/ai_explainer.py index 08444508..8817115f 100644 --- a/utils/llm/ai_explainer.py +++ b/utils/llm/ai_explainer.py @@ -280,6 +280,9 @@ confirmed? 5. Does the risk tag match the magnitude of change shown in the context? (A 10× change to a critical parameter is rarely LOW; a no-op is rarely HIGH.) +6. Are functional claims supported by the full supplied code and context? Do not turn + removal of one scaling branch into removal of all unscaled routes, or treat UNKNOWN + storage compatibility as evidence of a collision or increased collision likelihood. Hard rules for the revision (if you choose to revise): - Do NOT introduce a unit/scale assumption that wasn't supported by the context. @@ -290,7 +293,7 @@ - Do NOT remove an explicit hedge ("unit cannot be confirmed", "without source context", etc.). - Do NOT polish for style alone. Only edit if there's a concrete, specific issue - from items 1-5. + from items 1-6. If every check is satisfied AND no hard rule would be violated by the draft as-is, output exactly: @@ -1567,9 +1570,10 @@ def _build_prompt( if protocol_context: parts.append( - "\n--- Protocol Context (computed from protocol APIs and live on-chain reads) ---\n" - "Every fact below is VERIFIED for this protocol: identities, resolved hashes, decimals, " - "and current values. State them; do not hedge about them or call them unavailable.\n" + protocol_context + "\n--- Protocol Context (verified source, integration references and live on-chain reads) ---\n" + "Resolved identities, hashes, decimals, units and current values are VERIFIED facts. " + "State supplied facts; do not call them unavailable. Distinguish documented integration " + "architecture from live observations, and preserve any explicit validation limits.\n" + protocol_context ) if source_contexts: diff --git a/utils/llm/pendle_context.py b/utils/llm/pendle_context.py new file mode 100644 index 00000000..fa1a4526 --- /dev/null +++ b/utils/llm/pendle_context.py @@ -0,0 +1,247 @@ +"""Enrich PendleSwap upgrades with integration scope and verified implementation code.""" + +import re +from dataclasses import dataclass + +from eth_utils import to_checksum_address + +from utils.calldata.decoder import DecodedCall +from utils.chains import Chain +from utils.llm.report import address_link, checksum_or_none +from utils.logger import get_logger +from utils.proxy import get_current_implementation +from utils.solidity_text import contract_functions, declares_contract, strip_noise, struct_definitions +from utils.source_context import fetch_verified_contract +from utils.verified_contract import VerifiedContract +from utils.web3_wrapper import ChainManager + +logger = get_logger("utils.llm.pendle_context") + +# Pendle's published deployments/1-core.json and deployments/42161-core.json. +PENDLE_SWAP = "0xd4F480965D2347d421F1bEC7F545682E5Ec2151D" +SUPPORTED_CHAINS = {1, 42161} +_REFERENCE_BASE = ( + "https://github.com/pendle-finance/pendle-core-v2-public/blob/87685c89d05087535e9b9647eeda0e3d297d1f06" +) +_ROUTER_REFERENCE = f"{_REFERENCE_BASE}/contracts/router/base/ActionBase.sol" +_INTEGRATION_CONTEXT = ( + "PendleSwap is the external swap-aggregator adapter, separate from Pendle markets, PT/YT " + "contracts and SY implementations. This call upgrades the adapter only.\n" + "Integration reference (standard ActionBase router flow, not a live trace): the router " + "pre-funds PendleSwap with ERC20 input, or sends native input with the swap call. " + "It calls PendleSwap as an optional aggregator leg when minting/redeeming SY. " + "NONE and ETH_WETH branches bypass this adapter. Output returned to msg.sender in this " + "flow goes to the calling router for the next step, not necessarily to the end user. " + "The redeem flow separately checks TokenOutput.minTokenOut, and the mint flow supplies " + "minSyOut to SY.deposit. These are caller-supplied constraints, not proof every caller " + "uses a nonzero minimum.\n" + "Do not infer current balances, route usage, custody guarantees or the impact on all " + "Pendle deposits from this architecture. Assess the affected routes using the old/new " + "code below and the Proxy Upgrade diff; unchanged ABI does not prove unchanged payload " + "semantics. Removing an ODOS scaling branch does not establish removal of all ODOS " + "execution: inspect the needScale=false external-router dispatch separately. Describe " + "a scaler removal as 'removes ODOS calldata scaling' unless the code proves all routes " + "are disabled. Storage UNKNOWN is a validation gap, not a proven collision, and does " + "not by itself establish a higher likelihood of storage collisions. An upgrade-only " + "simulation does not test subsequent swaps. No fixed risk rating follows from this context." +) +_SOURCE_MEMBERS = { + "swap", + "_zeroExSwap", + "_getScaledInputData", + "_approveForExtRouter", + "_safeApproveInfV2", + "_authorizeUpgrade", +} +_OWNER_ABI = [ + { + "type": "function", + "name": "owner", + "stateMutability": "view", + "inputs": [], + "outputs": [{"type": "address", "name": ""}], + } +] + + +@dataclass(frozen=True) +class PendleSwapContext: + """Implementation evidence and current proxy owner for one adapter upgrade.""" + + proxy: str + implementation: str + current_implementation: str | None + owner: str | None + current_source: str + proposed_source: str + is_upgrade_and_call: bool + + @property + def addresses(self) -> list[str]: + """Addresses introduced by this context.""" + return [ + value + for value in ( + self.proxy, + self.implementation, + self.current_implementation, + self.owner, + ) + if value + ] + + @property + def labels(self) -> dict[str, str]: + """Name the known proxy without assuming the replacement's identity.""" + return {self.proxy: "PendleSwap"} + + +def _source_evidence(record: VerifiedContract | None) -> str: + """Extract complete members from the deployed target, plus its enum and ownership guard.""" + if record is None or record.contract_name != "PendleSwap" or not record.compilation_target: + return "PendleSwap target source unavailable or replacement is a different contract." + sections = [f"Verified target: {record.contract_name} in {record.contract_file}"] + members = contract_functions(record.target_source, record.contract_name) or [] + for member in members: + if member.name in _SOURCE_MEMBERS: + sections.append(record.target_source[slice(*member.span)]) + sections.extend(_supporting_evidence(record)) + return "\n\n".join(sections) + + +def _supporting_evidence(record: VerifiedContract) -> list[str]: + """Read uniquely declared payload and ownership definitions from the verified bundle.""" + sections: list[str] = [] + for name in ("IPSwapAggregator", "BoringOwnableUpgradeableV2"): + matches = [(path, source) for path, source in record.sources.items() if declares_contract(source, name)] + if len(matches) != 1: + continue + path, source = matches[0] + if name == "IPSwapAggregator": + sections.append(f"Verified swap payload/enum in {path}:\n{_payload_evidence(source)}") + else: + for member in contract_functions(source, name) or []: + if member.name == "onlyOwner": + sections.append(f"Verified ownership guard in {path}:\n{source[slice(*member.span)]}") + return sections + + +def _payload_evidence(source: str) -> str: + """Extract only swap declarations, even when verification provides a flattened file.""" + sections = [struct_definitions(source, None).get("SwapData", "SwapData declaration unavailable.")] + enums = list(re.finditer(r"\benum\s+SwapType\s*\{[^{}]*\}", strip_noise(source))) + if len(enums) == 1: + sections.append(source[enums[0].start() : enums[0].end()]) + else: + sections.append("SwapType enum unavailable or ambiguous.") + return "\n".join(sections) + + +def _read_owner(chain_id: int, proxy: str) -> str | None: + """Read ownership at the proxy; unavailable state must not hide the source evidence.""" + try: + client = ChainManager.get_client(Chain.from_chain_id(chain_id)) + contract = client.eth.contract(address=to_checksum_address(proxy), abi=_OWNER_ABI) + return to_checksum_address(contract.functions.owner().call()) + except Exception as error: # noqa: BLE001 - optional enrichment must not block the alert + logger.info("PendleSwap owner read failed on chain %s for %s: %s", chain_id, proxy, error) + return None + + +def resolve_pendle_context( + protocol: str, + chain_id: int, + targets_and_calls: list[tuple[str, DecodedCall]], +) -> list[PendleSwapContext]: + """Resolve direct PendleSwap upgrades on Ethereum and Arbitrum, including Safe batches. + + Args: + protocol: Alert protocol, matched case-insensitively. + chain_id: Chain containing the published PendleSwap proxy. + targets_and_calls: Decoded calls and their targets. + + Returns: + Integration context, verified source excerpts and live owner for each + distinct upgrade. Failed enrichments are logged without blocking alerts. + """ + if protocol.lower() != "pendle" or chain_id not in SUPPORTED_CHAINS: + return [] + contexts: list[PendleSwapContext] = [] + seen: set[tuple[str, bool]] = set() + for target, call in targets_and_calls: + implementation = _upgrade_implementation(target, call) + if implementation is None: + continue + is_upgrade_and_call = call.signature == "upgradeToAndCall(address,bytes)" + key = (implementation.lower(), is_upgrade_and_call) + if key in seen: + continue + seen.add(key) + try: + current = get_current_implementation(target, chain_id) + contexts.append( + PendleSwapContext( + proxy=to_checksum_address(target), + implementation=implementation, + current_implementation=current, + owner=_read_owner(chain_id, target), + current_source=_source_evidence(fetch_verified_contract(chain_id, current) if current else None), + proposed_source=_source_evidence(fetch_verified_contract(chain_id, implementation)), + is_upgrade_and_call=is_upgrade_and_call, + ) + ) + except Exception as error: # noqa: BLE001 - retain other batch members on enrichment failure + logger.info("PendleSwap context failed on chain %s for %s: %s", chain_id, implementation, error) + return contexts + + +def _upgrade_implementation(target: str, call: DecodedCall) -> str | None: + """Accept only well-formed upgrade arguments targeting the published swap adapter.""" + if target.lower() != PENDLE_SWAP.lower() or call.signature not in { + "upgradeTo(address)", + "upgradeToAndCall(address,bytes)", + }: + return None + expected_types = ("address", "bytes") if call.signature == "upgradeToAndCall(address,bytes)" else ("address",) + if tuple(param_type for param_type, _ in call.params) != expected_types: + return None + return checksum_or_none(call.params[0][1]) + + +def format_pendle_prompt(contexts: list[PendleSwapContext]) -> str: + """Render integration references separately from live state and verified code.""" + return "\n\n".join(_format_context(context, None, {}) for context in contexts) + + +def format_pendle_report( + contexts: list[PendleSwapContext], + chain_id: int, + labels: dict[str, str], +) -> str: + """Render the same evidence with full explorer links for the gist report.""" + return "\n\n".join(_format_context(context, chain_id, labels) for context in contexts) + + +def _format_context(context: PendleSwapContext, chain_id: int | None, labels: dict[str, str]) -> str: + """Keep prompt and report facts identical, varying only address rendering.""" + + def link(address: str | None) -> str: + """Link known addresses and keep missing state explicit.""" + return address_link(address, chain_id, labels) if address and chain_id else address or "unavailable" + + execution = ( + "upgradeToAndCall includes a bytes payload; inspect its initialization/migration effects separately." + if context.is_upgrade_and_call + else "upgradeTo has no initialization/migration payload." + ) + return ( + f"PendleSwap adapter upgrade on {link(context.proxy)}\n{_INTEGRATION_CONTEXT}\n" + f"Router integration reference: {_ROUTER_REFERENCE}\n" + f"Current implementation: {link(context.current_implementation)}\n" + f"Proposed implementation: {link(context.implementation)}\n" + f"Current proxy owner() (live read): {link(context.owner)}\n{execution}\n" + "Authorization must be assessed from the hook AND ownership guard below; " + "a modifier name alone is not proof.\n" + f"Current implementation evidence:\n```solidity\n{context.current_source}\n```\n" + f"Proposed implementation evidence:\n```solidity\n{context.proposed_source}\n```" + ) diff --git a/utils/llm/protocol_context.py b/utils/llm/protocol_context.py index 00d821d6..90e3619a 100644 --- a/utils/llm/protocol_context.py +++ b/utils/llm/protocol_context.py @@ -36,6 +36,11 @@ format_outland_report, resolve_outland_context, ) +from utils.llm.pendle_context import ( + format_pendle_prompt, + format_pendle_report, + resolve_pendle_context, +) from utils.llm.threejane_context import ( format_threejane_prompt, format_threejane_report, @@ -65,6 +70,7 @@ class _Adapter: _Adapter("infinifi", resolve_infinifi_context, format_infinifi_prompt, format_infinifi_report), _Adapter("infinifi-outland", resolve_outland_context, format_outland_prompt, format_outland_report), _Adapter("3jane", resolve_threejane_context, format_threejane_prompt, format_threejane_report), + _Adapter("pendle", resolve_pendle_context, format_pendle_prompt, format_pendle_report), _Adapter("yearn-v3", resolve_yearn_v3_context, format_yearn_v3_prompt, format_yearn_v3_report), _Adapter( "control-transfer", From 23dcdb0996de286e0d16bec86756b61f5b5e8db2 Mon Sep 17 00:00:00 2001 From: spalen0 Date: Mon, 5 Oct 2026 08:44:44 +0200 Subject: [PATCH 2/2] Simplify Pendle context test fixtures --- tests/fixtures/pendle_upgrade/new.json | 58 ----------- tests/fixtures/pendle_upgrade/old.json | 74 ------------- tests/test_pendle_context.py | 137 +++++++++++-------------- 3 files changed, 58 insertions(+), 211 deletions(-) delete mode 100644 tests/fixtures/pendle_upgrade/new.json delete mode 100644 tests/fixtures/pendle_upgrade/old.json diff --git a/tests/fixtures/pendle_upgrade/new.json b/tests/fixtures/pendle_upgrade/new.json deleted file mode 100644 index 058c4340..00000000 --- a/tests/fixtures/pendle_upgrade/new.json +++ /dev/null @@ -1,58 +0,0 @@ -{ - "schema": 2, - "contract_name": "PendleSwap", - "compiler_version": "v0.8.30+commit.73712a01", - "language": "Solidity", - "sources": { - "lib/pendle-core-v2/contracts/router/swap-aggregator/PendleSwap.sol": "// SPDX-License-Identifier: GPL-3.0-or-later\npragma solidity ^0.8.17;\n\nimport \"../../core/libraries/TokenHelper.sol\";\nimport \"./IPSwapAggregator.sol\";\nimport \"./OKXScaleHelper.sol\";\nimport \"./ParaswapScaleHelper.sol\";\nimport \"@openzeppelin/contracts/token/ERC20/utils/SafeERC20.sol\";\n\nimport \"../../core/libraries/BoringOwnableUpgradeableV2.sol\";\nimport \"@openzeppelin/contracts/proxy/utils/UUPSUpgradeable.sol\";\n\ncontract PendleSwap is\n IPSwapAggregator,\n TokenHelper,\n OKXScaleHelper,\n ParaswapScaleHelper,\n BoringOwnableUpgradeableV2,\n UUPSUpgradeable\n{\n using Address for address;\n using SafeERC20 for IERC20;\n\n address private constant KYBER_SCALING_HELPER = 0x2f577A41BeC1BE1152AeEA12e73b7391d15f655D;\n\n constructor(bool okx_allowUnsupportedChain) OKXScaleHelper(okx_allowUnsupportedChain) {\n _disableInitializers();\n }\n\n function initialize(address _owner) external initializer {\n __BoringOwnableV2_init(_owner);\n }\n\n function swap(address tokenIn, uint256 amountIn, SwapData calldata data) external payable {\n _approveForExtRouter(tokenIn, data);\n if (data.swapType == SwapType.ZEROX) {\n _zeroExSwap(tokenIn, amountIn, data);\n } else {\n data.extRouter\n .functionCallWithValue(\n data.needScale ? _getScaledInputData(data.swapType, data.extCalldata, amountIn) : data.extCalldata,\n tokenIn == NATIVE ? amountIn : 0\n );\n }\n\n emit SwapSingle(data.swapType, tokenIn, amountIn);\n }\n\n /// @dev extCalldata = abi.encode(tokenOut, quotedAmountIn, quotedMinOut, execCalldata), where execCalldata is the\n /// 0x AllowanceHolder.exec calldata, quoted with sellEntireBalance and the taker as recipient. 0x's own minBuyAmount\n /// is a fixed number, so it is only a backstop. The binding minimum is enforced here, scaled to the actual amountIn\n /// if needScale. The output is forwarded to msg.sender.\n function _zeroExSwap(address tokenIn, uint256 amountIn, SwapData calldata data) internal {\n (address tokenOut, uint256 quotedAmountIn, uint256 quotedMinOut, bytes memory execCalldata) =\n abi.decode(data.extCalldata, (address, uint256, uint256, bytes));\n\n uint256 balBefore = _selfBalance(tokenOut);\n data.extRouter.functionCallWithValue(execCalldata, tokenIn == NATIVE ? amountIn : 0);\n uint256 netOut = _selfBalance(tokenOut) - balBefore;\n\n uint256 minOut = data.needScale ? (quotedMinOut * amountIn) / quotedAmountIn : quotedMinOut;\n require(netOut >= minOut, \"PendleSwap: 0x insufficient out\");\n\n _transferOut(tokenOut, msg.sender, netOut);\n }\n\n function _approveForExtRouter(address token, SwapData calldata data) internal {\n if (token == NATIVE) return;\n\n if (data.swapType == SwapType.OKX) {\n _safeApproveInfV2(IERC20(token), _okx_getTokenApprove());\n } else {\n _safeApproveInfV2(IERC20(token), data.extRouter);\n }\n }\n\n function _safeApproveInfV2(IERC20 token, address spender) internal {\n if (token.allowance(address(this), spender) < type(uint256).max) {\n token.forceApprove(spender, type(uint256).max);\n }\n }\n\n function _getScaledInputData(SwapType swapType, bytes calldata rawCallData, uint256 amountIn)\n internal\n view\n returns (bytes memory scaledCallData)\n {\n if (swapType == SwapType.KYBERSWAP) {\n bool isSuccess;\n (isSuccess, scaledCallData) =\n IKyberScalingHelper(KYBER_SCALING_HELPER).getScaledInputData(rawCallData, amountIn);\n\n require(isSuccess, \"PendleSwap: Kyber scaling failed\");\n } else if (swapType == SwapType.PARASWAP) {\n scaledCallData = _paraswapScaling(rawCallData, amountIn);\n } else if (swapType == SwapType.OKX) {\n scaledCallData = _okxScaling(rawCallData, amountIn);\n } else {\n assert(false);\n }\n }\n\n receive() external payable {}\n\n function _authorizeUpgrade(address) internal virtual override onlyOwner {}\n}\n\ninterface IKyberScalingHelper {\n function getScaledInputData(bytes calldata inputData, uint256 newAmount)\n external\n view\n returns (bool isSuccess, bytes memory data);\n}\n", - "lib/pendle-core-v2/contracts/router/swap-aggregator/IPSwapAggregator.sol": "// SPDX-License-Identifier: GPL-3.0-or-later\npragma solidity ^0.8.0;\n\nstruct SwapData {\n SwapType swapType;\n address extRouter;\n bytes extCalldata;\n bool needScale;\n}\n\nstruct SwapDataExtra {\n address tokenIn;\n address tokenOut;\n uint256 minOut;\n SwapData swapData;\n}\n\nenum SwapType {\n NONE,\n KYBERSWAP,\n RESERVE_1,\n // ETH_WETH not used in Aggregator\n ETH_WETH,\n OKX,\n ONE_INCH,\n PARASWAP,\n ZEROX,\n RESERVE_3,\n RESERVE_4,\n RESERVE_5\n}\n\ninterface IPSwapAggregator {\n event SwapSingle(SwapType indexed swapType, address indexed tokenIn, uint256 amountIn);\n\n function swap(address tokenIn, uint256 amountIn, SwapData calldata swapData) external payable;\n}\n", - "lib/pendle-core-v2/contracts/core/libraries/BoringOwnableUpgradeableV2.sol": "// SPDX-License-Identifier: GPL-3.0-or-later\npragma solidity ^0.8.0;\n\nimport \"@openzeppelin/contracts-upgradeable/proxy/utils/Initializable.sol\";\n\ncontract BoringOwnableUpgradeableData {\n address public owner;\n address public pendingOwner;\n}\n\nabstract contract BoringOwnableUpgradeableV2 is BoringOwnableUpgradeableData, Initializable {\n event OwnershipTransferred(address indexed previousOwner, address indexed newOwner);\n\n function __BoringOwnableV2_init(address _owner) internal onlyInitializing {\n owner = _owner;\n }\n\n /// @notice Transfers ownership to `newOwner`. Either directly or claimable by the new pending owner.\n /// Can only be invoked by the current `owner`.\n /// @param newOwner Address of the new owner.\n /// @param direct True if `newOwner` should be set immediately. False if `newOwner` needs to use `claimOwnership`.\n /// @param renounce Allows the `newOwner` to be `address(0)` if `direct` and `renounce` is True. Has no effect\n /// otherwise.\n function transferOwnership(address newOwner, bool direct, bool renounce) public onlyOwner {\n if (direct) {\n // Checks\n require(newOwner != address(0) || renounce, \"Ownable: zero address\");\n\n // Effects\n emit OwnershipTransferred(owner, newOwner);\n owner = newOwner;\n pendingOwner = address(0);\n } else {\n // Effects\n pendingOwner = newOwner;\n }\n }\n\n /// @notice Needs to be called by `pendingOwner` to claim ownership.\n function claimOwnership() public {\n address _pendingOwner = pendingOwner;\n\n // Checks\n require(msg.sender == _pendingOwner, \"Ownable: caller != pending owner\");\n\n // Effects\n emit OwnershipTransferred(owner, _pendingOwner);\n owner = _pendingOwner;\n pendingOwner = address(0);\n }\n\n /// @notice Only allows the `owner` to execute the function.\n modifier onlyOwner() {\n require(msg.sender == owner, \"Ownable: caller is not the owner\");\n _;\n }\n\n uint256[48] private __gap;\n}\n" - }, - "settings": { - "remappings": [ - "@pendle/core-v2/=lib/pendle-core-v2/", - "@pendle/sy/=lib/pendle-sy/", - "forge-std/=lib/forge-std/", - "vendor/=vendor/", - "vendor-locked/=vendor-locked/", - "@chainlink/=node_modules/@chainlink/", - "@eth-optimism/=node_modules/@eth-optimism/", - "@layerzerolabs/=node_modules/@layerzerolabs/", - "@openzeppelin/=node_modules/@openzeppelin/", - "hardhat/=node_modules/hardhat/", - "solidity-bytes-utils/=node_modules/solidity-bytes-utils/" - ], - "optimizer": { - "enabled": true, - "runs": 1000000 - }, - "metadata": { - "useLiteralContent": false, - "bytecodeHash": "ipfs", - "appendCBOR": true - }, - "outputSelection": { - "*": { - "*": [ - "evm.bytecode", - "evm.deployedBytecode", - "devdoc", - "userdoc", - "metadata", - "abi" - ] - } - }, - "evmVersion": "cancun", - "viaIR": true - }, - "abi": [], - "contract_file": "lib/pendle-core-v2/contracts/router/swap-aggregator/PendleSwap.sol", - "provenance": { - "chain_id": 1, - "address": "0xd14feb6aaf8650bbfcc8abec299b249a80fe7c78", - "retrieved": "2026-10-04", - "source": "Etherscan v2 getsourcecode", - "scope": "Target source, swap interface and ownership base only; not a complete compilation bundle." - } -} diff --git a/tests/fixtures/pendle_upgrade/old.json b/tests/fixtures/pendle_upgrade/old.json deleted file mode 100644 index e5a80091..00000000 --- a/tests/fixtures/pendle_upgrade/old.json +++ /dev/null @@ -1,74 +0,0 @@ -{ - "schema": 2, - "contract_name": "PendleSwap", - "compiler_version": "v0.8.30+commit.73712a01", - "language": "Solidity", - "sources": { - "lib/pendle-core-v2/contracts/router/swap-aggregator/PendleSwap.sol": "// SPDX-License-Identifier: GPL-3.0-or-later\npragma solidity ^0.8.17;\n\nimport \"../../core/libraries/TokenHelper.sol\";\nimport \"./IPSwapAggregator.sol\";\nimport \"./OKXScaleHelper.sol\";\nimport \"./ParaswapScaleHelper.sol\";\nimport \"@openzeppelin/contracts/token/ERC20/utils/SafeERC20.sol\";\n\nimport \"../../core/libraries/BoringOwnableUpgradeableV2.sol\";\nimport \"@openzeppelin/contracts/proxy/utils/UUPSUpgradeable.sol\";\n\ncontract PendleSwap is\n IPSwapAggregator,\n TokenHelper,\n OKXScaleHelper,\n ParaswapScaleHelper,\n BoringOwnableUpgradeableV2,\n UUPSUpgradeable\n{\n using Address for address;\n using SafeERC20 for IERC20;\n\n address private constant KYBER_SCALING_HELPER = 0x2f577A41BeC1BE1152AeEA12e73b7391d15f655D;\n\n constructor(bool okx_allowUnsupportedChain) OKXScaleHelper(okx_allowUnsupportedChain) {\n _disableInitializers();\n }\n\n function initialize(address _owner) external initializer {\n __BoringOwnableV2_init(_owner);\n }\n\n function swap(address tokenIn, uint256 amountIn, SwapData calldata data) external payable {\n _approveForExtRouter(tokenIn, data);\n data.extRouter\n .functionCallWithValue(\n data.needScale ? _getScaledInputData(data.swapType, data.extCalldata, amountIn) : data.extCalldata,\n tokenIn == NATIVE ? amountIn : 0\n );\n\n emit SwapSingle(data.swapType, tokenIn, amountIn);\n }\n\n function _approveForExtRouter(address token, SwapData calldata data) internal {\n if (token == NATIVE) return;\n\n if (data.swapType == SwapType.OKX) {\n _safeApproveInfV2(IERC20(token), _okx_getTokenApprove());\n } else {\n _safeApproveInfV2(IERC20(token), data.extRouter);\n }\n }\n\n function _safeApproveInfV2(IERC20 token, address spender) internal {\n if (token.allowance(address(this), spender) < type(uint256).max) {\n token.forceApprove(spender, type(uint256).max);\n }\n }\n\n function _getScaledInputData(SwapType swapType, bytes calldata rawCallData, uint256 amountIn)\n internal\n view\n returns (bytes memory scaledCallData)\n {\n if (swapType == SwapType.KYBERSWAP) {\n bool isSuccess;\n (isSuccess, scaledCallData) =\n IKyberScalingHelper(KYBER_SCALING_HELPER).getScaledInputData(rawCallData, amountIn);\n\n require(isSuccess, \"PendleSwap: Kyber scaling failed\");\n } else if (swapType == SwapType.ODOS) {\n scaledCallData = _odosScaling(rawCallData, amountIn);\n } else if (swapType == SwapType.PARASWAP) {\n scaledCallData = _paraswapScaling(rawCallData, amountIn);\n } else if (swapType == SwapType.OKX) {\n scaledCallData = _okxScaling(rawCallData, amountIn);\n } else {\n assert(false);\n }\n }\n\n function _odosScaling(bytes calldata rawCallData, uint256 amountIn)\n internal\n pure\n returns (bytes memory scaledCallData)\n {\n bytes4 selector = bytes4(rawCallData[:4]);\n bytes calldata dataToDecode = rawCallData[4:];\n\n assert(selector == IOdosRouterV2.swap.selector);\n (\n IOdosRouterV2.swapTokenInfo memory tokenInfo,\n bytes memory pathDefinition,\n address executor,\n uint32 referralCode\n ) = abi.decode(dataToDecode, (IOdosRouterV2.swapTokenInfo, bytes, address, uint32));\n\n tokenInfo.outputQuote = (tokenInfo.outputQuote * amountIn) / tokenInfo.inputAmount;\n tokenInfo.outputMin = (tokenInfo.outputMin * amountIn) / tokenInfo.inputAmount;\n tokenInfo.inputAmount = amountIn;\n\n return abi.encodeWithSelector(selector, tokenInfo, pathDefinition, executor, referralCode);\n }\n\n receive() external payable {}\n\n function _authorizeUpgrade(address) internal virtual override onlyOwner {}\n}\n\ninterface IKyberScalingHelper {\n function getScaledInputData(bytes calldata inputData, uint256 newAmount)\n external\n view\n returns (bool isSuccess, bytes memory data);\n}\n\ninterface IOdosRouterV2 {\n struct swapTokenInfo {\n address inputToken;\n uint256 inputAmount;\n address inputReceiver;\n address outputToken;\n uint256 outputQuote;\n uint256 outputMin;\n address outputReceiver;\n }\n\n function swap(swapTokenInfo memory tokenInfo, bytes calldata pathDefinition, address executor, uint32 referralCode)\n external\n payable\n returns (uint256 amountOut);\n}\n", - "lib/pendle-core-v2/contracts/router/swap-aggregator/IPSwapAggregator.sol": "// SPDX-License-Identifier: GPL-3.0-or-later\npragma solidity ^0.8.0;\n\nstruct SwapData {\n SwapType swapType;\n address extRouter;\n bytes extCalldata;\n bool needScale;\n}\n\nstruct SwapDataExtra {\n address tokenIn;\n address tokenOut;\n uint256 minOut;\n SwapData swapData;\n}\n\nenum SwapType {\n NONE,\n KYBERSWAP,\n ODOS,\n // ETH_WETH not used in Aggregator\n ETH_WETH,\n OKX,\n ONE_INCH,\n PARASWAP,\n RESERVE_2,\n RESERVE_3,\n RESERVE_4,\n RESERVE_5\n}\n\ninterface IPSwapAggregator {\n event SwapSingle(SwapType indexed swapType, address indexed tokenIn, uint256 amountIn);\n\n function swap(address tokenIn, uint256 amountIn, SwapData calldata swapData) external payable;\n}\n", - "lib/pendle-core-v2/contracts/core/libraries/BoringOwnableUpgradeableV2.sol": "// SPDX-License-Identifier: GPL-3.0-or-later\npragma solidity ^0.8.0;\n\nimport \"@openzeppelin/contracts-upgradeable/proxy/utils/Initializable.sol\";\n\ncontract BoringOwnableUpgradeableData {\n address public owner;\n address public pendingOwner;\n}\n\nabstract contract BoringOwnableUpgradeableV2 is BoringOwnableUpgradeableData, Initializable {\n event OwnershipTransferred(address indexed previousOwner, address indexed newOwner);\n\n function __BoringOwnableV2_init(address _owner) internal onlyInitializing {\n owner = _owner;\n }\n\n /// @notice Transfers ownership to `newOwner`. Either directly or claimable by the new pending owner.\n /// Can only be invoked by the current `owner`.\n /// @param newOwner Address of the new owner.\n /// @param direct True if `newOwner` should be set immediately. False if `newOwner` needs to use `claimOwnership`.\n /// @param renounce Allows the `newOwner` to be `address(0)` if `direct` and `renounce` is True. Has no effect\n /// otherwise.\n function transferOwnership(address newOwner, bool direct, bool renounce) public onlyOwner {\n if (direct) {\n // Checks\n require(newOwner != address(0) || renounce, \"Ownable: zero address\");\n\n // Effects\n emit OwnershipTransferred(owner, newOwner);\n owner = newOwner;\n pendingOwner = address(0);\n } else {\n // Effects\n pendingOwner = newOwner;\n }\n }\n\n /// @notice Needs to be called by `pendingOwner` to claim ownership.\n function claimOwnership() public {\n address _pendingOwner = pendingOwner;\n\n // Checks\n require(msg.sender == _pendingOwner, \"Ownable: caller != pending owner\");\n\n // Effects\n emit OwnershipTransferred(owner, _pendingOwner);\n owner = _pendingOwner;\n pendingOwner = address(0);\n }\n\n /// @notice Only allows the `owner` to execute the function.\n modifier onlyOwner() {\n require(msg.sender == owner, \"Ownable: caller is not the owner\");\n _;\n }\n\n uint256[48] private __gap;\n}\n" - }, - "settings": { - "remappings": [ - "lib/pendle-core-v3/:@openzeppelin/contracts/=node_modules/@openzeppelin-v5/contracts/", - "lib/pendle-core-v3/:@openzeppelin/contracts-upgradeable/=node_modules/@openzeppelin-v5/contracts-upgradeable/", - "node_modules/@layerzerolabs/test-devtools-evm-foundry/:forge-std/=node_modules/forge-std/src/", - "node_modules/@layerzerolabs/oapp-evm/:@openzeppelin/=node_modules/@openzeppelin/", - "node_modules/@layerzerolabs/oft-evm/:@openzeppelin/=node_modules/@openzeppelin/", - "node_modules/@layerzerolabs/:@openzeppelin/=node_modules/@openzeppelin-v5/", - "@axelar-network/=node_modules/@axelar-network/", - "@chainlink/=node_modules/@chainlink/", - "@openzeppelin/contracts/=node_modules/@openzeppelin/contracts/", - "@openzeppelin/contracts-upgradeable/=node_modules/@openzeppelin/contracts-upgradeable/", - "@openzeppelin-v5/contracts/=node_modules/@openzeppelin-v5/contracts/", - "@openzeppelin-v5/contracts-upgradeable/=node_modules/@openzeppelin-v5/contracts-upgradeable/", - "@eth-optimism/=node_modules/@eth-optimism/", - "@prb/test/=node_modules/@prb/test/", - "forge-std/=node_modules/forge-std/", - "@pendle/core-v2/=lib/pendle-core-v2/", - "@pendle/core-v3/=lib/pendle-core-v3/", - "@pendle/sy/=lib/pendle-sy/", - "pendle-sy/=lib/pendle-sy/contracts/", - "hardhat-deploy/=node_modules/hardhat-deploy/", - "ds-test/=lib/surl/lib/forge-std/lib/ds-test/src/", - "solidity-bytes-utils/=node_modules/solidity-bytes-utils/", - "solidity-stringutils/=lib/surl/lib/solidity-stringutils/src/", - "surl/=lib/surl/src/", - "@layerzerolabs/=node_modules/@layerzerolabs/", - "pendle-core-v2/=lib/pendle-core-v2/contracts/", - "pendle-core-v3/=lib/pendle-core-v3/contracts/" - ], - "optimizer": { - "enabled": true, - "runs": 1000000 - }, - "metadata": { - "useLiteralContent": false, - "bytecodeHash": "ipfs", - "appendCBOR": true - }, - "outputSelection": { - "*": { - "*": [ - "evm.bytecode", - "evm.deployedBytecode", - "devdoc", - "userdoc", - "metadata", - "abi" - ] - } - }, - "evmVersion": "shanghai", - "viaIR": true - }, - "abi": [], - "contract_file": "lib/pendle-core-v2/contracts/router/swap-aggregator/PendleSwap.sol", - "provenance": { - "chain_id": 1, - "address": "0xbc17404b7bb500051c75c83e4aa5ae447d967811", - "retrieved": "2026-10-04", - "source": "Etherscan v2 getsourcecode", - "scope": "Target source, swap interface and ownership base only; not a complete compilation bundle." - } -} diff --git a/tests/test_pendle_context.py b/tests/test_pendle_context.py index 0defa2fc..661559b0 100644 --- a/tests/test_pendle_context.py +++ b/tests/test_pendle_context.py @@ -1,22 +1,17 @@ -"""PendleSwap upgrade context regressions using the 2026-10-04 verified sources.""" +"""PendleSwap context behavior using small, synthetic source bundles.""" -import json from collections.abc import Iterator from dataclasses import replace -from pathlib import Path from unittest.mock import MagicMock, patch import pytest from utils.calldata.decoder import DecodedCall from utils.llm import pendle_context -from utils.llm.ai_explainer import _build_prompt from utils.llm.pendle_context import ( PENDLE_SWAP, - _read_owner, _source_evidence, format_pendle_prompt, - format_pendle_report, resolve_pendle_context, ) from utils.llm.protocol_context import resolve_protocol_context @@ -27,13 +22,47 @@ OWNER = "0x8119EC16F0573B7dAc7C0CB94EB504FB32456ee1" type Boundaries = tuple[MagicMock, MagicMock, MagicMock] +_OLD_SOURCE = """ +contract PendleSwap { + function swap() external { _getScaledInputData(SwapType.ODOS); } + function _getScaledInputData(SwapType swapType) internal { + if (swapType == SwapType.ODOS) { _odosScaling(); } else { assert(false); } + } + function _authorizeUpgrade(address) internal onlyOwner {} +} +""" +_NEW_SOURCE = """ +contract PendleSwap { + function swap() external { _zeroExSwap(); } + function _zeroExSwap() internal { _transferOut(tokenOut, msg.sender, netOut); } + function _getScaledInputData(SwapType swapType) internal { assert(false); } + function _authorizeUpgrade(address) internal onlyOwner {} +} +""" +_OWNERSHIP_SOURCE = """ +abstract contract BoringOwnableUpgradeableV2 { + modifier onlyOwner() { require(msg.sender == owner, "not owner"); _; } +} +""" + def _record(name: str) -> VerifiedContract: - """Load the scoped verified bundle fixture.""" - data = json.loads((Path(__file__).parent / "fixtures" / "pendle_upgrade" / f"{name}.json").read_text()) - record = VerifiedContract.from_cache_dict(data) - assert record is not None - return record + """Build only the source declarations the adapter consumes, without cache metadata.""" + swap_types = "NONE, KYBERSWAP, RESERVE_1, ZEROX" if name == "new" else "NONE, KYBERSWAP, ODOS, RESERVE_2" + return VerifiedContract( + contract_name="PendleSwap", + compiler_version="", + language="Solidity", + contract_file="PendleSwap.sol", + sources={ + "PendleSwap.sol": _NEW_SOURCE if name == "new" else _OLD_SOURCE, + "IPSwapAggregator.sol": ( + "struct SwapData { SwapType swapType; address extRouter; bytes extCalldata; bool needScale; }\n" + f"enum SwapType {{ {swap_types} }}\ninterface IPSwapAggregator {{}}" + ), + "BoringOwnableUpgradeableV2.sol": _OWNERSHIP_SOURCE, + }, + ) def _upgrade(migrate: bool = False) -> DecodedCall: @@ -61,18 +90,12 @@ def boundaries() -> Iterator[Boundaries]: @pytest.mark.parametrize("chain_id", [1, 42161]) def test_upgrade_includes_scope_controls_and_route_semantics(boundaries: Boundaries, chain_id: int) -> None: """Unchanged ABI must not hide enum changes, unsupported scaling or existing authorization.""" - contexts = resolve_pendle_context("PENDLE", chain_id, [(PENDLE_SWAP.lower(), _upgrade())]) - assert len(contexts) == 1 - prompt = format_pendle_prompt(contexts) - assert "optional aggregator leg" in prompt - assert "NONE and ETH_WETH branches bypass" in prompt - assert "calling router for the next step" in prompt + resolved = resolve_protocol_context("PENDLE", chain_id, [(PENDLE_SWAP.lower(), _upgrade())]) + prompt = resolved.prompt assert "not a live trace" in prompt - assert "not proof every caller uses a nonzero minimum" in prompt assert f"Current proxy owner() (live read): {OWNER}" in prompt - assert "_authorizeUpgrade(address) internal virtual override onlyOwner" in prompt - assert 'require(msg.sender == owner, "Ownable: caller is not the owner")' in prompt - assert "type(uint256).max" in prompt + assert "_authorizeUpgrade(address) internal onlyOwner" in prompt + assert 'require(msg.sender == owner, "not owner")' in prompt before, after = prompt.split("Proposed implementation evidence:") assert "SwapType.ODOS" in before assert "RESERVE_2" in before @@ -80,44 +103,14 @@ def test_upgrade_includes_scope_controls_and_route_semantics(boundaries: Boundar assert "ZEROX" in after assert "SwapType.ODOS" not in after assert "assert(false)" in after - assert "(quotedMinOut * amountIn) / quotedAmountIn" in after assert "_transferOut(tokenOut, msg.sender, netOut)" in after - assert "No fixed risk rating" in prompt - assert "removes ODOS calldata scaling" in prompt - assert "does not by itself establish a higher likelihood of storage collisions" in prompt - - -def test_registry_publishes_context_and_full_address_links(boundaries: Boundaries) -> None: - """The adapter's facts and additional owner/implementation addresses reach the report.""" - resolved = resolve_protocol_context("pendle", 1, [(PENDLE_SWAP, _upgrade())]) - assert "optional aggregator leg" in resolved.prompt - assert "optional aggregator leg" in resolved.report + explorer = "etherscan.io" if chain_id == 1 else "arbiscan.io" for address in (PENDLE_SWAP, OLD, NEW, OWNER): assert address in resolved.addresses - assert f"https://etherscan.io/address/{address}" in resolved.report + assert f"https://{explorer}/address/{address}" in resolved.report assert resolved.labels[PENDLE_SWAP] == "PendleSwap" - - -def test_prompt_preserves_integration_reference_and_live_observation_distinction(boundaries: Boundaries) -> None: - """The explainer must not relabel documented router flow as a live execution trace.""" - contexts = resolve_pendle_context("pendle", 1, [(PENDLE_SWAP, _upgrade())]) - prompt = _build_prompt( - target=PENDLE_SWAP, - value=0, - decoded_calls=[_upgrade()], - simulation=None, - protocol_context=format_pendle_prompt(contexts), - ) - assert "Distinguish documented integration architecture from live observations" in prompt - assert "not a live trace" in prompt - - -def test_arbitrum_report_uses_arbitrum_links(boundaries: Boundaries) -> None: - """Same deployment addresses must link to the actual alert chain.""" - contexts = resolve_pendle_context("pendle", 42161, [(PENDLE_SWAP, _upgrade())]) - report = format_pendle_report(contexts, 42161, {}) - assert f"https://arbiscan.io/address/{PENDLE_SWAP}" in report - assert "etherscan.io/address" not in report + contract_call = boundaries[2].return_value.eth.contract.call_args + assert contract_call.kwargs["address"] == PENDLE_SWAP @pytest.mark.parametrize( @@ -181,15 +174,6 @@ def test_owner_read_failure_preserves_source_evidence(boundaries: Boundaries) -> assert contexts[0].owner is None -def test_owner_is_read_from_proxy_not_implementation() -> None: - """Ownership state lives at the proxy address.""" - client = MagicMock() - client.eth.contract.return_value.functions.owner.return_value.call.return_value = OWNER - with patch.object(pendle_context.ChainManager, "get_client", return_value=client): - assert _read_owner(1, PENDLE_SWAP) == OWNER - assert client.eth.contract.call_args.kwargs["address"] == PENDLE_SWAP - - def test_missing_source_and_implementation_are_explicit(boundaries: Boundaries) -> None: """Missing evidence never turns into a guessed old implementation or safe-storage claim.""" boundaries[0].return_value = None @@ -199,7 +183,6 @@ def test_missing_source_and_implementation_are_explicit(boundaries: Boundaries) prompt = format_pendle_prompt(contexts) assert "Current implementation: unavailable" in prompt assert prompt.count("target source unavailable") == 2 - assert "Storage UNKNOWN is a validation gap, not a proven collision" in prompt def test_failed_batch_member_does_not_hide_next_upgrade(boundaries: Boundaries) -> None: @@ -210,26 +193,22 @@ def test_failed_batch_member_does_not_hide_next_upgrade(boundaries: Boundaries) assert contexts[0].is_upgrade_and_call -def test_source_excerpts_are_scoped_to_the_deployed_contract() -> None: - """A same-name member in another bundled contract must not become PendleSwap evidence.""" +@pytest.mark.parametrize("flattened", [False, True]) +def test_source_excerpts_exclude_unrelated_contracts(flattened: bool) -> None: + """Only the deployed target and relevant declarations belong in context, in either source format.""" record = _record("new") - assert record.contract_file is not None sources = dict(record.sources) - sources[record.contract_file] += '\ncontract Decoy { function swap() external { revert("DECOY"); } }' - assert "DECOY" not in _source_evidence(replace(record, sources=sources)) - - -def test_flattened_source_does_not_include_unrelated_contracts() -> None: - """Imported swap declarations must not cause an entire flattened bundle to enter the prompt.""" - record = _record("new") - flattened = "\n".join(record.sources.values()) - flattened += '\ncontract Decoy { function swap() external { revert("DECOY"); } }' - evidence = _source_evidence(replace(record, sources={"flat.sol": flattened}, contract_file="flat.sol")) + sources["PendleSwap.sol"] += '\ncontract Decoy { function swap() external { revert("DECOY"); } }' + if flattened: + record = replace(record, sources={"flat.sol": "\n".join(sources.values())}, contract_file="flat.sol") + else: + record = replace(record, sources=sources) + evidence = _source_evidence(record) assert "DECOY" not in evidence assert "enum SwapType" in evidence assert "struct SwapData" in evidence assert "_zeroExSwap" in evidence - assert 'require(msg.sender == owner, "Ownable: caller is not the owner")' in evidence + assert 'require(msg.sender == owner, "not owner")' in evidence def test_other_replacement_and_unresolved_target_are_not_assumed_to_be_pendle() -> None: