diff --git a/tests/ai_explainer/test_state.py b/tests/ai_explainer/test_state.py index e4160279..c61bf22f 100644 --- a/tests/ai_explainer/test_state.py +++ b/tests/ai_explainer/test_state.py @@ -3,6 +3,9 @@ import unittest from unittest.mock import MagicMock, patch +from eth_abi import encode +from eth_utils import function_signature_to_4byte_selector + from utils.calldata.decoder import DecodedCall, decode_calldata from utils.llm.ai_explainer import ( MAX_STATE_READ_KEYS_PER_SIGNATURE, @@ -133,6 +136,69 @@ def test_resolves_role_arguments(self, mock_resolve: MagicMock) -> None: self.assertEqual(resolved["0xcore"][self.MINTER_HASH], "RECEIPT_TOKEN_MINTER") self.assertIn("is the role RECEIPT_TOKEN_MINTER", "\n".join(_format_role_name_notes(resolved))) + @patch("utils.source_context.fetch_source") + def test_infinifi_batch_roles_reach_prompt_and_report(self, mock_source: MagicMock) -> None: + """Reproduce the priority-hook batch with real ABI encoding and source harvesting.""" + from utils.llm.ai_explainer import _collect_role_names, _format_role_name_notes + from utils.llm.report import CallEntry, ReportContext, format_call_flow + + rebalancer = "0x81c2da19079a6b9586648186d4d64ede4daff8cc96fb1e2cbf9c9bb9c3b3bb94" + swap_caller = "0x456cfaf8d1ec98ae5bbe595a448911a58cb2e264d4686992e15dec9d0f363e03" + target = "0xF6d48735EcCf12bDC1DF2674b1ce3fcb3bD25490" + accounts = [ + "0x20f1B8FB379d4cc07b917C68Ee855a1F35ef3Ed6", + "0x006E2897400da44CF772cD2664C1a9e712755909", + "0x3D18480CC32B6AB3B833dCabD80E76CfD41c48a9", + ] + mock_source.return_value = ( + "InfiniFiCore", + 'bytes32 internal constant MANUAL_REBALANCER = keccak256("MANUAL_REBALANCER");\n' + 'bytes32 internal constant FARM_SWAP_CALLER = keccak256("FARM_SWAP_CALLER");', + ) + signature = "grantRoles(bytes32[],address[])" + data = function_signature_to_4byte_selector(signature) + encode( + ["bytes32[]", "address[]"], + [[bytes.fromhex(h[2:]) for h in (rebalancer, rebalancer, swap_caller)], accounts], + ) + with patch("utils.calldata.decoder.resolve_selector", return_value=signature): + call = decode_calldata("0x" + data.hex()) + assert call is not None + resolved = _collect_role_names([(target, call)], chain_id=1) + self.assertEqual(resolved[target.lower()], {rebalancer: "MANUAL_REBALANCER", swap_caller: "FARM_SWAP_CALLER"}) + mock_source.assert_called_once_with(1, target) + notes = "\n".join(_format_role_name_notes(resolved)) + self.assertIn("is the role MANUAL_REBALANCER", notes) + self.assertIn("is the role FARM_SWAP_CALLER", notes) + flow = format_call_flow( + ReportContext( + chain_id=1, + entries=[CallEntry(target=target, call=call, role_names=resolved[target.lower()])], + ) + ) + self.assertEqual(flow.count(f"`{rebalancer}` — role **MANUAL_REBALANCER**"), 2) + self.assertIn(f"`{swap_caller}` — role **FARM_SWAP_CALLER**", flow) + for account in accounts: + self.assertIn(f"https://etherscan.io/address/{account}", flow) + + @patch("utils.llm.ai_explainer.resolve_role_names", return_value={}) + def test_fixed_role_array_and_empty_array(self, mock_resolve: MagicMock) -> None: + """Skip malformed byte values and avoid lookups for an empty role array.""" + from utils.llm.ai_explainer import _collect_role_names + + call = DecodedCall( + function_name="grantRoles", + signature="grantRoles(bytes32[2])", + params=[("bytes32[2]", (bytes.fromhex(self.MINTER_HASH[2:]), b"short"))], + ) + _collect_role_names([("0xCore", call)], chain_id=1) + mock_resolve.assert_called_once_with([self.MINTER_HASH], chain_id=1, target="0xCore") + mock_resolve.reset_mock() + empty_call = DecodedCall( + function_name="grantRoles", signature="grantRoles(bytes32[])", params=[("bytes32[]", ())] + ) + self.assertEqual(_collect_role_names([("0xCore", empty_call)], chain_id=1), {}) + mock_resolve.assert_not_called() + @patch("utils.llm.ai_explainer.resolve_role_names") def test_ignores_bytes32_on_non_role_functions(self, mock_resolve: MagicMock) -> None: """A timelock's all-zero predecessor/salt must not become DEFAULT_ADMIN_ROLE.""" @@ -141,7 +207,12 @@ def test_ignores_bytes32_on_non_role_functions(self, mock_resolve: MagicMock) -> schedule = DecodedCall( function_name="scheduleBatch", signature="scheduleBatch(address[],uint256[],bytes[],bytes32,bytes32,uint256)", - params=[("bytes32", b"\x00" * 32), ("bytes32", b"\x00" * 32), ("uint256", 604800)], + params=[ + ("bytes32", b"\x00" * 32), + ("bytes32", b"\x00" * 32), + ("bytes32[]", (b"\x00" * 32,)), + ("uint256", 604800), + ], ) self.assertEqual(_collect_role_names([("0xTimelock", schedule)], chain_id=1), {}) mock_resolve.assert_not_called() diff --git a/tests/test_role_context.py b/tests/test_role_context.py new file mode 100644 index 00000000..7587d660 --- /dev/null +++ b/tests/test_role_context.py @@ -0,0 +1,245 @@ +"""Role names retain their call and target scope through governance wrappers.""" + +import unittest +from unittest.mock import MagicMock, patch + +from eth_utils import function_signature_to_4byte_selector + +from tests.ai_explainer.helpers import make_provider +from tests.test_calldata_wrappers import ZERO32, encode_call, execute_batch +from utils.calldata.decoder import MAX_BYTES_RECURSION_DEPTH, DecodedCall, decode_calldata +from utils.calldata.role_names import DEFAULT_ADMIN_ROLE +from utils.calldata.wrappers import unwrap_call_values, unwrap_calls +from utils.llm.ai_explainer import _collect_role_names, explain_batch_transaction, explain_transaction +from utils.llm.protocol_context import ResolvedProtocolContext +from utils.llm.report import CallEntry, ReportContext, format_call_flow + +CORE = "0xF6d48735EcCf12bDC1DF2674b1ce3fcb3bD25490" +OTHER = "0x20f1B8FB379d4cc07b917C68Ee855a1F35ef3Ed6" +TIMELOCK = "0x3D18480CC32B6AB3B833dCabD80E76CfD41c48a9" +MINTER = "0x615a688d53344290b742a2e72e4f187e5b88227c01f9d77ce2406d32f8bd0eda" +BURNER = "0xc843f0b739b89f713c29d5349e88538222378b9e38ad958b090593af5e3d0fd9" +SOURCE = ( + 'bytes32 internal constant RECEIPT_TOKEN_MINTER = keccak256("RECEIPT_TOKEN_MINTER");\n' + 'bytes32 internal constant RECEIPT_TOKEN_BURNER = keccak256("RECEIPT_TOKEN_BURNER");' +) +GRANT_SIGNATURE = "grantRoles(bytes32[],address[])" + + +def grant_data(role: str = MINTER) -> str: + """Encode a batch grant with one role and one recipient.""" + return encode_call(GRANT_SIGNATURE, ["bytes32[]", "address[]"], [[bytes.fromhex(role[2:])], [OTHER]]) + + +def decoded(data: str) -> DecodedCall: + """Decode test calldata, failing immediately if its signature was not registered.""" + call = decode_calldata(data) + assert call is not None + return call + + +def scheduled(data: str, target: str = CORE) -> str: + """Wrap one payload in a timelock schedule call.""" + return encode_call( + "schedule(address,uint256,bytes,bytes32,bytes32,uint256)", + ["address", "uint256", "bytes", "bytes32", "bytes32", "uint256"], + [target, 0, bytes.fromhex(data[2:]), ZERO32, ZERO32, 604800], + ) + + +class TestRoleContext(unittest.TestCase): + """Exercise source resolution and report rendering with real encoded wrappers.""" + + def setUp(self) -> None: + """Keep custom role signatures and verified source entirely offline.""" + signature_patch = patch.dict( + "utils.calldata.decoder.KNOWN_SELECTORS", + { + "0x" + function_signature_to_4byte_selector(signature).hex(): signature + for signature in (GRANT_SIGNATURE, "queueTransaction(address,uint256,string,bytes,uint256)") + }, + ) + signature_patch.start() + self.addCleanup(signature_patch.stop) + source_patch = patch("utils.source_context.fetch_source", return_value=("Core", SOURCE)) + self.source = source_patch.start() + self.addCleanup(source_patch.stop) + network_patch = patch( + "utils.calldata.decoder.fetch_json", side_effect=AssertionError("Unexpected network lookup") + ) + network_patch.start() + self.addCleanup(network_patch.stop) + + def _flow(self, data: str) -> str: + """Resolve a wrapped grant and render using each nested target's own roles.""" + call = decoded(data) + roles = _collect_role_names([(TIMELOCK, call)], 1) + self.assertEqual(roles, {CORE.lower(): {MINTER: "RECEIPT_TOKEN_MINTER"}}) + return format_call_flow( + ReportContext(entries=[CallEntry(target=TIMELOCK, call=call)], chain_id=1, role_names_by_target=roles) + ) + + def test_executed_and_scheduled_grants_are_resolved(self) -> None: + """Single, batch, and nested wrappers all expose their proposed role grants.""" + grant = grant_data() + schedule_batch = encode_call( + "scheduleBatch(address[],uint256[],bytes[],bytes32,bytes32,uint256)", + ["address[]", "uint256[]", "bytes[]", "bytes32", "bytes32", "uint256"], + [[CORE], [0], [bytes.fromhex(grant[2:])], ZERO32, ZERO32, 604800], + ) + for data in ( + scheduled(grant), + schedule_batch, + execute_batch([CORE], [grant]), + execute_batch([TIMELOCK], [scheduled(grant)]), + ): + with self.subTest(selector=data[:10]): + self.assertIn(f"`{MINTER}` — role **RECEIPT_TOKEN_MINTER**", self._flow(data)) + + def test_compound_signature_is_restored_for_nested_grant(self) -> None: + """Compound queues store selector-free arguments when a signature is supplied.""" + data = encode_call( + "queueTransaction(address,uint256,string,bytes,uint256)", + ["address", "uint256", "string", "bytes", "uint256"], + [CORE, 0, GRANT_SIGNATURE, bytes.fromhex(grant_data()[10:]), 1800000000], + ) + self.assertIn("role **RECEIPT_TOKEN_MINTER**", self._flow(data)) + + def test_executor_script_uses_each_inner_target(self) -> None: + """Equal hashes on different script targets never share source-derived labels.""" + payload = bytes.fromhex(grant_data()[2:]) + script = b"".join( + ((int(target, 16) << 96) | len(payload)).to_bytes(32, "big") + payload for target in (CORE, OTHER) + ) + data = encode_call("execute(bytes)", ["bytes"], [script]) + call = decoded(data) + roles = _collect_role_names([(TIMELOCK, call)], 1) + self.assertEqual(set(roles), {CORE.lower(), OTHER.lower()}) + roles[OTHER.lower()][MINTER] = "OTHER_TARGET_ROLE" + flow = format_call_flow( + ReportContext(entries=[CallEntry(target=TIMELOCK, call=call)], chain_id=1, role_names_by_target=roles) + ) + self.assertEqual(flow.count("role **RECEIPT_TOKEN_MINTER**"), 1) + self.assertEqual(flow.count("role **OTHER_TARGET_ROLE**"), 1) + + def test_batch_payloads_with_equal_calldata_keep_target_scope(self) -> None: + """Match payloads by their wrapper order, including duplicate calldata.""" + data = execute_batch([CORE, OTHER], [grant_data(), grant_data()]) + roles: dict[str, dict[str, str]] = {CORE.lower(): {MINTER: "CORE_ROLE"}, OTHER.lower(): {MINTER: "OTHER_ROLE"}} + flow = format_call_flow( + ReportContext( + entries=[CallEntry(target=TIMELOCK, call=decoded(data))], chain_id=1, role_names_by_target=roles + ) + ) + self.assertEqual(flow.count("role **CORE_ROLE**"), 1) + self.assertEqual(flow.count("role **OTHER_ROLE**"), 1) + + def test_generic_bytes_without_known_target_do_not_inherit_roles(self) -> None: + """Recognizing a nested selector does not establish which contract it targets.""" + call = DecodedCall("relay", "relay(bytes)", [("bytes", bytes.fromhex(grant_data()[2:]))]) + roles: dict[str, dict[str, str]] = {TIMELOCK.lower(): {MINTER: "PARENT_ROLE"}} + flow = format_call_flow( + ReportContext(entries=[CallEntry(target=TIMELOCK, call=call)], chain_id=1, role_names_by_target=roles) + ) + self.assertIn(GRANT_SIGNATURE, flow) + self.assertNotIn("role **", flow) + + def test_failed_nested_decode_keeps_raw_payload(self) -> None: + """A decoding outage must neither abort the report nor discard the calldata.""" + call = decoded(scheduled(grant_data())) + with patch("utils.llm.report.decode_calldata", side_effect=RuntimeError("lookup unavailable")): + flow = format_call_flow(ReportContext(entries=[CallEntry(target=TIMELOCK, call=call)], chain_id=1)) + self.assertIn(grant_data(), flow) + with patch("utils.llm.ai_explainer.decode_calldata", side_effect=RuntimeError("lookup unavailable")): + self.assertEqual(_collect_role_names([(TIMELOCK, call)], 1), {}) + + def test_non_role_salts_arrays_and_tuples_are_not_annotated(self) -> None: + """A resolved zero admin role must not label salts or unrelated composite fields.""" + roles = {DEFAULT_ADMIN_ROLE: "DEFAULT_ADMIN_ROLE"} + call = DecodedCall( + "setConfig", + "setConfig(bytes32,bytes32[],(bytes32,address))", + [("bytes32", ZERO32), ("bytes32[]", (ZERO32,)), ("(bytes32,address)", (ZERO32, OTHER))], + ) + entries = [ + CallEntry( + target=CORE, + call=decoded(encode_call("grantRole(bytes32,address)", ["bytes32", "address"], [ZERO32, OTHER])), + ), + CallEntry(target=CORE, call=call), + CallEntry(target=CORE, call=decoded(scheduled(grant_data()))), + ] + flow = format_call_flow(ReportContext(entries=entries, chain_id=1, role_names_by_target={CORE.lower(): roles})) + self.assertEqual(flow.count("role **DEFAULT_ADMIN_ROLE**"), 1) + + def test_struct_and_nested_array_roles_are_resolved(self) -> None: + """Visit all bytes32 leaves while leaving malformed values unresolved.""" + for type_str, value in ( + ("(bytes32,address)[]", [(bytes.fromhex(MINTER[2:]), OTHER)]), + ("bytes32[][]", [[bytes.fromhex(MINTER[2:]), b"short"], []]), + ("(address,(bytes32[1],uint256))", (OTHER, ((bytes.fromhex(MINTER[2:]),), 1))), + ): + with self.subTest(type=type_str): + call = DecodedCall("setRoles", f"setRoles({type_str})", [(type_str, value)]) + roles = _collect_role_names([(CORE, call)], 1) + flow = format_call_flow( + ReportContext(entries=[CallEntry(target=CORE, call=call)], chain_id=1, role_names_by_target=roles) + ) + self.assertIn("role **RECEIPT_TOKEN_MINTER**", flow) + + def test_mixed_case_target_roles_are_merged_before_resolution(self) -> None: + """One source lookup must retain the union of roles for an address.""" + roles = _collect_role_names([(CORE, decoded(grant_data())), (CORE.lower(), decoded(grant_data(BURNER)))], 1) + self.assertEqual(roles[CORE.lower()], {MINTER: "RECEIPT_TOKEN_MINTER", BURNER: "RECEIPT_TOKEN_BURNER"}) + self.source.assert_called_once() + + def test_wrapper_depth_is_bounded(self) -> None: + """Adversarial wrapper nesting stops at the shared decoding depth limit.""" + data = grant_data() + for _ in range(MAX_BYTES_RECURSION_DEPTH + 1): + data = scheduled(data, TIMELOCK) + self.assertEqual(_collect_role_names([(TIMELOCK, decoded(data))], 1), {}) + self.source.assert_not_called() + + def test_decoded_wrapper_extraction_matches_raw_and_ignores_malformed(self) -> None: + """The raw and decoded wrapper paths preserve identical target/payload pairs.""" + data = execute_batch([CORE, OTHER], [grant_data(), grant_data(BURNER)]) + call = decoded(data) + self.assertEqual( + unwrap_call_values(call.signature, tuple(value for _, value in call.params)), unwrap_calls(data) + ) + self.assertEqual(unwrap_call_values(call.signature, ()), []) + self.assertEqual(unwrap_call_values("unknown(bytes)", (b"",)), []) + + @patch("utils.llm.ai_explainer._collect_address_labels", return_value={}) + @patch("utils.llm.ai_explainer._collect_related_tokens", return_value=[]) + @patch("utils.llm.ai_explainer.resolve_protocol_context", return_value=ResolvedProtocolContext()) + @patch("utils.llm.ai_explainer.get_llm_provider") + def test_public_single_and_batch_explainers_keep_nested_roles( + self, provider: MagicMock, _protocol: MagicMock, _tokens: MagicMock, _labels: MagicMock + ) -> None: + """Both entrypoints feed wrapped role names into the prompt and final report.""" + data = execute_batch([CORE], [grant_data()]) + for batch in (False, True): + with self.subTest(batch=batch): + mock_provider = make_provider("TLDR: grants a role. MEDIUM.\n\nDETAIL:\nPermission update.") + provider.return_value = mock_provider + result = ( + explain_batch_transaction( + calls=[{"target": TIMELOCK, "data": data, "value": "0"}], + chain_id=1, + skip_simulation=True, + refine=False, + ) + if batch + else explain_transaction( + target=TIMELOCK, + calldata=data, + chain_id=1, + skip_simulation=True, + refine=False, + ) + ) + assert result is not None + self.assertIn("is the role RECEIPT_TOKEN_MINTER", mock_provider.complete.call_args_list[0].args[0]) + self.assertIn("role **RECEIPT_TOKEN_MINTER**", result.report) diff --git a/utils/calldata/wrappers.py b/utils/calldata/wrappers.py index a6a735b5..4d93caac 100644 --- a/utils/calldata/wrappers.py +++ b/utils/calldata/wrappers.py @@ -163,11 +163,32 @@ def unwrap_calls(data_hex: str) -> list[InnerCall]: name, types, extractor = _WRAPPERS[data_hex[:10].lower()] try: values = decode(types, bytes.fromhex(data_hex[10:])) - pairs = extractor(values) + return _inner_calls(name, extractor(values)) except Exception as e: # noqa: BLE001 - adversarial/malformed payloads logger.debug("could not unwrap %s calldata: %s", name, e) return [] + +def unwrap_call_values(signature: str, values: tuple) -> list[InnerCall]: + """Extract target/payload pairs from already-decoded supported wrapper values. + + Scheduled calls are included for inspection, without claiming they execute + in the wrapping transaction. Unknown or malformed wrappers return no calls. + """ + extractor = _WRAPPER_SIGNATURES.get(signature) + if extractor is None: + return [] + name = signature.split("(")[0] + try: + pairs = extractor(values) + return _inner_calls(name, pairs) + except Exception as e: # noqa: BLE001 - best-effort decoding of malformed wrappers + logger.debug("could not unwrap %s values: %s", name, e) + return [] + + +def _inner_calls(name: str, pairs: list[tuple[str, bytes]]) -> list[InnerCall]: + """Normalize extracted wrapper calls with their original ordering.""" multiple = len(pairs) > 1 return [ InnerCall( diff --git a/utils/llm/ai_explainer.py b/utils/llm/ai_explainer.py index 71d3ed69..f4523859 100644 --- a/utils/llm/ai_explainer.py +++ b/utils/llm/ai_explainer.py @@ -18,6 +18,7 @@ from utils.cache import cache_path from utils.calldata.decoder import MAX_BYTES_RECURSION_DEPTH, DecodedCall, decode_calldata, try_decode_inner_calldata from utils.calldata.role_names import normalize_role_hash, resolve_role_names +from utils.calldata.wrappers import unwrap_call_values from utils.erc20_metadata import ERC20Metadata, fetch_erc20_metadata from utils.formatting import format_decimal_amount, normalize_token_amount, parse_wei from utils.impl_diff import diff_implementations, format_impl_diff @@ -34,6 +35,7 @@ checksum_or_none, format_address_links_block, iter_address_values, + iter_typed_values, ) from utils.logger import get_logger from utils.on_chain_state import StateRead, format_state_reads, read_before_state @@ -926,7 +928,7 @@ def _collect_role_names( targets_and_calls: list[tuple[str, DecodedCall]], chain_id: int, ) -> dict[str, dict[str, str]]: - """Resolve ``bytes32`` role arguments to names, keyed by lowercased target. + """Resolve nested role arguments and wrapped grants, keyed by lowercased target. Only calls whose function name mentions a role (``grantRole``, ``revokeRole``, ``setRoleAdmin``, …) are inspected. That heuristic matters: @@ -938,23 +940,50 @@ def _collect_role_names( and that source is already memoized by the source-context fetch, so this is effectively free. Unresolvable roles are deliberately omitted: the system prompt tells the model to call anything absent here unidentified. + + Supported timelock and Executor wrappers are walked with their explicit + targets. Scheduled payloads are inspected for role names too; this does + not treat their proposed grants as already executed. """ hashes_by_target: dict[str, set[str]] = {} - for target, decoded in targets_and_calls: + target_addresses: dict[str, str] = {} + pending = [(target, call, 0) for target, call in targets_and_calls] + while pending: + target, decoded, depth = pending.pop() + if depth < MAX_BYTES_RECURSION_DEPTH: + if decoded.script_calls: + pending.extend( + (inner.target, inner.decoded, depth + 1) + for inner in decoded.script_calls + if inner.decoded is not None + ) + else: + for inner in unwrap_call_values(decoded.signature, tuple(value for _, value in decoded.params)): + try: + inner_decoded = decode_calldata(inner.data, chain_id=chain_id, target=inner.target) + except Exception as error: # noqa: BLE001 - enrichment must not block the alert + logger.info("Role-call decoding failed for %s: %s", inner.target, error) + continue + if inner_decoded is not None: + pending.append((inner.target, inner_decoded, depth + 1)) if not target or "role" not in (decoded.function_name or "").lower(): continue + target_key = target.lower() + target_addresses.setdefault(target_key, target) for type_str, value in decoded.params: # `decode_calldata` keeps raw eth_abi output, so this is 32 raw bytes # for a real call. Normalize here rather than stringifying, which # would produce an unparseable Python repr. - if type_str == "bytes32" and (role_hash := normalize_role_hash(value)): - hashes_by_target.setdefault(target, set()).add(role_hash) + for role_value in iter_typed_values(type_str, value, "bytes32"): + if role_hash := normalize_role_hash(role_value): + hashes_by_target.setdefault(target_key, set()).add(role_hash) if not hashes_by_target: return {} def resolve(item: tuple[str, set[str]]) -> tuple[str, dict[str, str]]: - target, role_hashes = item + target_key, role_hashes = item + target = target_addresses[target_key] try: return target.lower(), resolve_role_names(sorted(role_hashes), chain_id=chain_id, target=target) except Exception as e: # noqa: BLE001 - enrichment only; never block an explanation @@ -2323,6 +2352,7 @@ def explain_transaction( context_note=context_note, description=description, safety_notes=safety_notes, + role_names_by_target=roles_by_target, ) try: @@ -2707,6 +2737,7 @@ def explain_batch_transaction( context_note=context_note, description=description, safety_notes=safety_notes, + role_names_by_target=roles_by_target, ) try: diff --git a/utils/llm/report.py b/utils/llm/report.py index 617b0f60..e725338a 100644 --- a/utils/llm/report.py +++ b/utils/llm/report.py @@ -23,13 +23,18 @@ MAX_BYTES_RECURSION_DEPTH, DecodedCall, ScriptCall, + decode_calldata, split_top_level_types, try_decode_inner_calldata, ) from utils.calldata.role_names import normalize_role_hash +from utils.calldata.wrappers import InnerCall, unwrap_call_values from utils.chains import EXPLORER_URLS, Chain +from utils.logger import get_logger from utils.related_tokens import RelatedToken +logger = get_logger("utils.llm.report") + ZERO_ADDRESS = "0x0000000000000000000000000000000000000000" # The report already opens the section with "## Analysis", so a detail that @@ -173,6 +178,8 @@ class ReportContext: description: str = "" # Seatbelt-style checks (unverified target, ETH to nonpayable). safety_notes: list[str] = field(default_factory=list) + # Nested calls must use the names resolved for their own target. + role_names_by_target: dict[str, dict[str, str]] = field(default_factory=dict) @dataclass @@ -277,8 +284,13 @@ def _format_param_value( chain_id: int, labels: dict[str, str], token: RelatedToken | None = None, + role_names: dict[str, str] | None = None, ) -> str: """Render a single scalar parameter value for the markdown call flow.""" + if role_names and type_str == "bytes32": + role_hash = normalize_role_hash(value) + if role := role_names.get(role_hash): + return f"`{role_hash}` — role **{role}**" if type_str == "address" and isinstance(value, str): return address_link(value, chain_id, labels) if isinstance(value, bytes): @@ -315,28 +327,29 @@ def _is_composite(type_str: str) -> bool: return array_element_type(type_str) is not None or tuple_component_types(type_str) is not None -def iter_address_values(type_str: str, value: object) -> Iterator[str]: - """Yield every ``address`` leaf inside a decoded parameter value. - - Walks arrays and tuples (and their nesting) so a struct argument like - ``(address,address,uint256)`` contributes its addresses to label lookup - and to the prompt's Address Links section — without this they'd only ever - be stringified into the report as raw, unlinked text. - """ +def iter_typed_values(type_str: str, value: object, leaf_type: str) -> Iterator[object]: + """Yield leaves of one ABI type through arrays and tuples of any nesting.""" element = array_element_type(type_str) if element is not None and isinstance(value, (list, tuple)): for item in value: - yield from iter_address_values(element, item) + yield from iter_typed_values(element, item, leaf_type) return components = tuple_component_types(type_str) if components is not None and isinstance(value, (list, tuple)) and len(components) == len(value): for component, item in zip(components, value): - yield from iter_address_values(component, item) + yield from iter_typed_values(component, item, leaf_type) return - if type_str == "address" and isinstance(value, str): + if type_str == leaf_type: yield value +def iter_address_values(type_str: str, value: object) -> Iterator[str]: + """Yield string address leaves from decoded arrays, tuples, or scalars.""" + for address in iter_typed_values(type_str, value, "address"): + if isinstance(address, str): + yield address + + def _render_param( label: str, type_str: str, @@ -346,6 +359,9 @@ def _render_param( indent: str, depth: int = 0, token: "RelatedToken | None" = None, + role_names: dict[str, str] | None = None, + nested_calls: Iterator[InnerCall] | None = None, + role_names_by_target: dict[str, dict[str, str]] | None = None, ) -> list[str]: """Render one parameter, expanding arrays and tuples into nested bullets. @@ -361,11 +377,25 @@ def _render_param( return [f"{indent}- {label}: _(empty)_"] lines = [f"{indent}- {label}:"] for i, item in enumerate(value): - if _is_composite(element): + if _is_composite(element) or element == "bytes": # Nested tuples/structs must not inherit a parent amount token. - lines.extend(_render_param(f"`[{i}]`", element, item, chain_id, labels, indent + " ", depth, None)) + lines.extend( + _render_param( + f"`[{i}]`", + element, + item, + chain_id, + labels, + indent + " ", + depth, + None, + role_names, + nested_calls, + role_names_by_target, + ) + ) else: - lines.append(f"{indent} - {_format_param_value(element, item, chain_id, labels, token)}") + lines.append(f"{indent} - {_format_param_value(element, item, chain_id, labels, token, role_names)}") return lines components = tuple_component_types(type_str) @@ -374,17 +404,52 @@ def _render_param( return [f"{indent}- {label}: _(empty)_"] lines = [f"{indent}- {label}:"] for component, item in zip(components, value): - lines.extend(_render_param(f"`{component}`", component, item, chain_id, labels, indent + " ", depth, None)) + lines.extend( + _render_param( + f"`{component}`", + component, + item, + chain_id, + labels, + indent + " ", + depth, + None, + role_names, + nested_calls, + role_names_by_target, + ) + ) return lines if type_str == "bytes" and depth < MAX_BYTES_RECURSION_DEPTH: - inner = try_decode_inner_calldata(value) + nested = next(nested_calls, None) if nested_calls is not None else None + try: + inner = ( + decode_calldata(nested.data, chain_id=chain_id, target=nested.target) + if nested is not None + else try_decode_inner_calldata(value) + ) + except Exception as error: # noqa: BLE001 - preserve raw payload when enrichment fails + logger.info("Nested report calldata decoding failed: %s", error) + inner = None if inner is not None: lines = [f"{indent}- {label}: ↳ `{inner.signature}`"] - lines.extend(_format_params(inner, chain_id, labels, None, indent + " ", depth + 1)) + inner_roles = (role_names_by_target or {}).get(nested.target.lower(), {}) if nested is not None else {} + lines.extend( + _format_params( + inner, + chain_id, + labels, + None, + indent + " ", + depth + 1, + role_names=inner_roles, + role_names_by_target=role_names_by_target, + ) + ) return lines - return [f"{indent}- {label}: {_format_param_value(type_str, value, chain_id, labels, token)}"] + return [f"{indent}- {label}: {_format_param_value(type_str, value, chain_id, labels, token, role_names)}"] def _format_params( @@ -396,28 +461,43 @@ def _format_params( depth: int = 0, token: "RelatedToken | None" = None, role_names: dict[str, str] | None = None, + role_names_by_target: dict[str, dict[str, str]] | None = None, ) -> list[str]: """Render a call's parameters as an indented markdown bullet list. - A ``bytes32`` that ``role_names`` resolves is annotated with its role name. - Scalars render to exactly one line, so the annotation is appended to the - line just produced; composites are left alone. + A ``bytes32`` that ``role_names`` resolves is annotated with its role name, + including each element of a role array. """ lines: list[str] = [] + # A role resolved elsewhere on this target does not make a salt/id a role. + if "role" not in (call.function_name or "").lower(): + role_names = None + nested_calls = iter(unwrap_call_values(call.signature, tuple(value for _, value in call.params))) for i, (type_str, value) in enumerate(call.params): name = param_names[i] if param_names is not None and i < len(param_names) else None hint_token = token if is_token_amount_param(call.signature, i, name, token) else None if call.script_calls and type_str == "bytes": - lines.extend(_render_script_calls(call, _param_label(type_str, name), chain_id, labels, indent, depth)) + lines.extend( + _render_script_calls( + call, _param_label(type_str, name), chain_id, labels, indent, depth, role_names_by_target + ) + ) continue - rendered = _render_param( - _param_label(type_str, name), type_str, value, chain_id, labels, indent, depth, hint_token + lines.extend( + _render_param( + _param_label(type_str, name), + type_str, + value, + chain_id, + labels, + indent, + depth, + hint_token, + role_names, + nested_calls, + role_names_by_target, + ) ) - if role_names and type_str == "bytes32" and len(rendered) == 1: - role = role_names.get(normalize_role_hash(value)) - if role: - rendered = [f"{rendered[0]} — role **{role}**"] - lines.extend(rendered) return lines @@ -439,6 +519,7 @@ def _render_script_calls( labels: dict[str, str], indent: str, depth: int, + role_names_by_target: dict[str, dict[str, str]] | None = None, ) -> list[str]: """Render a Yearn Executor script as its inner calls instead of packed hex.""" count = len(call.script_calls) @@ -448,7 +529,18 @@ def _render_script_calls( target = address_link(script_call.target, chain_id, labels) lines.append(f"{indent} - Script call {k}: {_script_call_what(script_call)} on {target}") if script_call.decoded is not None and depth < MAX_BYTES_RECURSION_DEPTH: - lines.extend(_format_params(script_call.decoded, chain_id, labels, None, indent + " ", depth + 1)) + lines.extend( + _format_params( + script_call.decoded, + chain_id, + labels, + None, + indent + " ", + depth + 1, + role_names=(role_names_by_target or {}).get(script_call.target.lower(), {}), + role_names_by_target=role_names_by_target, + ) + ) return lines @@ -491,7 +583,8 @@ def format_call_flow(ctx: ReportContext) -> str: entry.param_names, indent=" ", token=entry.amount_token, - role_names=entry.role_names, + role_names=ctx.role_names_by_target.get(entry.target.lower(), entry.role_names), + role_names_by_target=ctx.role_names_by_target, ) lines.extend(param_lines or [" - _no inputs_"]) if entry.simulation_note: