diff --git a/tests/conftest.py b/tests/conftest.py index 82c7b70b..66eb5125 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -19,7 +19,7 @@ def _isolate_from_live_apis(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: """Block accidental live API/RPC calls and reset cross-test singletons. - Strips `ETHERSCAN_TOKEN`, every `PROVIDER_URL_*`, every `TELEGRAM_*` + Strips `ETHERSCAN_TOKEN`, `TENDERLY_API_KEY`, every `PROVIDER_URL_*`, every `TELEGRAM_*` credential, and emergency webhook credentials so a missing mock short-circuits cheaply via the "no token / no provider / no credentials" code paths that already exist for production use. Also clears `DEFAULT_PROVIDER_URLS` so a public HyperEVM @@ -36,7 +36,7 @@ def _isolate_from_live_apis(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> repo and never leak entries between tests. """ for key in list(os.environ): - if key in {"ETHERSCAN_TOKEN", "LIQUIDITY_WEBHOOK_SECRET"} or key.startswith( + if key in {"ETHERSCAN_TOKEN", "LIQUIDITY_WEBHOOK_SECRET", "TENDERLY_API_KEY"} or key.startswith( ("PROVIDER_URL_", "TELEGRAM_", "LIQUIDITY_WEBHOOK_") ): monkeypatch.delenv(key, raising=False) diff --git a/tests/test_ai_explainer.py b/tests/test_ai_explainer.py index 2b410c18..0826cfdc 100644 --- a/tests/test_ai_explainer.py +++ b/tests/test_ai_explainer.py @@ -1752,6 +1752,114 @@ def test_bytes_params_reach_the_prompt_as_hex_not_python_repr( self.assertNotIn("b'", prompt) +class TestBatchSequentialSimulation(unittest.TestCase): + """Batch calls are simulated in order on shared state, as the timelock executes them.""" + + CALLS = [ + {"target": "0xT1", "data": PAUSE_DATA, "value": "0"}, + {"target": "0xT2", "data": PAUSE_DATA, "value": "0"}, + {"target": "0xT3", "data": PAUSE_DATA, "value": "0"}, + ] + + def _provider(self) -> MagicMock: + provider = MagicMock() + provider.supports_structured_output = False + provider.complete.return_value = "TLDR: three calls. LOW.\n\nDETAIL:\nanalysis." + provider.model_name = "test" + return provider + + @patch("utils.llm.ai_explainer.get_source_context", return_value=None) + @patch("utils.llm.ai_explainer.get_contract_label", return_value="") + @patch("utils.llm.ai_explainer.get_llm_provider") + @patch("utils.llm.ai_explainer.simulate_transaction") + @patch("utils.llm.ai_explainer.simulate_bundle") + @patch("utils.llm.ai_explainer.decode_calldata", return_value=PAUSE) + def test_bundle_results_are_labeled_batch_order( + self, + _mock_decode: MagicMock, + mock_bundle: MagicMock, + mock_simulate: MagicMock, + mock_get_provider: MagicMock, + _mock_label: MagicMock, + _mock_source: MagicMock, + ) -> None: + mock_bundle.return_value = [ + SimulationResult(success=True, gas_used=111), + SimulationResult(success=True, gas_used=222), + SimulationResult(success=True, gas_used=333), + ] + provider = self._provider() + mock_get_provider.return_value = provider + + result = explain_batch_transaction(calls=self.CALLS, chain_id=1, from_address="0xTimelock", refine=False) + + assert result is not None + bundle_calls = mock_bundle.call_args.args[0] + self.assertEqual([call.target for call in bundle_calls], ["0xT1", "0xT2", "0xT3"]) + self.assertEqual(mock_bundle.call_args.kwargs["from_address"], "0xTimelock") + mock_simulate.assert_not_called() + prompt = provider.complete.call_args[0][0] + self.assertIn("Call 1 (simulated in batch order, first call):", prompt) + self.assertIn("Call 3 (simulated in batch order, after calls 1-2):", prompt) + self.assertNotIn("independent simulation", prompt) + self.assertIn("**Batch simulation:** SUCCESS, gas 333", result.report) + + @patch("utils.llm.ai_explainer.get_source_context", return_value=None) + @patch("utils.llm.ai_explainer.get_contract_label", return_value="") + @patch("utils.llm.ai_explainer.get_llm_provider") + @patch("utils.llm.ai_explainer.simulate_transaction") + @patch("utils.llm.ai_explainer.simulate_bundle") + @patch("utils.llm.ai_explainer.decode_calldata", return_value=PAUSE) + def test_calls_after_a_bundle_revert_fall_back_to_independent( + self, + _mock_decode: MagicMock, + mock_bundle: MagicMock, + mock_simulate: MagicMock, + mock_get_provider: MagicMock, + _mock_label: MagicMock, + _mock_source: MagicMock, + ) -> None: + mock_bundle.return_value = [ + SimulationResult(success=True, gas_used=111), + SimulationResult(success=False, error_message="execution reverted: not authorized"), + None, + ] + mock_simulate.return_value = SimulationResult(success=True, gas_used=333) + provider = self._provider() + mock_get_provider.return_value = provider + + result = explain_batch_transaction(calls=self.CALLS, chain_id=1, refine=False) + + assert result is not None + self.assertEqual(mock_simulate.call_count, 1) + self.assertEqual(mock_simulate.call_args.kwargs["target"], "0xT3") + prompt = provider.complete.call_args[0][0] + self.assertIn("Call 1 (simulated in batch order, first call):", prompt) + self.assertIn("Call 3 (independent simulation", prompt) + self.assertNotIn("not authorized", prompt) + self.assertIn("**Batch simulation diagnostic:** execution reverted: not authorized", result.report) + + @patch("utils.llm.ai_explainer.get_source_context", return_value=None) + @patch("utils.llm.ai_explainer.get_contract_label", return_value="") + @patch("utils.llm.ai_explainer.get_llm_provider") + @patch("utils.llm.ai_explainer.simulate_transaction") + @patch("utils.llm.ai_explainer.simulate_bundle") + @patch("utils.llm.ai_explainer.decode_calldata", return_value=PAUSE) + def test_skip_simulation_skips_the_bundle( + self, + _mock_decode: MagicMock, + mock_bundle: MagicMock, + mock_simulate: MagicMock, + mock_get_provider: MagicMock, + _mock_label: MagicMock, + _mock_source: MagicMock, + ) -> None: + mock_get_provider.return_value = self._provider() + explain_batch_transaction(calls=self.CALLS, chain_id=1, skip_simulation=True, refine=False) + mock_bundle.assert_not_called() + mock_simulate.assert_not_called() + + class TestBatchSimulationsAttributed(unittest.TestCase): @patch("utils.llm.ai_explainer.get_source_context", return_value=None) @patch("utils.llm.ai_explainer.get_contract_label", return_value="") @@ -2056,7 +2164,7 @@ def test_successful_simulations_are_capped( assert result is not None prompt = provider.complete.call_args[0][0] self.assertEqual(prompt.count("(independent simulation;"), MAX_PROMPT_SIMULATIONS) - self.assertIn("2 further successful independent simulations omitted", prompt) + self.assertIn("2 further successful simulations omitted", prompt) @patch("utils.llm.ai_explainer.get_source_context", return_value=None) @patch("utils.llm.ai_explainer.get_contract_label", return_value="") diff --git a/tests/test_infinifi_outland_context.py b/tests/test_infinifi_outland_context.py new file mode 100644 index 00000000..cda5608d --- /dev/null +++ b/tests/test_infinifi_outland_context.py @@ -0,0 +1,225 @@ +"""Tests for Infinifi Outland LLM governance context.""" + +import unittest +from unittest.mock import MagicMock, patch + +from utils.calldata.decoder import DecodedCall +from utils.erc20_metadata import ERC20Metadata +from utils.llm import abi_exposure, infinifi_outland_context +from utils.llm.infinifi_outland_context import ( + ZERO_ADDRESS, + ConnectorRouteContext, + FarmTypeContext, + HubVaultContext, + OracleAssignmentContext, + RouteConfig, + format_outland_prompt, + format_outland_report, + resolve_outland_context, +) + +ACCOUNTING = "0x7A5C5dbA4fbD0e1e1A2eCDBe752fAe55f6E842B3" +HUB = "0x13025F34C1ec2A16bF68f3a3c4e986a3E85CED61" +FARM = "0xA7c1DAEAA5D97e1319B4Ff6Cdf658F5C4582A27E" +REGISTRY = "0xF5f2718708f471e43968271956CC01aaA8c46119" +CONNECTOR = "0x3373784A7a52A07F9339aA8F60403420cC602c52" +VAULT = "0x77776F422B7EB0A95ccD35fBd088A5957D4408eA" +BASE_VAULT = "0xf0d0F1fdEE5595628De17B37E4134a5bAc4441C3" +ORACLE = "0x168DF792845BA1bd80d485399de63a4110b03242" +USDC = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48" +PEER = "0x1111111111111111111111111111111111111111" +NEW_PEER = "0x2222222222222222222222222222222222222222" + + +def _call(name: str, *params: tuple[str, object]) -> DecodedCall: + return DecodedCall(function_name=name, signature=f"{name}()", params=list(params)) + + +def _client(batch_results: list[list[object]], single_call: object = None) -> MagicMock: + """A ChainManager client whose batches return ``batch_results`` in order.""" + client = MagicMock() + client.execute_batch.side_effect = batch_results + client.get_contract.return_value.functions.getVault.return_value.call.return_value = single_call + client.get_contract.return_value.functions.price.return_value.call.return_value = single_call + return client + + +class TestGuards(unittest.TestCase): + def test_other_protocol_resolves_nothing(self) -> None: + call = _call("addFarms", ("uint256", 2), ("address[]", (FARM,))) + self.assertEqual(resolve_outland_context("3jane", 1, [(REGISTRY, call)]), []) + + def test_other_chain_resolves_nothing(self) -> None: + call = _call("addFarms", ("uint256", 2), ("address[]", (FARM,))) + self.assertEqual(resolve_outland_context("infinifi", 8453, [(REGISTRY, call)]), []) + + def test_resolution_failure_does_not_raise(self) -> None: + call = _call("setOracle", ("address", VAULT), ("address", ORACLE)) + with patch.object(infinifi_outland_context, "exposes", side_effect=RuntimeError("etherscan down")): + self.assertEqual(resolve_outland_context("infinifi", 1, [(ACCOUNTING, call)]), []) + + +class TestFarmType(unittest.TestCase): + def test_add_farms_type_is_named_without_rpc(self) -> None: + call = _call("addFarms", ("uint256", 2), ("address[]", (FARM.lower(),))) + with patch.object(infinifi_outland_context, "exposes") as probe: + contexts = resolve_outland_context("infinifi", 1, [(REGISTRY, call)]) + probe.assert_not_called() + self.assertEqual(contexts, [FarmTypeContext(REGISTRY, "addFarms", 2, (FARM,))]) + prompt = format_outland_prompt(contexts) + self.assertIn("farm type 2 = FarmTypes.MATURITY", prompt) + self.assertIn("principal is locked until the farm's maturity", prompt) + + def test_unknown_farm_type_is_flagged(self) -> None: + context = FarmTypeContext(REGISTRY, "addFarms", 7, (FARM,)) + self.assertIn("FarmTypes.UNKNOWN — not a FarmTypes constant", format_outland_prompt([context])) + + +class TestOracle(unittest.TestCase): + def test_price_is_scaled_by_asset_decimals(self) -> None: + call = _call("setOracle", ("address", VAULT), ("address", ORACLE)) + with ( + patch.object(infinifi_outland_context, "exposes", return_value=True), + patch.object(infinifi_outland_context, "fetch_erc20_metadata", return_value=ERC20Metadata("OV-143", 18)), + patch.object(infinifi_outland_context.ChainManager, "get_client", return_value=_client([], 10**18)), + ): + contexts = resolve_outland_context("infinifi", 1, [(ACCOUNTING, call)]) + self.assertEqual(contexts, [OracleAssignmentContext(ACCOUNTING, VAULT, "OV-143", 18, ORACLE, 10**18)]) + self.assertIn("one whole OV-143 is valued at 1 reference units", format_outland_prompt(contexts)) + + def test_usdc_scale_reads_as_parity(self) -> None: + # IOracle convention: a 6-decimal stable at parity is quoted at 1e30. + context = OracleAssignmentContext(ACCOUNTING, USDC, "USDC", 6, ORACLE, 10**30) + self.assertIn("= `1` reference units", format_outland_report([context], 1, {})) + + def test_removing_an_oracle_is_not_priced(self) -> None: + call = _call("setOracle", ("address", VAULT), ("address", ZERO_ADDRESS)) + with patch.object(infinifi_outland_context, "exposes") as probe: + self.assertEqual(resolve_outland_context("infinifi", 1, [(ACCOUNTING, call)]), []) + probe.assert_not_called() + + +class TestHubVault(unittest.TestCase): + def _resolve(self, registered: list[int], existing: str | None = None) -> list: + call = _call("setVault", ("address", VAULT)) + with ( + patch.object(infinifi_outland_context, "exposes", return_value=True), + patch.object( + infinifi_outland_context.ChainManager, "get_client", return_value=_client([[143, registered]], existing) + ), + ): + return resolve_outland_context("infinifi", 1, [(HUB, call)]) + + def test_new_chain_replaces_nothing(self) -> None: + contexts = self._resolve([8453]) + self.assertEqual(contexts, [HubVaultContext(HUB, VAULT, 143, (8453,), None)]) + prompt = format_outland_prompt(contexts) + self.assertIn("adds chain 143; no existing vault is replaced", prompt) + self.assertIn("Chains registered before this call: 8453", prompt) + + def test_existing_chain_names_the_replaced_vault(self) -> None: + contexts = self._resolve([8453, 143], existing=BASE_VAULT) + self.assertEqual(contexts[0].replaced_vault, BASE_VAULT) + self.assertIn(f"REPLACES the existing chain-143 vault {BASE_VAULT}", format_outland_prompt(contexts)) + + def test_farm_set_vault_is_not_treated_as_hub(self) -> None: + # OutlandFarm.setVault shares the selector but has no per-chain registry. + call = _call("setVault", ("address", VAULT)) + with patch.object(infinifi_outland_context, "exposes", return_value=False): + self.assertEqual(resolve_outland_context("infinifi", 1, [(FARM, call)]), []) + + +class TestConnectorRoute(unittest.TestCase): + def _resolve(self, calls: list[DecodedCall], config: tuple[str, int, int]) -> list: + with ( + patch.object(infinifi_outland_context, "exposes", return_value=True), + patch.object(infinifi_outland_context.ChainManager, "get_client", return_value=_client([[config]])), + ): + return resolve_outland_context("infinifi", 1, [(CONNECTOR, call) for call in calls]) + + def test_unconfigured_route_is_not_live(self) -> None: + calls = [ + _call("enableChainAsset", ("uint256", 143), ("address", USDC)), + _call("setCctpDomain", ("uint256", 143), ("uint32", 15)), + ] + contexts = self._resolve(calls, (ZERO_ADDRESS, 0, 0)) + # Both calls name chain 143, so it is read once. + self.assertEqual(contexts, [ConnectorRouteContext(CONNECTOR, 143, RouteConfig(ZERO_ADDRESS, 0, 0))]) + prompt = format_outland_prompt(contexts) + self.assertIn("NOT configured", prompt) + self.assertIn("route is not live after this batch", prompt) + self.assertIn("**not configured**", format_outland_report(contexts, 1, {})) + + def test_configured_route_names_the_peer(self) -> None: + contexts = self._resolve([_call("setCctpDomain", ("uint256", 143), ("uint32", 15))], (PEER, 200_000, 99)) + self.assertIn( + f"configured, unchanged by this batch — peer {PEER}, gas limit 200,000", format_outland_prompt(contexts) + ) + self.assertIn(PEER, contexts[0].addresses) + + @staticmethod + def _set_configuration(peer: str, selector: int, gas_limit: int) -> DecodedCall: + return _call( + "setConfiguration", + ("uint256", 143), + ("address", peer), + ("uint256", selector), + ("uint256", gas_limit), + ) + + def test_new_route_in_the_same_batch_shows_the_values_it_sets(self) -> None: + calls = [_call("setCctpDomain", ("uint256", 143), ("uint32", 15)), self._set_configuration(PEER, 99, 200_000)] + contexts = self._resolve(calls, (ZERO_ADDRESS, 0, 0)) + self.assertEqual(contexts[0].after_batch, RouteConfig(PEER, 200_000, 99)) + prompt = format_outland_prompt(contexts) + self.assertIn(f"this batch sets it via setConfiguration to peer {PEER}, gas limit 200,000", prompt) + self.assertIn("Before the batch: not configured", prompt) + + def test_reconfiguring_a_live_route_reports_the_new_peer_not_the_old(self) -> None: + contexts = self._resolve([self._set_configuration(NEW_PEER, 99, 300_000)], (PEER, 200_000, 99)) + prompt = format_outland_prompt(contexts) + self.assertIn(f"to peer {NEW_PEER}, gas limit 300,000", prompt) + self.assertIn(f"Before the batch: peer {PEER}", prompt) + self.assertNotIn("unchanged", prompt) + report = format_outland_report(contexts, 1, {}) + self.assertIn(f"set by this batch — peer [`{NEW_PEER}`]", report) + self.assertIn(f"(was peer [`{PEER}`]", report) + self.assertEqual(contexts[0].addresses, [CONNECTOR, PEER, NEW_PEER]) + + def test_clearing_a_live_route_is_flagged(self) -> None: + contexts = self._resolve([self._set_configuration(ZERO_ADDRESS, 0, 0)], (PEER, 200_000, 99)) + self.assertIn("this batch CLEARS the route", format_outland_prompt(contexts)) + self.assertIn("**cleared** by this batch", format_outland_report(contexts, 1, {})) + + def test_last_configuration_for_a_chain_wins(self) -> None: + calls = [self._set_configuration(PEER, 99, 200_000), self._set_configuration(NEW_PEER, 99, 300_000)] + contexts = self._resolve(calls, (ZERO_ADDRESS, 0, 0)) + self.assertEqual(contexts[0].proposed, RouteConfig(NEW_PEER, 300_000, 99)) + + +class TestAbiExposure(unittest.TestCase): + def setUp(self) -> None: + abi_exposure.reset_cache() + + def test_proxy_falls_back_to_implementation(self) -> None: + def abi_for(_chain_id: int, address: str) -> list[dict]: + name = "getVaultChainIds" if address == "0ximpl" else "upgradeTo" + return [{"type": "function", "name": name}] + + with ( + patch.object(abi_exposure, "fetch_abi_entries", side_effect=abi_for), + patch("utils.proxy.get_current_implementation", return_value="0ximpl"), + ): + self.assertTrue(abi_exposure.exposes(1, HUB, {"getVaultChainIds"})) + + def test_own_abi_skips_the_proxy_lookup(self) -> None: + with ( + patch.object(abi_exposure, "fetch_abi_entries", return_value=[{"type": "function", "name": "price"}]), + patch("utils.proxy.get_current_implementation") as lookup, + ): + self.assertTrue(abi_exposure.exposes(1, ORACLE, {"price"})) + lookup.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_on_chain_state.py b/tests/test_on_chain_state.py index bdf10b7c..a36ebbe2 100644 --- a/tests/test_on_chain_state.py +++ b/tests/test_on_chain_state.py @@ -75,6 +75,17 @@ def test_mapping_with_bytes32_key(self) -> None: result = _parse_var_declaration(snippet, "values") self.assertEqual(result, ("uint256", ["bytes32"])) + def test_named_mapping_parameters(self) -> None: + # ConnectorCCTP_Chainlink: the bool value was misreported as the setter's uint32. + snippet = "/// @notice Sentinel\nmapping(uint256 chainId => bool) public cctpDomainConfigured;" + self.assertEqual(_parse_var_declaration(snippet, "cctpDomainConfigured"), ("bool", ["uint256"])) + snippet = "mapping(uint256 chainId => uint32 domain) public cctpDomains;" + self.assertEqual(_parse_var_declaration(snippet, "cctpDomains"), ("uint32", ["uint256"])) + + def test_named_nested_mapping_skipped(self) -> None: + snippet = "mapping(address owner => mapping(address spender => uint256)) public allowances;" + self.assertIsNone(_parse_var_declaration(snippet, "allowances")) + def test_nested_mapping_skipped(self) -> None: snippet = "mapping(bytes32 => mapping(address => uint256)) public nested;" result = _parse_var_declaration(snippet, "nested") diff --git a/tests/test_protocol_context.py b/tests/test_protocol_context.py index 556a5d75..ce3901da 100644 --- a/tests/test_protocol_context.py +++ b/tests/test_protocol_context.py @@ -79,7 +79,9 @@ def capture(contexts: list, chain_id: int, labels: dict[str, str]) -> str: self.assertEqual(seen, {TARGET: "Timelock", TOKEN: "Token"}) def test_registered_adapters_cover_the_known_protocols(self) -> None: - self.assertEqual({adapter.name for adapter in protocol_context._ADAPTERS}, {"infinifi", "3jane"}) + self.assertEqual( + {adapter.name for adapter in protocol_context._ADAPTERS}, {"infinifi", "infinifi-outland", "3jane"} + ) if __name__ == "__main__": diff --git a/tests/test_tenderly_simulation.py b/tests/test_tenderly_simulation.py index db6d7f84..a841cd7c 100644 --- a/tests/test_tenderly_simulation.py +++ b/tests/test_tenderly_simulation.py @@ -1,92 +1,70 @@ -"""Tests for utils/tenderly/simulation.py.""" - -import unittest -from unittest.mock import patch - -from utils.tenderly.simulation import ( - _merge_balance_override, - _parse_asset_changes, - _parse_state_changes, - simulate_transaction, -) - - -class TestParseAssetChanges(unittest.TestCase): - """Tests for _parse_asset_changes.""" - - def test_empty_list(self) -> None: - self.assertEqual(_parse_asset_changes([]), []) - - def test_single_transfer(self) -> None: - raw = [ - { - "token_info": { - "contract_address": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", - "name": "USD Coin", - "symbol": "USDC", - "decimals": "6", - }, - "from": "0xSender", - "to": "0xReceiver", - "amount": "1000", - "raw_amount": "1000000000", - } - ] - result = _parse_asset_changes(raw) - self.assertEqual(len(result), 1) - self.assertEqual(result[0].token_symbol, "USDC") - self.assertEqual(result[0].token_name, "USD Coin") - self.assertEqual(result[0].from_address, "0xSender") - self.assertEqual(result[0].to_address, "0xReceiver") - self.assertEqual(result[0].decimals, 6) - - def test_missing_token_info(self) -> None: - raw = [{"from": "0xA", "to": "0xB", "amount": "100", "raw_amount": "100"}] - result = _parse_asset_changes(raw) - self.assertEqual(len(result), 1) - self.assertEqual(result[0].token_symbol, "???") - self.assertEqual(result[0].token_name, "Unknown") +"""Tests for utils/tenderly/simulation.py. +Tests cover balance overrides and status interpretation at the API boundary. +""" -class TestParseStateChanges(unittest.TestCase): - """Tests for _parse_state_changes.""" - - def test_empty_list(self) -> None: - self.assertEqual(_parse_state_changes([]), []) - - def test_single_state_change(self) -> None: - raw = [ - { - "address": "0xContract", - "storage_changes": [ - {"key": "0x01", "original": "0x00", "dirty": "0x01"}, - ], - } - ] - result = _parse_state_changes(raw) - self.assertEqual(len(result), 1) - self.assertEqual(result[0].contract_address, "0xContract") - self.assertEqual(result[0].key, "0x01") - self.assertEqual(result[0].original, "0x00") - self.assertEqual(result[0].dirty, "0x01") - - def test_multiple_storage_changes_same_contract(self) -> None: - raw = [ - { - "address": "0xContract", - "storage_changes": [ - {"key": "0x01", "original": "0x00", "dirty": "0x01"}, - {"key": "0x02", "original": "0x10", "dirty": "0x20"}, - ], - } +import unittest +from unittest.mock import MagicMock, patch + +from utils.tenderly.simulation import BundleCall, _merge_balance_override, _parse_transaction, simulate_bundle + + +class TestSimulationStatus(unittest.TestCase): + """Only explicit success statuses may mark a simulation successful.""" + + def test_status_values(self) -> None: + cases = [ + (True, True), + (False, False), + ("success", True), + ("failed", False), + ("false", False), + ("pending", False), + ("", False), + (None, False), + (1, False), + (["success"], False), + ({"status": "success"}, False), ] - result = _parse_state_changes(raw) - self.assertEqual(len(result), 2) + for status, expected in cases: + with self.subTest(status=status): + result = _parse_transaction( + { + "status": status, + "transaction_info": {"stack_trace": [{"error_reason": "execution reverted"}]}, + }, + raw_response={}, + ) + self.assertIs(result.success, expected) + self.assertEqual(result.error_message, "" if expected else "execution reverted") + + def test_missing_status_is_not_success(self) -> None: + result = _parse_transaction({}, raw_response={}) + self.assertIs(result.success, False) - def test_no_storage_changes(self) -> None: - raw = [{"address": "0xContract"}] - result = _parse_state_changes(raw) - self.assertEqual(len(result), 0) + @patch("utils.tenderly.simulation.fetch_json") + @patch.dict("os.environ", {"TENDERLY_API_KEY": "test-key"}) + def test_failed_bundle_preserves_error_and_skipped_calls(self, mock_fetch: MagicMock) -> None: + mock_fetch.return_value = { + "simulation_results": [ + {"transaction": {"status": "success"}}, + { + "transaction": { + "status": "failed", + "transaction_info": {"stack_trace": [{"error_reason": "execution reverted"}]}, + } + }, + ] + } + results = simulate_bundle([BundleCall("0xTarget", "0x")] * 3, chain_id=1, from_address="0xExec") + assert results is not None + self.assertEqual(len(results), 3) + first, failed, skipped = results + assert first is not None and failed is not None + self.assertIs(first.success, True) + self.assertIs(failed.success, False) + self.assertEqual(failed.error_message, "execution reverted") + self.assertIsNone(skipped) class TestMergeBalanceOverride(unittest.TestCase): @@ -104,107 +82,5 @@ def test_caller_override_wins(self) -> None: self.assertEqual(out["0xExec"]["storage"], {"0x1": "0x2"}) -class TestSimulateTransaction(unittest.TestCase): - """Tests for simulate_transaction.""" - - @patch("utils.tenderly.simulation.fetch_json") - @patch.dict("os.environ", {"TENDERLY_API_KEY": "test-key"}, clear=False) - def test_value_injects_state_objects(self, mock_fetch: unittest.mock.MagicMock) -> None: - mock_fetch.return_value = {"transaction": {"status": True, "transaction_info": {"gas_used": 1}}} - simulate_transaction( - target="0xTarget", - calldata="0x12345678", - chain_id=1, - value=10**18, - from_address="0xExec", - ) - body = mock_fetch.call_args.kwargs["json"] - self.assertEqual(body["state_objects"], {"0xExec": {"balance": hex(10**18)}}) - - @patch.dict("os.environ", {"TENDERLY_API_KEY": ""}, clear=False) - def test_no_api_key_returns_none(self) -> None: - result = simulate_transaction( - target="0xTarget", - calldata="0x12345678", - chain_id=1, - ) - self.assertIsNone(result) - - @patch("utils.tenderly.simulation.fetch_json") - @patch.dict("os.environ", {"TENDERLY_API_KEY": "test-key"}, clear=False) - def test_successful_simulation(self, mock_fetch: unittest.mock.MagicMock) -> None: - mock_fetch.return_value = { - "transaction": { - "status": True, - "transaction_info": { - "gas_used": 50000, - "asset_changes": [ - { - "token_info": { - "contract_address": "0xToken", - "name": "Test Token", - "symbol": "TST", - "decimals": "18", - }, - "from": "0xA", - "to": "0xB", - "amount": "100", - "raw_amount": "100000000000000000000", - } - ], - "state_diff": [], - "logs": [], - }, - } - } - - result = simulate_transaction( - target="0xTarget", - calldata="0x12345678", - chain_id=1, - ) - - self.assertIsNotNone(result) - assert result is not None # for type narrowing - self.assertTrue(result.success) - self.assertEqual(result.gas_used, 50000) - self.assertEqual(len(result.asset_changes), 1) - self.assertEqual(result.asset_changes[0].token_symbol, "TST") - - @patch("utils.tenderly.simulation.fetch_json") - @patch.dict("os.environ", {"TENDERLY_API_KEY": "test-key"}, clear=False) - def test_failed_simulation(self, mock_fetch: unittest.mock.MagicMock) -> None: - mock_fetch.return_value = { - "transaction": { - "status": False, - "transaction_info": { - "gas_used": 21000, - "asset_changes": [], - "state_diff": [], - "logs": [], - "stack_trace": [{"error_reason": "execution reverted"}], - }, - } - } - - result = simulate_transaction( - target="0xTarget", - calldata="0x12345678", - chain_id=1, - ) - - self.assertIsNotNone(result) - assert result is not None - self.assertFalse(result.success) - self.assertEqual(result.error_message, "execution reverted") - - @patch("utils.tenderly.simulation.fetch_json") - @patch.dict("os.environ", {"TENDERLY_API_KEY": "test-key"}, clear=False) - def test_fetch_returns_none(self, mock_fetch: unittest.mock.MagicMock) -> None: - mock_fetch.return_value = None - result = simulate_transaction(target="0xTarget", calldata="0x12345678", chain_id=1) - self.assertIsNone(result) - - if __name__ == "__main__": unittest.main() diff --git a/utils/llm/README.md b/utils/llm/README.md index f896985b..3a342ace 100644 --- a/utils/llm/README.md +++ b/utils/llm/README.md @@ -117,7 +117,7 @@ Requires `TENDERLY_API_KEY`. Simulation failure is non-blocking — the pipeline Callers can pass `skip_simulation=True` to bypass Tenderly entirely. Used for Safe transactions with `operation=DELEGATECALL` (typically multiSend batches), where our plain-CALL simulator can't model the real execution and would produce a spurious "revert" verdict. -Timelock batches still simulate each inner call independently. Successful results are attributed to their original call index and labeled as independent simulations — they do not prove the batch succeeds atomically. Failed simulations are omitted from the risk prompt (Tenderly often false-reverts governance calls) but kept as call-flow diagnostics so a reviewer can see them without treating them as a predicted on-chain failure. +Timelock batches are simulated as one **sequential bundle** (`simulate_bundle`, Tenderly `simulate-bundle`): every call is sent from the executor in batch order, and each sees the state the earlier calls left — the way `executeBatch` runs them. Simulating calls one by one against current state produced false reverts for calls that depend on an earlier one in the same batch (e.g. `Accounting.setOracle(vault, oracle)` followed by `OutlandFarm.setVault(vault)`, which requires that oracle). Tenderly stops at the first revert; calls it never reached, and every call when the bundle request itself fails, fall back to an independent simulation against current state. Results are attributed to their original call index and labeled by mode (`Batch simulation` vs `Independent simulation`). Failed simulations of either kind are omitted from the risk prompt (Tenderly often false-reverts governance calls) but kept as call-flow diagnostics so a reviewer can see them without treating them as a predicted on-chain failure. ### 5. Proxy Upgrade Detection & Implementation Diff (`utils/proxy.py`, `utils/impl_diff.py`) @@ -199,6 +199,17 @@ For Infinifi mainnet alerts, the adapter: The result is added to the LLM prompt as verified protocol context and rendered independently in the Wavey Gist under `## Protocol Context`. The report distinguishes the escrow's accounting asset from non-accounting ERC20 targets it is allowed to interact with; whitelist membership does not establish how a token is valued downstream. Failures are best-effort and never block the governance alert. +### 5e-2. Infinifi Outland Context (`utils/llm/infinifi_outland_context.py`) + +Onboarding a chain to Infinifi's cross-chain Outland spans several contracts whose calls carry no reviewable facts on their own. For Infinifi mainnet alerts, the adapter: + +1. Names the `FarmRegistry.addFarms` / `removeFarms` farm type from the `FarmTypes` library (`0 PROTOCOL`, `1 LIQUID`, `2 MATURITY`). No RPC. +2. For `Accounting.setOracle(asset, oracle)`, reads the oracle's `price()` and the asset's decimals and states the whole-token value in the reference unit (`price * 10^decimals / 1e36`, the `IOracle` convention under which USDC is ~1e30). +3. For `PortalHub.setVault(vault)`, reads the vault's `chainId()` and the hub's registered chains, and says whether the call adds a chain or replaces a live vault. +4. For connector calls naming a destination chain (`enableChainAsset`, `setCctpDomain`, …), reads `chainConfig(chainId)` and states whether the route can send. An unset peer/gas limit means `sendTokens` reverts until a separate `setConfiguration` — the call that sets the destination-side recipient of bridged funds — executes. + +Contracts are identified by the functions their verified ABI exposes (`utils/llm/abi_exposure.py`, following EIP-1967), not by hard-coded addresses. Failures are best-effort and never block the governance alert. + ### 5f. 3Jane Governance Context (`utils/llm/threejane_context.py`) Both 3Jane timelocks schedule calls that arrive as opaque data. `ProtocolConfig.setConfig(bytes32,uint256)` names the parameter it changes only by `keccak256("")`, and `RewardsDistributor.setEpochEmissions` / `updateRoot` allocate JANE without revealing whether a claim mints new supply or moves an existing balance. diff --git a/utils/llm/abi_exposure.py b/utils/llm/abi_exposure.py new file mode 100644 index 00000000..47b563bf --- /dev/null +++ b/utils/llm/abi_exposure.py @@ -0,0 +1,57 @@ +"""Check which functions a contract's verified ABI exposes, following EIP-1967 proxies. + +Protocol context adapters identify contracts by the getters they expose rather +than by hard-coded addresses. Upgradeable contracts sit behind proxies whose own +ABI lists only the proxy's functions, so the implementation is consulted when +the proxy ABI comes up short. +""" + +from functools import lru_cache + +from utils.source_context import fetch_abi_entries + + +def _abi_function_names(entries: list[dict]) -> frozenset[str]: + """Function names present in a verified ABI.""" + return frozenset( + str(entry.get("name")) for entry in entries if entry.get("type") == "function" and entry.get("name") + ) + + +@lru_cache(maxsize=128) +def _own_function_names(chain_id: int, address: str) -> frozenset[str]: + """Function names on the address's own verified ABI. No RPC — Etherscan is cached.""" + return _abi_function_names(fetch_abi_entries(chain_id, address) or []) + + +@lru_cache(maxsize=128) +def _implementation_function_names(chain_id: int, address: str) -> frozenset[str]: + """Function names behind an EIP-1967 proxy, or empty when there is no proxy.""" + from utils.proxy import get_current_implementation + + implementation = get_current_implementation(address, chain_id) + if not implementation or implementation.lower() == address.lower(): + return frozenset() + return _abi_function_names(fetch_abi_entries(chain_id, implementation) or []) + + +def exposes(chain_id: int, address: str, wanted: set[str]) -> bool: + """Whether a contract exposes every wanted function, following EIP-1967. + + Args: + chain_id: Chain the contract lives on. + address: Contract (or proxy) address. + wanted: Function names that must all be present. + + Returns: + True when the address's own ABI, or its implementation's, has them all. + """ + if wanted.issubset(_own_function_names(chain_id, address)): + return True + return wanted.issubset(_implementation_function_names(chain_id, address)) + + +def reset_cache() -> None: + """Reset process caches for tests or long-running workers.""" + _own_function_names.cache_clear() + _implementation_function_names.cache_clear() diff --git a/utils/llm/ai_explainer.py b/utils/llm/ai_explainer.py index 4ad36ffc..4e3f3200 100644 --- a/utils/llm/ai_explainer.py +++ b/utils/llm/ai_explainer.py @@ -50,7 +50,7 @@ get_verification_status, ) from utils.telegram import escape_markdown -from utils.tenderly.simulation import SimulationResult, simulate_transaction +from utils.tenderly.simulation import BundleCall, SimulationResult, simulate_bundle, simulate_transaction from utils.wavey_gist import upload_to_gist logger = get_logger("utils.llm.ai_explainer") @@ -355,6 +355,9 @@ class _PreparedCall: value: int decoded: DecodedCall | None simulation: SimulationResult | None = None + # True when simulated in batch order on the state left by the earlier calls; + # False for an independent simulation against the current chain state. + sequential: bool = False def _state_value_markdown(value: object, chain_id: int, labels: dict[str, str]) -> str: @@ -435,7 +438,7 @@ def _decode_status(data: str, decoded: DecodedCall | None) -> str: return "unknown_selector" -def _simulation_note(sim: SimulationResult | None, *, independent: bool) -> str: +def _simulation_note(sim: SimulationResult | None, *, independent: bool, sequential: bool = False) -> str: """Deterministic per-call simulation line for the gist call flow. Failed sims are labeled as diagnostics so a reviewer can see them without @@ -445,10 +448,17 @@ def _simulation_note(sim: SimulationResult | None, *, independent: bool) -> str: return "" if sim.success: gas = f", gas {sim.gas_used:,}" if sim.gas_used else "" + if sequential: + return f"**Batch simulation:** SUCCESS{gas} (run in batch order on the state left by the preceding calls)" if independent: return f"**Independent simulation:** SUCCESS{gas} (does not prove the batch succeeds atomically)" return f"**Simulation:** SUCCESS{gas}" error = sim.error_message or "reverted" + if sequential: + return ( + f"**Batch simulation diagnostic:** {error} when run in batch order — not a predicted governance " + "failure (omitted from the risk prompt; later calls were simulated independently)" + ) independent_note = "independent simulation; " if independent else "" return ( f"**Simulation diagnostic:** {error} — not a predicted governance failure " @@ -1148,7 +1158,7 @@ def _format_prepared_calldata( def _format_batch_simulation_section(items: list[_PreparedCall]) -> str: - """Prompt section for successful independent per-call simulations. + """Prompt section for successful per-call simulations (batch-order or independent). Failed and missing sims are omitted here (they bias the model toward a false revert). They are still logged and attached to the gist call flow. @@ -1162,12 +1172,16 @@ def _format_batch_simulation_section(items: list[_PreparedCall]) -> str: if len(blocks) >= MAX_PROMPT_SIMULATIONS: omitted += 1 continue - header = f"Call {item.index} (independent simulation; does not prove the batch succeeds atomically):" + header = ( + f"Call {item.index} (simulated in batch order, after calls 1-{item.index - 1}):" + if item.sequential and item.index > 1 + else f"Call {item.index} (simulated in batch order, first call):" + if item.sequential + else f"Call {item.index} (independent simulation; does not prove the batch succeeds atomically):" + ) blocks.append(header + "\n" + _format_simulation_context(sim)) if omitted: - blocks.append( - f"{omitted} further successful independent simulations omitted from this prompt; see the call flow." - ) + blocks.append(f"{omitted} further successful simulations omitted from this prompt; see the call flow.") return "\n\n".join(blocks) @@ -1857,29 +1871,57 @@ def _prepare_batch_items( from_address: str, skip_simulation: bool, ) -> list[_PreparedCall]: - """Decode and optionally simulate each input call, preserving original indices.""" - items: list[_PreparedCall] = [] + """Decode and optionally simulate each input call, preserving original indices. + + Calls are simulated as one sequential bundle, so each sees the state the + earlier calls leave — the way a timelock ``executeBatch`` runs them. A batch + whose call depends on an earlier one (``setOracle`` then ``setVault``) would + otherwise show a false revert. Calls the bundle could not reach (after a + revert), or every call when the bundle request itself fails, fall back to an + independent simulation against the current chain state. + """ + decoded_items: list[tuple[int, str, str, int, DecodedCall | None, str]] = [] for i, call in enumerate(calls, start=1): target = call.get("target", "") data = _normalize_calldata(call.get("data")) value = parse_wei(call.get("value", 0)) decoded = decode_calldata(data, chain_id=chain_id, target=target) if _has_function_selector(data) else None - status = _decode_status(data, decoded) + decoded_items.append((i, target, data, value, decoded, _decode_status(data, decoded))) + + bundle: list[SimulationResult | None] | None = None + if not skip_simulation and len(decoded_items) > 1: + bundle = simulate_bundle( + [BundleCall(target=target, calldata=data, value=value) for _, target, data, value, _, _ in decoded_items], + chain_id=chain_id, + from_address=from_address, + ) + if bundle is None: + logger.warning("Bundle simulation unavailable; simulating batch calls independently") + + items: list[_PreparedCall] = [] + for position, (i, target, data, value, decoded, status) in enumerate(decoded_items): simulation: SimulationResult | None = None + sequential = False # Empty calldata has no function to simulate; a SUCCESS here would look - # like confirmed native delivery, which we must not assert. + # like confirmed native delivery, which we must not assert. It still runs + # in the bundle so later calls see its value transfer. if not skip_simulation and status != "empty_calldata": - simulation = simulate_transaction( - target=target, - calldata=data, - chain_id=chain_id, - value=value, - from_address=from_address, - ) + bundled = bundle[position] if bundle is not None else None + if bundled is not None: + simulation, sequential = bundled, True + else: + simulation = simulate_transaction( + target=target, + calldata=data, + chain_id=chain_id, + value=value, + from_address=from_address, + ) if simulation is not None and not simulation.success: logger.warning( - "Batch call %d simulation reported failure (%s); omitting from prompt", + "Batch call %d %s simulation reported failure (%s); omitting from prompt", i, + "sequential" if sequential else "independent", simulation.error_message, ) items.append( @@ -1890,6 +1932,7 @@ def _prepare_batch_items( value=value, decoded=decoded, simulation=simulation, + sequential=sequential, ) ) return items @@ -1901,7 +1944,6 @@ def _call_entry_from_item( param_names: list[str] | None, role_names: dict[str, str], amount_token: RelatedToken | None, - independent_sim: bool, ) -> CallEntry: """Build a report entry that keeps unknown payloads visible.""" return CallEntry( @@ -1914,7 +1956,7 @@ def _call_entry_from_item( raw_calldata=item.data, original_index=item.index, decode_status=_decode_status(item.data, item.decoded), - simulation_note=_simulation_note(item.simulation, independent=independent_sim), + simulation_note=_simulation_note(item.simulation, independent=True, sequential=item.sequential), ) @@ -2002,7 +2044,6 @@ def _deterministic_undecoded_explanation( param_names=None, role_names={}, amount_token=None, - independent_sim=True, ) for item in items ], @@ -2167,7 +2208,6 @@ def explain_batch_transaction( param_names=names_by_index.get(item.index), role_names=roles_by_target.get(item.target.lower(), {}) if item.target else {}, amount_token=sole_tokens.get(item.target.lower()) if item.target else None, - independent_sim=True, ) for item in items ], diff --git a/utils/llm/infinifi_outland_context.py b/utils/llm/infinifi_outland_context.py new file mode 100644 index 00000000..5656da30 --- /dev/null +++ b/utils/llm/infinifi_outland_context.py @@ -0,0 +1,483 @@ +"""Resolve Infinifi Outland (cross-chain) context for governance calls. + +Onboarding a chain to Infinifi's Outland spans several contracts, and each call +arrives at the LLM without the facts that make it reviewable: + +- ``FarmRegistry.addFarms(type, farms)`` names the farm type by number only. +- ``Accounting.setOracle(asset, oracle)`` carries the oracle address, not the + price it reports or what that price means for the asset's decimals. +- ``PortalHub.setVault(vault)`` keys vaults by the vault's own ``chainId()``, so + the calldata cannot say whether it adds a chain or replaces a live vault. +- Connector calls (``enableChainAsset``, ``setCctpDomain``) do not show whether + the connector can actually send to the chain: ``sendTokens`` also needs a + peer, gas limit and selector from ``setConfiguration`` — and the peer is the + destination-side address that receives the bridged funds. + +This adapter runs only for Infinifi on Ethereum, identifies contracts by the +getters their verified ABI exposes, and reads the surrounding state on-chain. +""" + +from dataclasses import dataclass +from decimal import Decimal + +from eth_utils import to_checksum_address + +from utils.calldata.decoder import DecodedCall +from utils.chains import Chain +from utils.erc20_metadata import fetch_erc20_metadata +from utils.llm.abi_exposure import exposes +from utils.llm.report import address_link +from utils.logger import get_logger +from utils.web3_wrapper import ChainManager + +logger = get_logger("utils.llm.infinifi_outland_context") + +PROTOCOL = "infinifi" + +ZERO_ADDRESS = "0x0000000000000000000000000000000000000000" + +# FarmTypes library: the uint256 FarmRegistry.addFarms takes as its first argument. +FARM_TYPES: dict[int, tuple[str, str]] = { + 0: ("PROTOCOL", "not generating yield but capable of storing funds"), + 1: ("LIQUID", "instant principal withdrawals (e.g. Aave)"), + 2: ("MATURITY", "illiquid: principal is locked until the farm's maturity"), +} + +# IOracle.price() scale: a whole token's value in the reference unit is +# price * 10**decimals / 1e36 (USDC is quoted at ~1e30 for a 1:1 price). +_ORACLE_PRICE_SCALE = Decimal(10) ** 36 + +_CONNECTOR_CHAIN_CALLS = {"enableChainAsset", "disableChainAsset", "setCctpDomain", "setConfiguration"} + +_ADDRESS_OUT = [{"name": "", "type": "address"}] +_UINT_OUT = [{"name": "", "type": "uint256"}] + +_ORACLE_ABI = [{"name": "price", "type": "function", "stateMutability": "view", "inputs": [], "outputs": _UINT_OUT}] +_VAULT_ABI = [{"name": "chainId", "type": "function", "stateMutability": "view", "inputs": [], "outputs": _UINT_OUT}] +_HUB_ABI = [ + { + "name": "getVaultChainIds", + "type": "function", + "stateMutability": "view", + "inputs": [], + "outputs": [{"name": "", "type": "uint256[]"}], + }, + { + "name": "getVault", + "type": "function", + "stateMutability": "view", + "inputs": [{"name": "_chainId", "type": "uint256"}], + "outputs": _ADDRESS_OUT, + }, +] +_CONNECTOR_ABI = [ + { + "name": "chainConfig", + "type": "function", + "stateMutability": "view", + "inputs": [{"name": "chainId", "type": "uint256"}], + "outputs": [ + {"name": "peer", "type": "address"}, + {"name": "gasLimit", "type": "uint128"}, + {"name": "chainSelector", "type": "uint128"}, + ], + } +] + + +@dataclass(frozen=True) +class FarmTypeContext: + """A FarmRegistry farm-type number resolved to its FarmTypes name.""" + + registry: str + function_name: str + farm_type: int + farms: tuple[str, ...] + + @property + def addresses(self) -> list[str]: + return [self.registry, *self.farms] + + @property + def labels(self) -> dict[str, str]: + return {} + + @property + def type_name(self) -> str: + return FARM_TYPES.get(self.farm_type, ("UNKNOWN", ""))[0] + + @property + def type_note(self) -> str: + return FARM_TYPES.get(self.farm_type, ("", "not a FarmTypes constant"))[1] + + +@dataclass(frozen=True) +class OracleAssignmentContext: + """The price an oracle reports for the asset it is being assigned to.""" + + accounting: str + asset: str + asset_symbol: str + asset_decimals: int + oracle: str + price_raw: int + + @property + def addresses(self) -> list[str]: + return [self.asset, self.oracle] + + @property + def labels(self) -> dict[str, str]: + return {} + + @property + def unit_price(self) -> Decimal: + """Reference-unit value of one whole asset token (1 = parity with USDC).""" + return Decimal(self.price_raw) * (Decimal(10) ** self.asset_decimals) / _ORACLE_PRICE_SCALE + + +@dataclass(frozen=True) +class HubVaultContext: + """How a PortalHub.setVault changes the per-chain vault registry.""" + + hub: str + vault: str + vault_chain_id: int + registered_chain_ids: tuple[int, ...] + replaced_vault: str | None + + @property + def addresses(self) -> list[str]: + return [self.hub, self.vault] + ([self.replaced_vault] if self.replaced_vault else []) + + @property + def labels(self) -> dict[str, str]: + return {} + + +@dataclass(frozen=True) +class RouteConfig: + """A connector's send configuration for one destination chain.""" + + peer: str + gas_limit: int + chain_selector: int + + @property + def is_configured(self) -> bool: + return self.peer != ZERO_ADDRESS and self.gas_limit != 0 + + def describe(self) -> str: + return f"peer {self.peer}, gas limit {self.gas_limit:,}, selector {self.chain_selector}" + + +@dataclass(frozen=True) +class ConnectorRouteContext: + """Whether a connector can send to a chain once the batch executes.""" + + connector: str + chain_id: int + # chainConfig read on-chain, before the batch executes. + current: RouteConfig + # The last setConfiguration this batch makes for the chain, if any; it + # overrides ``current`` once the batch executes. + proposed: RouteConfig | None = None + + @property + def addresses(self) -> list[str]: + peers = [config.peer for config in (self.current, self.proposed) if config and config.peer != ZERO_ADDRESS] + return list(dict.fromkeys([self.connector, *peers])) + + @property + def labels(self) -> dict[str, str]: + return {} + + @property + def after_batch(self) -> RouteConfig: + """The configuration in force once the batch executes.""" + return self.proposed or self.current + + +OutlandContext = FarmTypeContext | OracleAssignmentContext | HubVaultContext | ConnectorRouteContext + + +def _uint_param(call: DecodedCall, position: int) -> int | None: + """The call's ``position``-th argument when it is an unsigned integer.""" + if len(call.params) <= position: + return None + type_str, value = call.params[position] + return int(value) if type_str.startswith("uint") and isinstance(value, int) else None + + +def _address_param(call: DecodedCall, position: int) -> str | None: + """The call's ``position``-th argument as a checksum address, when it is one.""" + if len(call.params) <= position: + return None + type_str, value = call.params[position] + if type_str != "address" or not isinstance(value, str): + return None + try: + return to_checksum_address(value) + except ValueError: + return None + + +def _farm_type_context(target: str, call: DecodedCall) -> FarmTypeContext | None: + """Name the farm type in an ``addFarms`` / ``removeFarms`` call. No RPC.""" + if call.function_name not in {"addFarms", "removeFarms"}: + return None + farm_type = _uint_param(call, 0) + if farm_type is None or len(call.params) < 2: + return None + type_str, farms = call.params[1] + if type_str != "address[]" or not isinstance(farms, (list, tuple)): + return None + return FarmTypeContext( + registry=target, + function_name=call.function_name, + farm_type=farm_type, + farms=tuple(to_checksum_address(str(farm)) for farm in farms), + ) + + +def _oracle_context(chain_id: int, target: str, call: DecodedCall) -> OracleAssignmentContext | None: + """Read the price a newly assigned oracle reports and scale it by the asset's decimals.""" + if call.function_name != "setOracle": + return None + asset, oracle = _address_param(call, 0), _address_param(call, 1) + if asset is None or oracle is None or oracle == ZERO_ADDRESS: + return None + if not exposes(chain_id, target, {"setOracle", "oracle", "price"}): + return None + metadata = fetch_erc20_metadata(chain_id, asset) + if metadata is None: + return None + client = ChainManager.get_client(Chain.from_chain_id(chain_id)) + price = client.get_contract(oracle, _ORACLE_ABI).functions.price().call() + return OracleAssignmentContext( + accounting=target, + asset=asset, + asset_symbol=metadata.symbol, + asset_decimals=metadata.decimals, + oracle=oracle, + price_raw=int(price), + ) + + +def _hub_vault_context(chain_id: int, target: str, call: DecodedCall) -> HubVaultContext | None: + """Read which chains a PortalHub has vaults for, and what this setVault replaces.""" + if call.function_name != "setVault": + return None + vault = _address_param(call, 0) + if vault is None or not exposes(chain_id, target, {"getVaultChainIds", "getVault", "setVault"}): + return None + client = ChainManager.get_client(Chain.from_chain_id(chain_id)) + hub = client.get_contract(target, _HUB_ABI) + with client.batch_requests() as batch: + batch.add(client.get_contract(vault, _VAULT_ABI).functions.chainId()) + batch.add(hub.functions.getVaultChainIds()) + vault_chain_id, chain_ids = client.execute_batch(batch) + registered = tuple(int(chain) for chain in chain_ids) + replaced = None + if int(vault_chain_id) in registered: + replaced = to_checksum_address(hub.functions.getVault(int(vault_chain_id)).call()) + return HubVaultContext( + hub=target, + vault=vault, + vault_chain_id=int(vault_chain_id), + registered_chain_ids=registered, + replaced_vault=replaced, + ) + + +def _proposed_route_configs(calls: list[DecodedCall]) -> dict[int, RouteConfig]: + """Route configs the batch's ``setConfiguration(chainId, peer, selector, gasLimit)`` calls set. + + A later call for the same chain wins, as it would on execution. + """ + configs: dict[int, RouteConfig] = {} + for call in calls: + if call.function_name != "setConfiguration": + continue + destination, peer = _uint_param(call, 0), _address_param(call, 1) + selector, gas_limit = _uint_param(call, 2), _uint_param(call, 3) + if destination is None or peer is None or selector is None or gas_limit is None: + continue + configs[destination] = RouteConfig(peer, gas_limit, selector) + return configs + + +def _connector_contexts(chain_id: int, target: str, calls: list[DecodedCall]) -> list[ConnectorRouteContext]: + """Read each touched destination chain's send configuration on a connector.""" + chain_ids = list( + dict.fromkeys( + chain + for call in calls + if call.function_name in _CONNECTOR_CHAIN_CALLS + for chain in [_uint_param(call, 0)] + if chain is not None + ) + ) + if not chain_ids or not exposes(chain_id, target, {"chainConfig", "portal"}): + return [] + proposed = _proposed_route_configs(calls) + client = ChainManager.get_client(Chain.from_chain_id(chain_id)) + connector = client.get_contract(target, _CONNECTOR_ABI) + with client.batch_requests() as batch: + for destination in chain_ids: + batch.add(connector.functions.chainConfig(destination)) + configs = client.execute_batch(batch) + return [ + ConnectorRouteContext( + connector=target, + chain_id=destination, + current=RouteConfig(to_checksum_address(str(peer)), int(gas_limit), int(selector)), + proposed=proposed.get(destination), + ) + for destination, (peer, gas_limit, selector) in zip(chain_ids, configs) + ] + + +def resolve_outland_context( + protocol: str, + chain_id: int, + targets_and_calls: list[tuple[str, DecodedCall]], +) -> list[OutlandContext]: + """Resolve deterministic Outland context for the calls in one alert.""" + if protocol.lower() != PROTOCOL or chain_id != Chain.MAINNET.chain_id: + return [] + + calls_by_target: dict[str, list[DecodedCall]] = {} + for target, call in targets_and_calls: + try: + calls_by_target.setdefault(to_checksum_address(target), []).append(call) + except ValueError: + continue + + contexts: list[OutlandContext] = [] + for target, calls in calls_by_target.items(): + for call in calls: + try: + resolved = ( + _farm_type_context(target, call) + or _oracle_context(chain_id, target, call) + or _hub_vault_context(chain_id, target, call) + ) + except Exception as error: # noqa: BLE001 - enrichment must never block an alert + logger.info("Outland context failed for %s.%s: %s", target, call.function_name, error) + continue + if resolved is not None: + contexts.append(resolved) + try: + contexts.extend(_connector_contexts(chain_id, target, calls)) + except Exception as error: # noqa: BLE001 - enrichment must never block an alert + logger.info("Outland connector context failed for %s: %s", target, error) + return contexts + + +def _format_unit_price(context: OracleAssignmentContext) -> str: + """Whole-token price in the reference unit, e.g. ``1`` or ``0.9985``.""" + return f"{context.unit_price.normalize():f}" + + +def _route_status(context: ConnectorRouteContext) -> str: + """One sentence on whether the connector can send to the chain after this batch.""" + current, proposed = context.current, context.proposed + before = current.describe() if current.is_configured else "not configured" + if proposed is not None: + if not proposed.is_configured: + return ( + f"this batch CLEARS the route via setConfiguration ({proposed.describe()}); sendTokens to this " + f"chain reverts afterwards. Before the batch: {before}" + ) + return ( + f"this batch sets it via setConfiguration to {proposed.describe()} — the peer is the destination-side " + f"address that receives the bridged funds. Before the batch: {before}" + ) + if current.is_configured: + return f"configured, unchanged by this batch — {current.describe()}" + return ( + "NOT configured (peer, gas limit and selector all unset). sendTokens to this chain reverts " + "(MissingPeer / NoGasLimit) until a separate setConfiguration(chainId, peer, selector, gasLimit) " + "executes — that later call sets the peer, the destination-side address that receives the bridged " + "funds, so the route is not live after this batch" + ) + + +def _hub_vault_line(context: HubVaultContext) -> str: + registered = ", ".join(str(chain) for chain in context.registered_chain_ids) or "none" + if context.replaced_vault: + effect = f"REPLACES the existing chain-{context.vault_chain_id} vault {context.replaced_vault}" + else: + effect = f"adds chain {context.vault_chain_id}; no existing vault is replaced" + return ( + f"PortalHub {context.hub} keys vaults by chain id. setVault({context.vault}) targets chain " + f"{context.vault_chain_id} and {effect}. Chains registered before this call: {registered}." + ) + + +def format_outland_prompt(contexts: list[OutlandContext]) -> str: + """Render verified Outland context for the LLM prompt.""" + lines: list[str] = [] + for context in contexts: + if isinstance(context, FarmTypeContext): + lines.append( + f"FarmRegistry {context.registry}.{context.function_name}: farm type {context.farm_type} = " + f"FarmTypes.{context.type_name} — {context.type_note}" + ) + elif isinstance(context, OracleAssignmentContext): + lines.append( + f"Oracle {context.oracle} assigned to {context.asset} ({context.asset_symbol}, " + f"{context.asset_decimals} decimals) reports price() = {context.price_raw}, i.e. one whole " + f"{context.asset_symbol} is valued at {_format_unit_price(context)} reference units " + "(IOracle scale: price * 10^decimals / 1e36; USDC is ~1)" + ) + elif isinstance(context, HubVaultContext): + lines.append(_hub_vault_line(context)) + else: + lines.append(f"Connector {context.connector} route to chain {context.chain_id}: {_route_status(context)}.") + return "\n".join(lines) + + +def format_outland_report(contexts: list[OutlandContext], chain_id: int, labels: dict[str, str]) -> str: + """Render the deterministic Outland section for the gist report.""" + lines: list[str] = [] + for context in contexts: + if isinstance(context, FarmTypeContext): + lines.append( + f"- **Farm type {context.farm_type}:** `FarmTypes.{context.type_name}` — {context.type_note} " + f"({address_link(context.registry, chain_id, labels)} `{context.function_name}`)" + ) + elif isinstance(context, OracleAssignmentContext): + lines.append( + f"- **Oracle price:** {address_link(context.oracle, chain_id, labels)} reports `{context.price_raw}` " + f"→ one whole `{context.asset_symbol}` = `{_format_unit_price(context)}` reference units " + "(USDC ≈ 1)" + ) + elif isinstance(context, HubVaultContext): + registered = ", ".join(f"`{chain}`" for chain in context.registered_chain_ids) or "none" + effect = ( + f"replaces {address_link(context.replaced_vault, chain_id, labels)}" + if context.replaced_vault + else "new chain, no vault replaced" + ) + lines.append( + f"- **PortalHub vault for chain `{context.vault_chain_id}`:** {effect}; " + f"chains registered before: {registered}" + ) + else: + if context.proposed is not None: + change = "set" if context.proposed.is_configured else "**cleared**" + status = f"{change} by this batch — peer {address_link(context.proposed.peer, chain_id, labels)}" + status += f", gas limit `{context.proposed.gas_limit:,}`, selector `{context.proposed.chain_selector}`" + if context.current.is_configured: + status += f" (was peer {address_link(context.current.peer, chain_id, labels)})" + elif context.current.is_configured: + status = f"configured, unchanged — peer {address_link(context.current.peer, chain_id, labels)}" + else: + status = "**not configured** — sends revert until a later `setConfiguration` sets the peer" + lines.append( + f"- **Connector route to chain `{context.chain_id}`** " + f"({address_link(context.connector, chain_id, labels)}): {status}" + ) + return "\n".join(lines) diff --git a/utils/llm/protocol_context.py b/utils/llm/protocol_context.py index 27072392..4b031bb2 100644 --- a/utils/llm/protocol_context.py +++ b/utils/llm/protocol_context.py @@ -22,6 +22,11 @@ format_infinifi_report, resolve_infinifi_context, ) +from utils.llm.infinifi_outland_context import ( + format_outland_prompt, + format_outland_report, + resolve_outland_context, +) from utils.llm.threejane_context import ( format_threejane_prompt, format_threejane_report, @@ -44,6 +49,7 @@ class _Adapter: _ADAPTERS: tuple[_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), ) diff --git a/utils/on_chain_state.py b/utils/on_chain_state.py index 9d84e2c0..2f318c2f 100644 --- a/utils/on_chain_state.py +++ b/utils/on_chain_state.py @@ -82,11 +82,14 @@ def _parse_var_declaration(snippet: str, var_name: str) -> tuple[str, list[str]] decl = " ".join(line.strip() for line in decl_lines).strip() decl = decl.rstrip(";") - # Mapping case: mapping(K => V) public name - m = re.match(r"mapping\s*\(\s*(\w+)\s*=>\s*(.+?)\s*\)\s+(?:public|external)\s+\w+\s*$", decl) + # Mapping case: mapping(K => V) public name. Solidity >=0.8.18 also allows + # named parameters — mapping(uint256 chainId => bool configured) — whose + # names must be dropped, or the declaration falls through to the setter- + # signature guess and reports the wrong value type. + m = re.match(r"mapping\s*\(\s*(\w+)(?:\s+\w+)?\s*=>\s*(.+?)\s*\)\s+(?:public|external)\s+\w+\s*$", decl) if m: key_type = m.group(1).strip() - value_type = m.group(2).strip() + value_type = re.sub(r"^(\w+)\s+\w+$", r"\1", m.group(2).strip()) if "mapping" in value_type or "[" in value_type: return None # nested mapping or array value — skip if not _is_simple_type(value_type): diff --git a/utils/tenderly/simulation.py b/utils/tenderly/simulation.py index 67101d57..f85aec1d 100644 --- a/utils/tenderly/simulation.py +++ b/utils/tenderly/simulation.py @@ -174,34 +174,110 @@ def simulate_transaction( return None try: - tx = data.get("transaction", {}) - tx_info = tx.get("transaction_info", {}) - - success = tx.get("status", False) - gas_used = int(tx_info.get("gas_used", 0)) - error_message = tx_info.get("stack_trace", [{}])[0].get("error_reason", "") if not success else "" - - # Parse asset changes - raw_asset_changes = tx_info.get("asset_changes", []) or [] - asset_changes = _parse_asset_changes(raw_asset_changes) - - # Parse state changes - raw_state_diff = tx_info.get("state_diff", []) or [] - state_changes = _parse_state_changes(raw_state_diff) - - # Parse logs/events - logs = tx_info.get("logs", []) or [] - - return SimulationResult( - success=success, - gas_used=gas_used, - asset_changes=asset_changes, - state_changes=state_changes, - logs=logs, - error_message=error_message, - raw_response=data, - ) - + return _parse_transaction(data.get("transaction", {}) or {}, raw_response=data) except Exception: logger.exception("Failed to parse Tenderly simulation response") return None + + +def _parse_transaction(tx: dict[str, Any], raw_response: dict[str, Any]) -> SimulationResult: + """Parse one simulated ``transaction`` object (single or bundle response).""" + tx_info = tx.get("transaction_info", {}) or {} + status = tx.get("status", False) + if isinstance(status, str): + success = status == "success" + else: + success = status is True + # Single simulations report gas inside transaction_info; bundle results only on the transaction. + gas_used = int(tx_info.get("gas_used") or tx.get("gas_used") or 0) + error_message = "" if success else ((tx_info.get("stack_trace") or [{}])[0].get("error_reason") or "") + return SimulationResult( + success=success, + gas_used=gas_used, + asset_changes=_parse_asset_changes(tx_info.get("asset_changes", []) or []), + state_changes=_parse_state_changes(tx_info.get("state_diff", []) or []), + logs=tx_info.get("logs", []) or [], + error_message=error_message, + raw_response=raw_response, + ) + + +@dataclass(frozen=True) +class BundleCall: + """One call in a sequential bundle simulation.""" + + target: str + calldata: str + value: int = 0 + + +def simulate_bundle( + calls: list[BundleCall], + chain_id: int, + from_address: str, +) -> list[SimulationResult | None] | None: + """Simulate calls in order on shared state, as a timelock batch executes them. + + Each call sees the state left by the ones before it, so a call that depends + on an earlier one (``setOracle`` then ``setVault``) is not reported as a + false revert. Tenderly stops at the first failure: the result list is + aligned with ``calls`` and holds ``None`` for every call after it. + + Args: + calls: Calls in batch order, all sent from ``from_address``. + chain_id: Chain ID (e.g. 1 for mainnet). + from_address: The executor (``msg.sender`` of every call in the batch). + + Returns: + One entry per call, or None when the bundle could not be simulated at + all (no API key, request failure, unparsable response). + """ + api_key = os.getenv("TENDERLY_API_KEY") + if not api_key: + logger.warning("TENDERLY_API_KEY not set, skipping bundle simulation") + return None + if not calls: + return [] + + overrides = _merge_balance_override(None, from_address, sum(call.value for call in calls)) + simulations: list[dict[str, Any]] = [] + for index, call in enumerate(calls): + simulation: dict[str, Any] = { + "network_id": str(chain_id), + "from": from_address, + "to": call.target, + "input": call.calldata, + "value": str(call.value), + "save": False, + "save_if_fails": False, + "simulation_type": "full", + } + # Overrides apply before the first call; later calls inherit the state. + if index == 0 and overrides: + simulation["state_objects"] = overrides + simulations.append(simulation) + + logger.info("Simulating bundle: %d calls from=%s chain=%s", len(calls), from_address, chain_id) + data = fetch_json( + f"{_get_simulation_url()}-bundle", + method="post", + json={"simulations": simulations}, + headers={"X-Access-Key": api_key}, + timeout=60, + ) + if not data: + logger.error("Tenderly bundle simulation returned no data") + return None + + try: + raw_results = data.get("simulation_results", []) or [] + results: list[SimulationResult | None] = [ + _parse_transaction(item.get("transaction", {}) or {}, raw_response=item) for item in raw_results + ] + except Exception: + logger.exception("Failed to parse Tenderly bundle simulation response") + return None + if not results: + logger.error("Tenderly bundle simulation returned no results") + return None + return (results + [None] * len(calls))[: len(calls)]