diff --git a/certora_autosetup/autosetup/cli.py b/certora_autosetup/autosetup/cli.py index a6b7f068..e05c470b 100644 --- a/certora_autosetup/autosetup/cli.py +++ b/certora_autosetup/autosetup/cli.py @@ -15,6 +15,7 @@ from certora_autosetup.autosetup.cli_args import create_parser from certora_autosetup.autosetup.types import AutosetupConfig from certora_autosetup.cache.cache_fs import get_fs, init_cache_fs +from certora_autosetup.harnesser.swap import swap_library_main_contract, with_harnessed_library from certora_autosetup.cache.content_cache import ContentCache from certora_autosetup.reporting.reporter import Reporter from certora_autosetup.setup.sanity_rule_generator import SanityRuleGenerator @@ -107,6 +108,26 @@ def main(): ) contract_handles = with_contract_handle(contract_handles, main_contract_handle) + # A library cannot be a verification target — the Prover accepts it and instantiates + # no parametric methods. Swap in a generated harness before anything downstream keys + # on the contract name (sanity spec, base conf, verify target, result keys). + harnessed_library = main_contract_handle + main_contract_handle, contract_handles, _library_harness = swap_library_main_contract( + project_root=cwd, + main_contract_handle=main_contract_handle, + contract_handles=contract_handles, + solc=args.solc_default, + certora_run_command=args.certora_run_command, + ) + # AutoProver's pipeline swaps before it invokes us, so on that path the target + # arrives already harnessed and only the manifest still names the library. + args.additional_contracts = with_harnessed_library( + cwd, + main_contract_handle, + args.additional_contracts or [], + swapped_from=harnessed_library if _library_harness is not None else None, + ) + # TODO: a bare `path.sol` spec drops only the contract whose name matches the file # stem. Expand to "drop every concrete contract in the file" for symmetry with # auto-detect's emit-all default. Mirror the same expansion for include specs diff --git a/certora_autosetup/harnesser/__init__.py b/certora_autosetup/harnesser/__init__.py new file mode 100644 index 00000000..0b5821a1 --- /dev/null +++ b/certora_autosetup/harnesser/__init__.py @@ -0,0 +1,12 @@ +"""Generate a verifiable contract harness for a library main contract.""" + +from certora_autosetup.harnesser.detect import contract_kind, is_library_main_contract +from certora_autosetup.harnesser.model import HarnessPlan, LibraryApi, LibraryHarnessError + +__all__ = [ + "contract_kind", + "is_library_main_contract", + "HarnessPlan", + "LibraryApi", + "LibraryHarnessError", +] diff --git a/certora_autosetup/harnesser/__main__.py b/certora_autosetup/harnesser/__main__.py new file mode 100644 index 00000000..efa4069a --- /dev/null +++ b/certora_autosetup/harnesser/__main__.py @@ -0,0 +1,3 @@ +from certora_autosetup.harnesser.cli import main + +raise SystemExit(main()) diff --git a/certora_autosetup/harnesser/cli.py b/certora_autosetup/harnesser/cli.py new file mode 100644 index 00000000..d5fe6789 --- /dev/null +++ b/certora_autosetup/harnesser/cli.py @@ -0,0 +1,103 @@ +"""``python -m certora_autosetup.harnesser`` — generate a library harness. + + python -m certora_autosetup.harnesser --library src/utils/BitMaps.sol:BitMaps \ + --project-dir . --output harness.json + +It compiles a probe build to learn the library's API, writes +``certora/harnesses/CertoraLibraryHarness_.sol``, and records what it wrapped. +The JSON record goes to the file named by ``--output`` rather than to stdout, because +the probe build and the logger both write there; whoever runs this decides what to do +with the harness, so nothing here swaps a main contract. +""" + +import argparse +import json +import sys +from pathlib import Path + +from certora_autosetup.harnesser.model import LibraryHarnessError +from certora_autosetup.harnesser.run import ensure_library_harness +from certora_autosetup.utils.contract_utils import parse_contract_files, split_contract_spec +from certora_autosetup.utils.types import ContractHandle + + +def _project_relative(handle: ContractHandle, root: Path) -> ContractHandle: + """Probe-build file arguments are resolved from the project root, so keep them there. + + ``parse_contract_files`` absolutizes against the root in order to check the file + exists; the build wants the path back the way the user wrote it. + """ + path = Path(handle.source_file) + if path.is_absolute() and path.is_relative_to(root): + path = path.relative_to(root) + return ContractHandle(contract_name=handle.contract_name, source_file=path.as_posix()) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="certora_autosetup.harnesser", + description="Generate a verifiable contract harness for a library main contract.", + ) + parser.add_argument( + "--library", + required=True, + help="Library to wrap, as path/To/Lib.sol:LibName (name defaults to the file stem)", + ) + parser.add_argument( + "--project-dir", default=".", help="Project root; defaults to the current directory" + ) + parser.add_argument("--solc", default=None, help="solc to build with, e.g. solc8.16") + parser.add_argument( + "--extra-file", + action="append", + default=[], + dest="extra_files", + help="Additional path:Contract to include in the probe build; repeatable", + ) + parser.add_argument( + "--skip-validation", + action="store_true", + help="Do not recompile the filled harness (faster; leaves compile errors to autosetup)", + ) + parser.add_argument( + "--output", + default=None, + help="Write the JSON result here; without it only the human summary is printed", + ) + args = parser.parse_args(argv) + + library_path, library_name = split_contract_spec(args.library) + project_root = Path(args.project_dir).resolve() + + try: + # Parsed rather than passed through, so a mistyped --extra-file is reported here + # instead of as a probe-build failure minutes later. + extra_files = [ + _project_relative(handle, project_root) + for handle in parse_contract_files(args.extra_files, project_root) + ] if args.extra_files else [] + result = ensure_library_harness( + project_root=project_root, + library=ContractHandle(contract_name=library_name, source_file=library_path), + solc=args.solc, + extra_files=extra_files, + validate=not args.skip_validation, + ) + except (LibraryHarnessError, ValueError) as e: + print(f"library harness generation failed: {e}", file=sys.stderr) + return 1 + + if args.output: + Path(args.output).write_text(json.dumps(result.to_dict(), indent=2) + "\n") + + coverage = result.coverage + print( + f"{result.harness.contract_name} -> {result.harness.source_file}: " + f"{coverage['wrapped']}/{coverage['total']} function(s) wrapped, " + f"{coverage['readers']} storage reader(s), {coverage['skipped']} skipped" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/certora_autosetup/harnesser/detect.py b/certora_autosetup/harnesser/detect.py new file mode 100644 index 00000000..e3f221fe --- /dev/null +++ b/certora_autosetup/harnesser/detect.py @@ -0,0 +1,150 @@ +"""Decide whether a main contract is a ``library``, before anything is built. + +The Certora Prover does not reject a library verification target — it accepts it and +silently verifies nothing, because parametric rules filter libraries out of the method +set. There is therefore no error message to key on: detection has to read the +declaration itself. + +It reads it from solc's own AST rather than from the source text. ``solc +--standard-json`` with ``stopAfter: "parsing"`` returns ``ContractDefinition`` nodes +carrying ``contractKind`` without resolving imports, type-checking, or generating code, +so a single unresolved-import-laden library file parses in milliseconds with no build +and no dependency setup. + +``stopAfter`` requires solc >= 0.7. Below that, solc refuses to emit an AST for a file +whose imports it cannot resolve, which is every real library file, and pre-build +detection is not possible; such a project keeps today's behavior and logs why. + +Reading the declarations out of that AST is shared with the post-build dump path, in +``solidity_ast.contracts``. +""" + +import json +import shutil +import subprocess +from pathlib import Path +from typing import Dict, Optional + +from packaging.version import Version + +from certora_autosetup.solidity_ast import parse_only_declarations +from certora_autosetup.utils.logger import logger +from certora_autosetup.utils.solc_version_resolver import ( + convert_solc_version_to_certora_format, + read_pragma_from_source_file, + resolve_pragma_to_version, +) +from certora_autosetup.utils.types import ContractKind + +#: Below this, solc has no ``stopAfter`` and cannot parse a file with unresolved imports. +MIN_SOLC_FOR_PARSE_ONLY = Version("0.7.0") + + +def _solc_binary(version: str) -> Optional[str]: + """Locate an installed solc binary for ``version`` under either naming convention.""" + for name in (convert_solc_version_to_certora_format(version), f"solc-{version}"): + if path := shutil.which(name): + return path + return None + + +def _parse_only_ast(solc: str, source_file: Path, content: str) -> Optional[Dict]: + """Parse a single file and return its AST node dict, or None if solc could not. + + Imports are deliberately left unresolved: ``stopAfter: "parsing"`` never follows + them, so the ``sources`` map holds exactly one entry. + """ + request = { + "language": "Solidity", + "sources": {source_file.name: {"content": content}}, + "settings": { + "stopAfter": "parsing", + "outputSelection": {"*": {"": ["ast"]}}, + }, + } + try: + completed = subprocess.run( + [solc, "--standard-json"], + input=json.dumps(request), + capture_output=True, + text=True, + timeout=60, + ) + response = json.loads(completed.stdout) + except (subprocess.SubprocessError, json.JSONDecodeError, OSError) as e: + logger.log(f"solc parse-only probe failed for {source_file}: {e}", "DEBUG", "Harnesser") + return None + + for error in response.get("errors", []): + if error.get("severity") == "error": + logger.log( + f"solc could not parse {source_file}: {error.get('message', '')}", + "DEBUG", + "Harnesser", + ) + return None + + sources = response.get("sources", {}) + entry = sources.get(source_file.name) or next(iter(sources.values()), None) + return entry.get("ast") if entry else None + + +def contract_kind( + source_file: Path, + contract_name: str, + project_root: Optional[Path] = None, + preferred_solc: Optional[str] = None, +) -> Optional[ContractKind]: + """Return the declared kind of ``contract_name``. + + None means the question could not be answered (no usable solc, unparseable file, or + the name is not declared here); callers treat that as "not a library" and proceed + unchanged, which is the behavior that predates this feature. + """ + if not source_file.exists(): + return None + + content = source_file.read_text(errors="replace") + pragma = read_pragma_from_source_file(source_file, project_root) + version = resolve_pragma_to_version(pragma, preferred_solc) if pragma else preferred_solc + if not version: + logger.log( + f"No solc version resolvable for {source_file}; skipping library detection", + "DEBUG", + "Harnesser", + ) + return None + + if Version(version) < MIN_SOLC_FOR_PARSE_ONLY: + logger.log( + f"{source_file} resolves to solc {version}; parse-only AST needs >= " + f"{MIN_SOLC_FOR_PARSE_ONLY}, so a library main contract cannot be detected " + f"before the build", + "WARNING", + "Harnesser", + ) + return None + + solc = _solc_binary(version) + if not solc: + logger.log(f"No installed solc binary for {version}", "DEBUG", "Harnesser") + return None + + ast = _parse_only_ast(solc, source_file, content) + if not ast: + return None + + for decl in parse_only_declarations(ast, str(source_file)): + if decl.name == contract_name: + return decl.contract_kind + return None + + +def is_library_main_contract( + source_file: Path, + contract_name: str, + project_root: Optional[Path] = None, + preferred_solc: Optional[str] = None, +) -> bool: + """Whether verifying ``contract_name`` requires a generated harness.""" + return contract_kind(source_file, contract_name, project_root, preferred_solc) is ContractKind.LIBRARY diff --git a/certora_autosetup/harnesser/model.py b/certora_autosetup/harnesser/model.py new file mode 100644 index 00000000..6068b2db --- /dev/null +++ b/certora_autosetup/harnesser/model.py @@ -0,0 +1,216 @@ +"""Data model for the library harnesser. + +A ``library`` cannot be the Certora Prover's verification target: parametric rules +filter libraries out of the method set, and CVL rejects direct library calls from a +spec. The harnesser generates a plain contract that calls the library, so the library's +bodies get inlined into a verifiable target. + +The flow is ``LibraryApi`` (what the build says the library exposes) → ``HarnessPlan`` +(what we will emit, decided) → Solidity text. ``LibraryApi`` mirrors the build JSON; +``HarnessPlan`` is fully resolved, so rendering is a pure formatting step with no +decisions left in it. +""" + +from dataclasses import dataclass, field +from enum import Enum +from typing import Dict, Optional + +from certora_autosetup.utils.types import ContractHandle + + +class SkipReason(Enum): + """Why a library function got no wrapper. + + Every skip is reported to the user: a library whose interesting half is skipped + yields a harness that verifies nothing, and that has to be visible rather than + silently counted as coverage. + """ + + PRIVATE = "private" + STORAGE_POINTER_RETURN = "storage_pointer_return" + INTERNAL_ONLY_TYPE = "internal_only_type" + UNRESOLVED_TYPE = "unresolved_type" + OPAQUE_HANDLE = "opaque_handle" + ASSEMBLY_TERMINATOR = "assembly_terminator" + CONSTRUCTOR = "constructor" + + +#: Data locations as they appear in the build JSON's per-argument ``location`` field. +LOC_STORAGE = "storage" +LOC_MEMORY = "memory" +LOC_CALLDATA = "calldata" + + +@dataclass(frozen=True) +class LibParam: + """One parameter or return value of a library function. + + ``solidity_type`` is already rendered source text (via ``parse_type_descriptor`` in + SOLIDITY mode), so it carries the qualification the harness needs — e.g. + ``EnumerableSet.Bytes32Set``, not a bare ``Bytes32Set``. + """ + + name: str + solidity_type: str + location: str = "" + #: Raw ``typeDesc.type`` from the build. Retained because some kinds have no Solidity + #: rendering at all — a ``Function`` type is internal-only and renders as nothing — + #: and the reason a function was skipped should say which case it was. + desc_kind: str = "" + + @property + def is_storage(self) -> bool: + return self.location == LOC_STORAGE + + @property + def is_reference(self) -> bool: + return self.location in (LOC_STORAGE, LOC_MEMORY, LOC_CALLDATA) + + +#: How a struct member is shaped, which decides how a reader reaches through it. +KIND_VALUE = "value" +KIND_STRUCT = "struct" +KIND_MAPPING = "mapping" +KIND_ARRAY = "array" + + +@dataclass(frozen=True) +class MemberNode: + """One node of an owned struct's member tree. + + Kept as a tree rather than a flat type string because storage readers are derived + by walking it: a mapping contributes a key parameter, an array an index parameter, + a nested struct just a longer access path. OpenZeppelin's ``_indexOf`` — the reader + its own spec needs — is the leaf of ``AddressSet -> _inner -> _indexes[key]``. + """ + + name: str + solidity_type: str + kind: str + key_type: str = "" + value: Optional["MemberNode"] = None + members: tuple["MemberNode", ...] = () + + +@dataclass(frozen=True) +class LibFunction: + """One function declared by the library, as the build reports it.""" + + name: str + visibility: str + state_mutability: str + params: tuple[LibParam, ...] + returns: tuple[LibParam, ...] + source_line: int = 0 + + @property + def storage_params(self) -> tuple[LibParam, ...]: + return tuple(p for p in self.params if p.is_storage) + + +@dataclass(frozen=True) +class LibraryApi: + """Everything the harnesser knows about the library it is wrapping. + + ``contract.source_file`` is the path the build reported, which is what the harness + imports and what disambiguates same-named libraries (solady ships 17 library names + twice, under ``src/utils/`` and ``src/utils/g/``). + """ + + contract: ContractHandle + functions: tuple[LibFunction, ...] + #: Qualified struct type (e.g. "EnumerableSet.AddressSet") -> its member tree, as + #: the build reports it. Storage readers are derived from this; member names are + #: never hardcoded because they change across library versions (OpenZeppelin + #: renamed EnumerableSet's ``_indexes`` to ``_positions`` in v5). + struct_members: Dict[str, tuple[MemberNode, ...]] = field(default_factory=dict) + + +@dataclass(frozen=True) +class OwnedVar: + """A state variable the harness declares so it can supply a ``storage`` argument. + + A library function taking ``EnumerableSet.AddressSet storage`` has no callable + external form: the caller cannot construct a storage pointer. The harness owns one + instance per distinct storage type and binds it, which is also what makes the + wrappers stateful enough to state invariants over. + """ + + var_name: str + solidity_type: str + + +@dataclass(frozen=True) +class Wrapper: + """One public function the harness will expose.""" + + #: Name after CVL-keyword renaming and ABI-collision mangling. + name: str + library_function: str + params: tuple[LibParam, ...] + returns: tuple[LibParam, ...] + state_mutability: str + #: Positional call arguments, already resolved: either a wrapper parameter name or + #: an owned state variable name. + call_args: tuple[str, ...] + #: Set when the library function mutates a memory reference in place and returns + #: nothing; the wrapper returns the mutated argument so the effect is observable + #: across the ABI boundary. + returns_mutated_param: Optional[str] = None + + +@dataclass(frozen=True) +class StorageReader: + """A getter over a member of an owned struct. + + Properties worth verifying are usually about the library's internal + representation — OpenZeppelin's own EnumerableSet spec needs ``_indexOf`` to relate + ``at(i)`` back to the index. Without readers the harness only echoes its own API. + """ + + name: str + solidity_type: str + access_expression: str + params: tuple[LibParam, ...] = () + location: str = "" + + +@dataclass(frozen=True) +class Skipped: + """A library function that got no wrapper, and why.""" + + library_function: str + reason: SkipReason + detail: str = "" + + +@dataclass(frozen=True) +class HarnessPlan: + """The fully-resolved decision of what the harness file contains.""" + + harness: ContractHandle + library: ContractHandle + pragma_line: str + import_lines: tuple[str, ...] + owned_vars: tuple[OwnedVar, ...] + wrappers: tuple[Wrapper, ...] + readers: tuple[StorageReader, ...] + skipped: tuple[Skipped, ...] + + @property + def coverage(self) -> Dict[str, int]: + wrapped = len(self.wrappers) + return { + "total": wrapped + len(self.skipped), + "wrapped": wrapped, + "skipped": len(self.skipped), + "readers": len(self.readers), + } + + +class LibraryHarnessError(Exception): + """The harness could not be generated. + + Raised rather than degraded: a harness missing the wrappers the user cares about + still runs, still reports "verified", and proves nothing. + """ diff --git a/certora_autosetup/harnesser/plan.py b/certora_autosetup/harnesser/plan.py new file mode 100644 index 00000000..24d5d610 --- /dev/null +++ b/certora_autosetup/harnesser/plan.py @@ -0,0 +1,517 @@ +"""Decide what the harness contains: classify, own, mangle, order. + +Pure functions over a ``LibraryApi``. Every decision is made here so that rendering is +plain formatting and the whole thing is testable without solc, disk or network. + +The shape of the problem: a library function is written for an internal caller, and the +harness has to re-expose it across an ABI boundary. Three things do not survive that +crossing, and each has one right answer: + +- A ``storage`` parameter cannot be passed in — an external caller has no way to build a + storage pointer. The harness owns an instance of that type and binds it, which is also + what gives the harness state to state invariants over. +- Some types cannot appear in a public signature at all (a struct containing a mapping, + an internal function type). Those functions are skipped and reported. +- A function that mutates a ``memory`` argument in place and returns nothing is the + identity function once it is called externally, because the callee mutates a fresh + copy. The wrapper returns the mutated argument so the effect is observable. + +Some library functions compile perfectly as wrappers and are still wrong to emit; those +are skipped deliberately, see ``_opaque_handle_reason``. +""" + +import re +from collections import defaultdict +from typing import Dict, List, Mapping, Optional, Sequence, Tuple + +from certora_autosetup.utils.cvl_keywords import escape_reserved +from certora_autosetup.utils.types import ContractHandle +from certora_autosetup.harnesser.model import ( + KIND_ARRAY, + KIND_MAPPING, + KIND_STRUCT, + KIND_VALUE, + LOC_MEMORY, + MemberNode, + HarnessPlan, + LibFunction, + LibParam, + LibraryApi, + LibraryHarnessError, + OwnedVar, + Skipped, + SkipReason, + StorageReader, + Wrapper, +) + +#: Turns any type expression into an identifier fragment usable in a function name. +#: Needed because a mangling suffix may be derived from a raw mapping type +#: (``mapping(uint256 => uint256)``), which is a real storage receiver in Solady's LibMap. +_NON_IDENTIFIER = re.compile(r"[^A-Za-z0-9]+") + +#: A parameter type solc will not accept in a public/external signature. ``function`` +#: types are internal-only; a mapping cannot cross the ABI boundary in any position. +_INTERNAL_ONLY_MARKERS = ("function(", "mapping(") + +#: ``typeDesc.type`` of an internal function type, which has no Solidity rendering. +_FUNCTION_TYPE_KIND = "Function" + +#: Bounds on reader derivation. Depth stops a self-referential struct; the count keeps a +#: deeply nested representation from burying the library's own API in getters. +_MAX_READER_DEPTH = 4 +_MAX_READERS = 24 + + +def sanitize_identifier(text: str) -> str: + """Collapse a type expression into a name fragment.""" + return _NON_IDENTIFIER.sub("_", text).strip("_") + + +def _is_internal_only(solidity_type: str) -> bool: + return any(marker in solidity_type for marker in _INTERNAL_ONLY_MARKERS) + + +def _node_contains_mapping(node: MemberNode, depth: int = 0) -> bool: + if depth > 8 or node.kind == KIND_MAPPING: + return True + if node.kind == KIND_STRUCT: + return any(_node_contains_mapping(child, depth + 1) for child in node.members) + if node.kind == KIND_ARRAY and node.value is not None: + return _node_contains_mapping(node.value, depth + 1) + return False + + +def _contains_mapping( + solidity_type: str, struct_members: Mapping[str, tuple[MemberNode, ...]] +) -> bool: + """Whether a type transitively contains a mapping, and so cannot cross the ABI. + + Struct members are expanded recursively: OpenZeppelin's ``Bytes32ToBytes32Map`` + holds an ``EnumerableSet.Bytes32Set``, which holds a mapping, so a getter returning + the outer struct is rejected by solc even though its own members look harmless. + """ + if "mapping(" in solidity_type: + return True + base = solidity_type.replace("[]", "").strip() + members = struct_members.get(base) + if not members: + return False + return any(_node_contains_mapping(m) for m in members) + + +def _opaque_handle_reason(fn: LibFunction) -> Optional[str]: + """Whether this function traffics in pointers that are meaningless across a call. + + Solady's RedBlackTreeLib returns a ``bytes32 ptr`` that encodes a storage offset, and + JSONParserLib passes an ``Item`` whose single ``uint256`` is a memory address. Wrapped + naively they compile, and then the Prover is free to invent a pointer value — which + produces counterexamples that cannot happen in reality. A skipped function is a + visible gap; an unsound one is a false bug report to the user. + """ + for param in fn.params: + if param.name in ("ptr", "pointer") and param.solidity_type == "bytes32": + return f"parameter '{param.name}' is a library-internal pointer" + return None + + +def _classify( + fn: LibFunction, struct_members: Mapping[str, tuple[MemberNode, ...]] +) -> Optional[Skipped]: + """Return why ``fn`` gets no wrapper, or None if it is wrappable. First match wins.""" + if fn.visibility == "private": + # Private functions are unreachable from outside the library by construction. + # No semantic loss in these corpora: the internal functions that call them + # expose the same behavior. + return Skipped(fn.name, SkipReason.PRIVATE) + + for param in (*fn.params, *fn.returns): + if param.solidity_type: + continue + if param.desc_kind == _FUNCTION_TYPE_KIND: + # An internal function type has no external form at all. OpenZeppelin's + # Checkpoints.push takes a comparator this way; it is the single function + # hardhat-exposed also declines to expose. + return Skipped( + fn.name, SkipReason.INTERNAL_ONLY_TYPE, f"'{param.name}' is an internal function type" + ) + return Skipped(fn.name, SkipReason.UNRESOLVED_TYPE, f"parameter '{param.name}'") + + for param in fn.params: + if param.is_storage: + continue + if _is_internal_only(param.solidity_type) or _contains_mapping( + param.solidity_type, struct_members + ): + return Skipped( + fn.name, SkipReason.INTERNAL_ONLY_TYPE, f"{param.solidity_type} {param.name}" + ) + + for ret in fn.returns: + if ret.is_storage: + # Copying a storage pointer to memory compiles and silently drops the write + # path, so callers could read the slot but never assign to it. The whole of + # OpenZeppelin's StorageSlot is this shape. + return Skipped( + fn.name, SkipReason.STORAGE_POINTER_RETURN, f"returns {ret.solidity_type} storage" + ) + if _is_internal_only(ret.solidity_type) or _contains_mapping( + ret.solidity_type, struct_members + ): + return Skipped(fn.name, SkipReason.INTERNAL_ONLY_TYPE, f"returns {ret.solidity_type}") + + if reason := _opaque_handle_reason(fn): + return Skipped(fn.name, SkipReason.OPAQUE_HANDLE, reason) + + return None + + +def _external_location(param: LibParam) -> str: + """The data location this parameter takes in the wrapper's signature. + + Reference parameters stay ``memory``: ``calldata`` would compile but leaves nothing + to return for an in-place mutator, and the ABI encoding is identical either way. + """ + return LOC_MEMORY if param.is_reference else "" + + +def _type_location(solidity_type: str) -> str: + """The data location a bare type name needs to cross the ABI boundary. + + ``_external_location`` reads the location off the parameter, which works for a + library function's own arguments. A reader's key and leaf types come from struct + members instead, and a member declaration carries no location, so the type itself + has to say whether one is required. + """ + if solidity_type in ("bytes", "string") or solidity_type.endswith("]"): + return LOC_MEMORY + return "" + + +def short_type_name(solidity_type: str, library_name: str) -> str: + """Identifier fragment for a type, without the wrapped library's own qualifier. + + Every storage receiver belongs to the library being wrapped, so repeating its name in + each generated identifier only makes them long: ``at_Bytes32Set`` says as much as + ``at_EnumerableSet_Bytes32Set`` and reads like the hand-written harnesses it sits + beside. A type from elsewhere keeps its qualifier, which is what still tells it apart. + """ + unqualified = solidity_type + prefix = f"{library_name}." + if unqualified.startswith(prefix): + unqualified = unqualified[len(prefix):] + return sanitize_identifier(unqualified) + + +def _owned_var_name(solidity_type: str, library_name: str) -> str: + """Name the harness state variable that backs a given storage type.""" + return f"_certoraStore_{short_type_name(solidity_type, library_name)}" + + +def _mutated_memory_param(fn: LibFunction) -> Optional[LibParam]: + """The memory argument a void function mutates in place, if that is its whole effect. + + Solady's LibSort is 16 such functions (``sort``, ``reverse``, ``uniquifySorted``, + ``insertionSort``): ``internal pure``, no return, mutating the array they are given. + Called across the ABI the callee mutates a decoded copy, so without a synthesized + return the wrapper is observationally the identity function. + """ + if fn.returns: + return None + memory_refs = [p for p in fn.params if p.location == LOC_MEMORY] + if len(memory_refs) != 1: + return None + return memory_refs[0] + + +def _wrapper_mutability(fn: LibFunction, binds_state: bool) -> str: + """Mutability of the wrapper, widened when it reads harness state. + + A ``pure`` library function becomes ``view`` once the harness feeds it a state + variable, because the wrapper now reads storage the library function did not. + """ + if binds_state and fn.state_mutability == "pure": + return "view" + return fn.state_mutability + + +def _build_wrapper( + fn: LibFunction, owned: Dict[str, OwnedVar] +) -> Wrapper: + """Turn a wrappable library function into an exposed harness function.""" + exposed: List[LibParam] = [] + call_args: List[str] = [] + binds_state = False + + for param in fn.params: + if param.is_storage: + var = owned[param.solidity_type] + call_args.append(var.var_name) + binds_state = True + continue + exposed.append( + LibParam( + name=param.name, + solidity_type=param.solidity_type, + location=_external_location(param), + ) + ) + call_args.append(param.name) + + returns = tuple( + LibParam( + name=ret.name, + solidity_type=ret.solidity_type, + location=_external_location(ret), + ) + for ret in fn.returns + ) + + mutated = _mutated_memory_param(fn) + if mutated is not None: + returns = ( + LibParam(name="", solidity_type=mutated.solidity_type, location=LOC_MEMORY), + ) + + return Wrapper( + name=fn.name, + library_function=fn.name, + params=tuple(exposed), + returns=returns, + state_mutability=_wrapper_mutability(fn, binds_state), + call_args=tuple(call_args), + returns_mutated_param=mutated.name if mutated is not None else None, + ) + + +def _abi_key(wrapper: Wrapper) -> Tuple[str, ...]: + """The signature solc uses to detect a duplicate definition. + + Return types are excluded deliberately — they are not part of the key, which is why + OpenZeppelin's three ``values()`` overloads collide once their storage receivers are + dropped. + """ + return (wrapper.name, *(p.solidity_type for p in wrapper.params)) + + +def _suffixed(name: str, suffix: str) -> str: + """Append a mangling suffix without doubling the separator. + + A CVL-escaped name already ends in ``_`` (``at`` becomes ``at_``), so the naive join + would produce ``at__AddressSet``. + """ + return f"{name}{suffix}" if name.endswith("_") else f"{name}_{suffix}" + + +def _mangle_collisions( + wrappers: Sequence[Wrapper], originals: Sequence[LibFunction], library_name: str +) -> List[Wrapper]: + """Give colliding wrappers distinct names, leaving unique ones untouched. + + Dropping the storage receiver is what creates the collisions: OpenZeppelin's + ``length(Bytes32Set)``, ``length(AddressSet)`` and ``length(UintSet)`` all become + ``length()``. The receiver type is therefore what distinguishes them again. + + ``originals`` is positional, not keyed by name: overloads share a name, so a lookup + by name would hand every member of a collision group the same receiver and rename + them all identically — reproducing the collision it set out to remove. + + A group that collides with no storage receiver to name falls back to an ordinal, and + the whole pass repeats until no key is shared, so a suffix that happens to recreate + a collision still converges. + """ + resolved = list(wrappers) + for _ in range(len(resolved) + 1): + grouped: Dict[Tuple[str, ...], List[int]] = defaultdict(list) + for index, wrapper in enumerate(resolved): + grouped[_abi_key(wrapper)].append(index) + + colliding = [indices for indices in grouped.values() if len(indices) > 1] + if not colliding: + return resolved + + for indices in colliding: + for ordinal, index in enumerate(indices): + wrapper = resolved[index] + receivers = originals[index].storage_params + if receivers: + suffix = "_".join( + short_type_name(p.solidity_type, library_name) for p in receivers + ) + else: + suffix = str(ordinal) + resolved[index] = Wrapper( + name=_suffixed(wrapper.name, suffix), + library_function=wrapper.library_function, + params=wrapper.params, + returns=wrapper.returns, + state_mutability=wrapper.state_mutability, + call_args=wrapper.call_args, + returns_mutated_param=wrapper.returns_mutated_param, + ) + + raise LibraryHarnessError( + "could not give every wrapper a unique signature; names still collide after " + "repeated mangling" + ) + + +def _walk_member( + node: MemberNode, + access: str, + name_parts: List[str], + params: List[LibParam], + readers: List[StorageReader], + depth: int, +) -> None: + """Reach through one member towards an ABI-encodable leaf, emitting a reader there. + + A struct extends the access path; a mapping or array adds a lookup parameter and + descends into the value. The leaf is what can actually cross the ABI boundary — so + a mapping is never returned, but the value it holds is. + """ + if depth > _MAX_READER_DEPTH or len(readers) >= _MAX_READERS: + return + + if node.kind == KIND_STRUCT: + for child in node.members: + _walk_member( + child, + f"{access}.{child.name}", + [*name_parts, child.name], + params, + readers, + depth + 1, + ) + return + + if node.kind in (KIND_MAPPING, KIND_ARRAY) and node.value is not None: + key_type = node.key_type if node.kind == KIND_MAPPING else "uint256" + key_name = f"key{len(params)}" + _walk_member( + node.value, + f"{access}[{key_name}]", + name_parts, + [ + *params, + LibParam( + name=key_name, + solidity_type=key_type, + location=_type_location(key_type), + ), + ], + readers, + depth + 1, + ) + return + + if node.kind != KIND_VALUE: + return + + readers.append( + StorageReader( + name="_".join(sanitize_identifier(part) for part in name_parts if part), + solidity_type=node.solidity_type, + access_expression=access, + params=tuple(params), + location=_type_location(node.solidity_type), + ) + ) + + +def _storage_readers( + owned: Sequence[OwnedVar], + struct_members: Mapping[str, tuple[MemberNode, ...]], + library_name: str, +) -> List[StorageReader]: + """Expose the owned structs' representation so a spec can state properties about it. + + Without these the harness only echoes the library's own API, and the interesting + properties are exactly the ones relating the API to the representation — + OpenZeppelin's own EnumerableSet spec needs ``_indexOf`` to say that ``at(i)`` maps + back to index ``i``. That reader is the leaf of ``AddressSet -> _inner -> _indexes[key]``, + which is why the walk descends through mappings instead of refusing them. + """ + readers: List[StorageReader] = [] + for var in owned: + # Named after the type and member path rather than the state variable, so a + # reader reads as an accessor (``Bytes32Set_inner_indexes``) instead of repeating + # the storage plumbing in every name. + root = short_type_name(var.solidity_type, library_name) + for member in struct_members.get(var.solidity_type, ()): + _walk_member( + member, + f"{var.var_name}.{member.name}", + [root, member.name], + [], + readers, + 0, + ) + return readers + + +def build_plan( + api: LibraryApi, + harness: ContractHandle, + pragma_line: str, + import_lines: Sequence[str], +) -> HarnessPlan: + """Decide the complete contents of the harness for ``api``. + + Ordering is by library source line, so regenerating an unchanged library produces a + byte-identical file and the harness does not churn in diffs. + """ + library_name = api.contract.contract_name + ordered = sorted(api.functions, key=lambda f: (f.source_line, f.name)) + + skipped: List[Skipped] = [] + wrappable: List[LibFunction] = [] + for fn in ordered: + if reason := _classify(fn, api.struct_members): + skipped.append(reason) + else: + wrappable.append(fn) + + if not wrappable: + raise LibraryHarnessError( + f"no function of library {library_name} can be exposed through a harness " + f"({len(skipped)} skipped) — verifying it would prove nothing" + ) + + owned: Dict[str, OwnedVar] = {} + for fn in wrappable: + for param in fn.storage_params: + if param.solidity_type not in owned: + owned[param.solidity_type] = OwnedVar( + var_name=_owned_var_name(param.solidity_type, library_name), + solidity_type=param.solidity_type, + ) + + wrappers = [_build_wrapper(fn, owned) for fn in wrappable] + # CVL renaming first: doing it after collision mangling could rename a wrapper onto + # a name the mangling had just handed out. + wrappers = [ + Wrapper( + name=escape_reserved(w.name), + library_function=w.library_function, + params=w.params, + returns=w.returns, + state_mutability=w.state_mutability, + call_args=w.call_args, + returns_mutated_param=w.returns_mutated_param, + ) + for w in wrappers + ] + wrappers = _mangle_collisions(wrappers, wrappable, library_name) + + owned_vars = tuple(owned[key] for key in sorted(owned)) + return HarnessPlan( + harness=harness, + library=api.contract, + pragma_line=pragma_line, + import_lines=tuple(import_lines), + owned_vars=owned_vars, + wrappers=tuple(wrappers), + readers=tuple(_storage_readers(owned_vars, api.struct_members, library_name)), + skipped=tuple(skipped), + ) diff --git a/certora_autosetup/harnesser/read_build.py b/certora_autosetup/harnesser/read_build.py new file mode 100644 index 00000000..e4efaeca --- /dev/null +++ b/certora_autosetup/harnesser/read_build.py @@ -0,0 +1,221 @@ +"""Read a library's API out of ``.certora_build.json``. + +The build is the only trustworthy description of what a library declares: it has +resolved inheritance, overloads and type aliases that source text does not. The +harnesser therefore runs a probe compilation first and reads the result here, rather +than parsing Solidity (which is what ``utils/library_harness.py`` must do, because it +runs inside the compilation retry loop before any build artifact exists). + +Two things about the build JSON are easy to get wrong and are load-bearing here: + +- Internal functions live in ``allMethods``, not ``internalFunctions``. ``allMethods`` + is ``contract.methods + contract.internal_funcs`` (external plus internal plus + private), while ``internalFunctions`` holds *autofinder instrumentation*. certora-cli + deliberately generates no autofinders for library-hosted functions, so + ``internalFunctions`` is systematically empty for libraries — reading it would yield + zero wrappers for a library like OpenZeppelin's EnumerableSet, whose entire surface + is internal. +- A library is located by ``(name, source file)``, never by name alone. Solady ships 17 + library names twice, under ``src/utils/`` and ``src/utils/g/``, with differently + scoped structs; matching on the name alone picks an arbitrary one and emits types + that do not resolve against the imported file. +""" + +import json +from pathlib import Path +from typing import Any, Dict, Iterator, List, Optional + +from certora_autosetup.harnesser.model import ( + KIND_ARRAY, + KIND_MAPPING, + KIND_STRUCT, + KIND_VALUE, + LibFunction, + LibParam, + LibraryApi, + LibraryHarnessError, + MemberNode, +) +from certora_autosetup.utils.build_json import contract_source_file, iter_contracts +from certora_autosetup.utils.paths import same_source_file +from certora_autosetup.utils.types import ContractHandle, TypeParseMode, parse_type_descriptor + + +def _param_list(raw: List[Dict[str, Any]], names: List[str], contract_name: str) -> tuple[LibParam, ...]: + """Render one ``fullArgs``/``returns`` list into typed parameters. + + Types are rendered in SOLIDITY mode so nested types keep the qualification the + generated source needs (``EnumerableSet.Bytes32Set``, not ``Bytes32Set``). + """ + params: List[LibParam] = [] + for index, entry in enumerate(raw): + type_desc = entry.get("typeDesc", {}) + solidity_type = parse_type_descriptor(type_desc, TypeParseMode.SOLIDITY, contract_name) + if not solidity_type or solidity_type == "unknown": + # Signalled to the caller as an unnamed unresolved type; the classifier + # turns it into a skip rather than emitting source that will not compile. + solidity_type = "" + name = names[index] if index < len(names) else "" + params.append( + LibParam( + name=name or f"arg{index}", + solidity_type=solidity_type, + location=entry.get("location", "") or "", + desc_kind=str(type_desc.get("type", "")) if isinstance(type_desc, dict) else "", + ) + ) + return tuple(params) + + +#: Guards against a self-referential struct sending the walk into a loop. +_MAX_MEMBER_DEPTH = 6 + + +def _member_node(name: str, type_desc: Dict[str, Any], contract_name: str, depth: int = 0) -> Optional[MemberNode]: + """Describe one struct member as a tree node, recursing through its shape. + + Kept structural rather than flattened to a type string: a reader has to reach + *through* a mapping or array, supplying a key or index, and that is only decidable + from the shape. + """ + if depth > _MAX_MEMBER_DEPTH or not isinstance(type_desc, dict): + return None + + rendered = parse_type_descriptor(type_desc, TypeParseMode.SOLIDITY, contract_name) + if not rendered or rendered == "unknown": + return None + + kind = type_desc.get("type") + + if kind == "UserDefinedStruct": + children = _member_nodes(type_desc.get("structMembers", []), contract_name, depth + 1) + return MemberNode(name=name, solidity_type=rendered, kind=KIND_STRUCT, members=children) + + if kind == "Mapping": + key_desc = type_desc.get("mappingKeyType") or type_desc.get("key") or {} + value_desc = type_desc.get("mappingValueType") or type_desc.get("value") or {} + key_type = parse_type_descriptor(key_desc, TypeParseMode.SOLIDITY, contract_name) + value_node = _member_node("", value_desc, contract_name, depth + 1) + if not key_type or key_type == "unknown" or value_node is None: + return None + return MemberNode( + name=name, solidity_type=rendered, kind=KIND_MAPPING, key_type=key_type, value=value_node + ) + + if kind == "Array": + base_desc = type_desc.get("dynamicArrayBaseType") or type_desc.get("base") or {} + element = _member_node("", base_desc, contract_name, depth + 1) + if element is None: + return None + return MemberNode(name=name, solidity_type=rendered, kind=KIND_ARRAY, value=element) + + return MemberNode(name=name, solidity_type=rendered, kind=KIND_VALUE) + + +def _member_nodes( + raw_members: List[Dict[str, Any]], contract_name: str, depth: int = 0 +) -> tuple[MemberNode, ...]: + nodes: List[MemberNode] = [] + for member in raw_members: + if not isinstance(member, dict): + continue + member_name = member.get("name") or member.get("fieldName") + if not member_name: + continue + node = _member_node( + member_name, member.get("type", {}) or member.get("typeDesc", {}), contract_name, depth + ) + if node is not None: + nodes.append(node) + return tuple(nodes) + + +def _struct_members(contract: Dict[str, Any]) -> Dict[str, tuple[MemberNode, ...]]: + """Collect every struct's member tree, keyed by its qualified Solidity type.""" + members: Dict[str, tuple[MemberNode, ...]] = {} + for type_info in contract.get("solidityTypes", []): + if not isinstance(type_info, dict) or type_info.get("type") != "UserDefinedStruct": + continue + struct_name = type_info.get("structName") + if not struct_name: + continue + containing = type_info.get("containingContract") + qualified = f"{containing}.{struct_name}" if containing else struct_name + members[qualified] = _member_nodes( + type_info.get("structMembers", []), containing or "" + ) + return members + + +def read_library_api(build_json: Path, library: ContractHandle) -> LibraryApi: + """Extract the library's full declared API from a completed build. + + ``library.source_file`` disambiguates same-named libraries; it is matched against + the build's own path for the contract. + """ + library_name = library.contract_name + library_source_file = library.source_file + if not build_json.exists(): + raise LibraryHarnessError( + f"probe build produced no {build_json} — cannot read the library's API" + ) + + with open(build_json, "r") as f: + build_data = json.load(f) + + matched: Optional[Dict[str, Any]] = None + seen_names: List[str] = [] + for contract in iter_contracts(build_data): + name = contract.get("name", "") + if name != library_name: + continue + candidate_file = contract_source_file(contract) + seen_names.append(candidate_file) + if same_source_file(candidate_file, library_source_file): + matched = contract + break + + if matched is None: + if seen_names: + raise LibraryHarnessError( + f"library {library_name} is in the build, but none of its records match " + f"{library_source_file} (found: {', '.join(sorted(set(seen_names)))})" + ) + raise LibraryHarnessError( + f"library {library_name} ({library_source_file}) is absent from the probe " + f"build — the harness stub does not reference it" + ) + + functions: List[LibFunction] = [] + for method in matched.get("allMethods", []): + if not isinstance(method, dict): + continue + name = method.get("name", "") + if not name or name == "constructor": + continue + functions.append( + LibFunction( + name=name, + visibility=method.get("visibility", "") or "internal", + state_mutability=method.get("stateMutability", "") or "nonpayable", + params=_param_list( + method.get("fullArgs", []), method.get("paramNames", []), library_name + ), + returns=_param_list(method.get("returns", []), [], library_name), + source_line=method.get("sourceLine", 0) or 0, + ) + ) + + if not functions: + raise LibraryHarnessError( + f"library {library_name} declares no functions in the build output" + ) + + return LibraryApi( + contract=ContractHandle( + contract_name=library_name, + source_file=contract_source_file(matched) or library_source_file, + ), + functions=tuple(functions), + struct_members=_struct_members(matched), + ) diff --git a/certora_autosetup/harnesser/render.py b/certora_autosetup/harnesser/render.py new file mode 100644 index 00000000..06a2630b --- /dev/null +++ b/certora_autosetup/harnesser/render.py @@ -0,0 +1,208 @@ +"""Turn a decided ``HarnessPlan`` into Solidity source. + +Formatting only — every choice was made in ``plan.py``. The emitted contract holds the +library at arm's length rather than inheriting from it, because a library cannot be a +base contract. + +The file carries a sentinel comment recording the plan hash. That, not the sidecar JSON, +is what makes regeneration idempotent: the file and its provenance cannot drift apart if +they travel together. +""" + +import hashlib +import json +from typing import List, Sequence + +from certora_autosetup.harnesser.model import ( + HarnessPlan, + LibParam, + StorageReader, + Wrapper, +) +from certora_autosetup.utils.contract_linker import render_wrapper_contract + +#: Marks a file as ours and records which plan produced it, so a re-run can tell an +#: up-to-date harness from a stale one without re-deriving the plan. +SENTINEL_PREFIX = "// certora-library-harness:" + +#: The stub's placeholder function. A contract with no external functions is dropped by +#: contract discovery and by the signature database (both skip a contract with no +#: methods), so the probe build would not see the harness at all. It is removed once the +#: real wrappers are known. +STUB_FUNCTION_NAME = "certoraLibraryHarnessPlaceholder" + + +def _render_param(param: LibParam, include_name: bool) -> str: + parts = [param.solidity_type] + if param.location: + parts.append(param.location) + if include_name and param.name: + parts.append(param.name) + return " ".join(parts) + + +def _render_params(params: Sequence[LibParam]) -> str: + return ", ".join(_render_param(p, include_name=True) for p in params) + + +def _render_returns(returns: Sequence[LibParam]) -> str: + if not returns: + return "" + rendered = ", ".join(_render_param(r, include_name=False) for r in returns) + return f" returns ({rendered})" + + +def _render_wrapper(wrapper: Wrapper, library_name: str) -> str: + """Emit one public wrapper delegating to the library.""" + mutability = f" {wrapper.state_mutability}" if wrapper.state_mutability in ("view", "pure") else "" + signature = ( + f" function {wrapper.name}({_render_params(wrapper.params)}) " + f"public{mutability}{_render_returns(wrapper.returns)} {{" + ) + call = f"{library_name}.{wrapper.library_function}({', '.join(wrapper.call_args)});" + + if wrapper.returns_mutated_param: + # The library mutates the argument in place and returns nothing; handing the + # argument back is the only way the mutation is observable to a caller. + body = [f" {call}", f" return {wrapper.returns_mutated_param};"] + elif wrapper.returns: + body = [f" return {call}"] + else: + body = [f" {call}"] + + return "\n".join([signature, *body, " }"]) + + +def _render_reader(reader: StorageReader) -> str: + """Emit a getter over a member of an owned struct.""" + location = f" {reader.location}" if reader.location else "" + return "\n".join( + [ + f" function {reader.name}({_render_params(reader.params)}) " + f"public view returns ({reader.solidity_type}{location}) {{", + f" return {reader.access_expression};", + " }", + ] + ) + + +def plan_hash(plan: HarnessPlan) -> str: + """Stable digest of everything that determines the emitted source.""" + payload = { + "library": plan.library.contract_name, + "source": plan.library.source_file, + "owned": [(v.var_name, v.solidity_type) for v in plan.owned_vars], + "wrappers": [ + ( + w.name, + w.library_function, + [(p.solidity_type, p.location) for p in w.params], + [(r.solidity_type, r.location) for r in w.returns], + w.state_mutability, + list(w.call_args), + w.returns_mutated_param, + ) + for w in plan.wrappers + ], + "readers": [ + (r.name, r.solidity_type, r.location, r.access_expression) for r in plan.readers + ], + } + return hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + + +def sentinel_line(library_name: str, digest: str) -> str: + return ( + f"{SENTINEL_PREFIX} " + + json.dumps({"v": 1, "library": library_name, "plan_hash": digest}, separators=(",", ":")) + ) + + +def read_sentinel(source: str) -> dict | None: + """Recover the provenance record from an existing harness file, if it is ours.""" + for line in source.splitlines(): + stripped = line.strip() + if stripped.startswith(SENTINEL_PREFIX): + try: + return json.loads(stripped[len(SENTINEL_PREFIX):].strip()) + except json.JSONDecodeError: + return None + return None + + +def render_stub( + harness_name: str, + library_name: str, + pragma_line: str, + import_lines: Sequence[str], +) -> str: + """Emit the placeholder harness compiled by the probe build. + + It must import the library so the library lands in the same compilation unit and the + build reports its full API, and it must declare one external function so contract + discovery keeps it. + """ + body = [ + f" function {STUB_FUNCTION_NAME}() external pure returns (uint256) {{", + " return 42;", + " }", + ] + return render_wrapper_contract( + harness_name=harness_name, + parent_name=None, + pragma_line=pragma_line, + import_lines=list(import_lines), + ctor_forward=None, + body_blocks=body, + header_comment_lines=[ + f"{SENTINEL_PREFIX} " + json.dumps({"v": 1, "library": library_name, "stub": True}, separators=(",", ":")), + f"// Placeholder harness for library {library_name}; the probe build reports the", + "// library's API and this file is then regenerated with one wrapper per function.", + ], + ) + + +def render_harness(plan: HarnessPlan) -> str: + """Emit the finished harness: owned state, wrappers, readers.""" + digest = plan_hash(plan) + + body: List[str] = [] + for var in plan.owned_vars: + body.append(f" {var.solidity_type} internal {var.var_name};") + if plan.owned_vars: + body.append("") + + for wrapper in plan.wrappers: + body.append(_render_wrapper(wrapper, plan.library.contract_name)) + body.append("") + + if plan.readers: + body.append(" // Getters over the harness-owned storage, so a spec can relate the") + body.append(" // library's API to the representation it maintains.") + for reader in plan.readers: + body.append(_render_reader(reader)) + body.append("") + + while body and body[-1] == "": + body.pop() + + header = [ + sentinel_line(plan.library.contract_name, digest), + f"// Generated harness exposing library {plan.library.contract_name} as a verifiable contract.", + "// The Prover skips libraries when instantiating parametric rules, so the library's", + "// functions are only reachable through a contract that calls them.", + ] + if plan.skipped: + header.append(f"// {len(plan.skipped)} library function(s) could not be exposed; see the run report.") + + return render_wrapper_contract( + harness_name=plan.harness.contract_name, + parent_name=None, + pragma_line=plan.pragma_line, + import_lines=list(plan.import_lines), + ctor_forward=None, + body_blocks=body, + header_comment_lines=header, + ) diff --git a/certora_autosetup/harnesser/run.py b/certora_autosetup/harnesser/run.py new file mode 100644 index 00000000..b996c235 --- /dev/null +++ b/certora_autosetup/harnesser/run.py @@ -0,0 +1,267 @@ +"""Drive the harnesser end to end: stub, probe build, plan, emit, validate. + +Everything happens here, before autosetup's own run starts. The fill deliberately does +*not* re-enter autosetup's compilation analysis: that path re-seeds its config from the +build-system defaults, so a second pass through it would discard every workaround the +first pass discovered. Autosetup therefore sees only a finished harness and runs exactly +as it does for any hand-written contract. + +The probe build lists the library in ``files`` explicitly. Importing it from the stub is +not enough — an imported-but-unused library is not compiled as its own contract, so the +build reports its structs but none of its functions. +""" + +import json +import os +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import List, Optional, Sequence + +from certora_autosetup.harnesser.model import HarnessPlan, LibraryHarnessError +from certora_autosetup.harnesser.plan import build_plan +from certora_autosetup.harnesser.read_build import read_library_api +from certora_autosetup.harnesser.render import plan_hash, read_sentinel, render_harness, render_stub +from certora_autosetup.utils.build_json import BUILD_JSON_RELPATH, build_json_path +from certora_autosetup.utils.constants import DIR_CERTORA_INTERNAL +from certora_autosetup.utils.logger import logger +from certora_autosetup.utils.paths import user_harness_path +from certora_autosetup.utils.remappings import build_packages_from_remapping_sources +from certora_autosetup.utils.solc_version_resolver import read_pragma_from_source_file +from certora_autosetup.utils.types import ContractHandle + +#: Prefix of the generated contract, so a harness is recognisable in a conf, a report and +#: a rule name without consulting the manifest. +HARNESS_PREFIX = "CertoraLibraryHarness_" + +#: certoraRun refuses to build without a verification target, so the probe supplies a +#: trivially-true spec. It proves nothing and is never used for verification. +_PROBE_SPEC = "rule certoraLibraryHarnessProbe { assert true; }\n" + + +@dataclass(frozen=True) +class HarnessResult: + """What the caller needs in order to swap the main contract and report the outcome.""" + + library: ContractHandle + harness: ContractHandle + plan_hash: str + coverage: dict + wrappers: List[str] + skipped: List[dict] + + #: Flat keys, because the manifest is read back by ``swap.library_behind_harness`` + #: in a later process — and, for a run that failed, by a human. + def to_dict(self) -> dict: + return { + "library_name": self.library.contract_name, + "library_file": self.library.source_file, + "harness_name": self.harness.contract_name, + "harness_file": self.harness.source_file, + "plan_hash": self.plan_hash, + "coverage": self.coverage, + "wrappers": self.wrappers, + "skipped": self.skipped, + } + + +def harness_name_for(library_name: str) -> str: + return f"{HARNESS_PREFIX}{library_name}" + + +def _pragma_line(library_file: Path, project_root: Path) -> str: + """Reuse the library's own pragma so the harness cannot fall outside its range.""" + spec = read_pragma_from_source_file(library_file, project_root) + return f"pragma solidity {spec};" if spec else "" + + +def _import_line(library_file: Path, harness_file: Path) -> str: + """Import the library by a path relative to the harness's own directory. + + The harness lives under ``certora/harnesses/`` rather than beside the library, so + that generating it never dirties a vendored dependency; that makes the relative path + a rebase rather than a plain ``./``. + """ + relative = os.path.relpath(library_file.resolve(), harness_file.parent.resolve()) + return f'import "{relative}";' + + +def _run_probe_build( + project_root: Path, + harness_file: Path, + harness_name: str, + library_file: Path, + library_name: str, + solc: Optional[str], + extra_files: Sequence[ContractHandle], + certora_run_command: str, +) -> None: + """Compile the stub together with the library so the build reports the library's API.""" + spec_path = project_root / DIR_CERTORA_INTERNAL / "certora_library_harness_probe.spec" + spec_path.parent.mkdir(parents=True, exist_ok=True) + spec_path.write_text(_PROBE_SPEC) + + harness_arg = f"{harness_file.relative_to(project_root).as_posix()}:{harness_name}" + library_arg = f"{library_file.relative_to(project_root).as_posix()}:{library_name}" + + command = [ + certora_run_command, + harness_arg, + library_arg, + *(handle.to_config_str() for handle in extra_files), + "--verify", + f"{harness_name}:{spec_path.relative_to(project_root).as_posix()}", + "--compilation_steps_only", + ] + if solc: + command += ["--solc", solc] + # Resolve the packages ourselves rather than letting certora-cli fall back to its own + # scan. That fallback concatenates package.json with remappings.txt and refuses the + # build outright when a key appears in both, which is the normal state of a project + # whose remappings were generated with node_modules installed. This is the same merge + # AutoSetup performs, and the probe build is the one build that runs before it. + packages = build_packages_from_remapping_sources(project_root, logger.log) + if packages: + command += ["--packages", *packages] + + logger.log(f"Probe build: {' '.join(command)}", "INFO", "Harnesser") + completed = subprocess.run( + command, cwd=project_root, capture_output=True, text=True, timeout=1800 + ) + if completed.returncode != 0: + raise LibraryHarnessError( + f"probe build failed for library {library_name}:\n" + f"{completed.stdout[-4000:]}\n{completed.stderr[-4000:]}" + ) + + +def ensure_library_harness( + project_root: Path, + library: ContractHandle, + solc: Optional[str] = None, + extra_files: Sequence[ContractHandle] = (), + certora_run_command: str = "certoraRun", + validate: bool = True, +) -> HarnessResult: + """Generate (or refresh) the harness that makes ``library_name`` verifiable. + + Returns the record the caller needs to swap the main contract. Raises rather than + degrading: a harness that silently omits the functions the user cares about still + runs and still reports success. + """ + project_root = project_root.resolve() + library_name = library.contract_name + library_file = Path(library.source_file) + absolute_library = library_file if library_file.is_absolute() else project_root / library_file + if not absolute_library.exists(): + raise LibraryHarnessError(f"library source {library_file} does not exist") + + harness_name = harness_name_for(library_name) + harness_file = user_harness_path(project_root, harness_name) + harness_file.parent.mkdir(parents=True, exist_ok=True) + + pragma = _pragma_line(absolute_library, project_root) + import_lines = [_import_line(absolute_library, harness_file)] + + # The stub exists only so the probe build has something to compile that is not the + # library itself; it carries one external function because a method-less contract is + # dropped by contract discovery and by the signature database. + harness_file.write_text(render_stub(harness_name, library_name, pragma, import_lines)) + + _run_probe_build( + project_root, + harness_file, + harness_name, + absolute_library, + library_name, + solc, + extra_files, + certora_run_command, + ) + + api = read_library_api( + build_json_path(project_root) or project_root / BUILD_JSON_RELPATH, + ContractHandle( + contract_name=library_name, + source_file=absolute_library.relative_to(project_root).as_posix(), + ), + ) + + plan = build_plan( + api, + harness=ContractHandle( + contract_name=harness_name, + source_file=harness_file.relative_to(project_root).as_posix(), + ), + pragma_line=pragma, + import_lines=import_lines, + ) + harness_file.write_text(render_harness(plan)) + logger.log( + f"Generated {harness_name}: {plan.coverage['wrapped']} wrapper(s), " + f"{len(plan.owned_vars)} owned state var(s), {plan.coverage['skipped']} skipped", + "INFO", + "Harnesser", + ) + + if validate: + # Re-run the same probe against the filled harness. Compiling it in the project's + # real build environment is what proves the wrappers are legal; a bespoke solc + # invocation here would have to reinvent the project's import resolution. + _run_probe_build( + project_root, + harness_file, + harness_name, + absolute_library, + library_name, + solc, + extra_files, + certora_run_command, + ) + + result = _result(plan) + # Written beside the harness, under certora/, because that is the tree a cloud run + # uploads — the skipped list is the only record of what the harness does not cover. + manifest = harness_file.with_suffix(".manifest.json") + manifest.write_text(json.dumps(result.to_dict(), indent=2) + "\n") + return result + + +def _result(plan: HarnessPlan) -> HarnessResult: + return HarnessResult( + library=plan.library, + harness=plan.harness, + plan_hash=plan_hash(plan), + coverage=plan.coverage, + wrappers=[w.name for w in plan.wrappers], + skipped=[ + {"function": s.library_function, "reason": s.reason.value, "detail": s.detail} + for s in plan.skipped + ], + ) + + +def is_generated_library_harness(path: Path) -> bool: + """Whether ``path`` is a harness this module generated. + + Decided by the sentinel in the file, not by its name, so a hand-written file that + happens to match the naming convention is still the author's to overwrite. + """ + if not path.exists() or path.suffix != ".sol": + return False + try: + return read_sentinel(path.read_text(errors="replace")) is not None + except OSError: + return False + + +def existing_harness_provenance(project_root: Path, library_name: str) -> Optional[dict]: + """The provenance record of an already-generated harness, if one is present. + + Read from the file itself rather than a sidecar, so the harness and the record of + what produced it cannot drift apart. + """ + harness_file = user_harness_path(project_root.resolve(), harness_name_for(library_name)) + if not harness_file.exists(): + return None + return read_sentinel(harness_file.read_text(errors="replace")) diff --git a/certora_autosetup/harnesser/swap.py b/certora_autosetup/harnesser/swap.py new file mode 100644 index 00000000..65c1bce1 --- /dev/null +++ b/certora_autosetup/harnesser/swap.py @@ -0,0 +1,156 @@ +"""Replace a library main contract with its generated harness. + +Both entry points — autosetup's CLI and AutoProver's pipeline — perform the swap here, +before anything downstream derives state from the contract name. That matters because +the name reaches the sanity spec, the base conf, the ``verify`` target and the result +keys; swapping later would leave those naming a contract the Prover cannot verify. + +The library stays in the scene alongside the harness. It has to: the harness's wrappers +call into it, and the build only reports a library's own functions when it is named as a +file in its own right. +""" + +import json +from pathlib import Path +from typing import List, Optional, Sequence, Tuple + +from certora_autosetup.harnesser.detect import is_library_main_contract +from certora_autosetup.harnesser.run import HarnessResult, ensure_library_harness +from certora_autosetup.utils.contract_utils import split_contract_spec +from certora_autosetup.utils.logger import logger +from certora_autosetup.utils.types import ContractHandle + + +def library_behind_harness( + project_root: Path, main_contract_handle: ContractHandle +) -> Optional[ContractHandle]: + """The library a generated harness wraps, or ``None`` if this is not one of ours. + + A run can reach AutoSetup with the swap already done: AutoProver's pipeline swaps + before it invokes AutoSetup, which then sees a contract that is not a library and has + nothing to detect. The manifest written beside the harness is what still names the + library, and the library has to be in the conf either way, because the Prover's scene + is the files the conf lists and solc inlining does not put it there. + """ + manifest = (project_root / main_contract_handle.source_file).with_suffix(".manifest.json") + if not manifest.is_file(): + return None + try: + record = json.loads(manifest.read_text()) + except (OSError, json.JSONDecodeError): + return None + if record.get("harness_name") != main_contract_handle.contract_name: + return None + library_name = record.get("library_name") + library_file = record.get("library_file") + if not library_name or not library_file: + return None + return ContractHandle(contract_name=library_name, source_file=library_file) + + +def with_harnessed_library( + project_root: Path, + main_contract_handle: ContractHandle, + additional_contracts: Sequence[str], + swapped_from: Optional[ContractHandle] = None, +) -> List[str]: + """``additional_contracts`` plus the library the harness wraps, if there is one. + + The conf is the Prover's scene, and a library reaches it only by being listed in its + own right: solc inlines the calls, which leaves curated summaries that name the + library unable to typecheck and the build reporting none of its functions. + + ``swapped_from`` is the library this run just swapped away from. Without it the + library is recovered from the harness manifest, which is the case that matters when + the swap happened in an earlier process. + """ + library = swapped_from or library_behind_harness(project_root, main_contract_handle) + if library is None or library == main_contract_handle: + return list(additional_contracts) + # Compare as handles rather than as strings: ``to_config_str`` drops the name when it + # matches the file stem, so one contract has two spellings in a conf. + already = { + ContractHandle(contract_name=name, source_file=path) + for path, name in (split_contract_spec(spec) for spec in additional_contracts) + } + if library in already: + return list(additional_contracts) + return [*additional_contracts, library.to_config_str()] + + +def swap_library_main_contract( + project_root: Path, + main_contract_handle: ContractHandle, + contract_handles: Sequence[ContractHandle], + solc: Optional[str] = None, + certora_run_command: str = "certoraRun", + validate: bool = True, +) -> Tuple[ContractHandle, List[ContractHandle], Optional[HarnessResult]]: + """Return the handle to verify, the scene to verify it in, and what was generated. + + When the main contract is not a library everything is returned unchanged and no + build is run, so this is safe to call on every run. + """ + source_file = Path(main_contract_handle.source_file) + absolute_source = source_file if source_file.is_absolute() else project_root / source_file + + if not is_library_main_contract( + absolute_source, main_contract_handle.contract_name, project_root, solc + ): + scene = list(contract_handles) + library = library_behind_harness(project_root, main_contract_handle) + if library is not None and library not in scene: + scene.append(library) + return main_contract_handle, scene, None + + logger.log( + f"Main contract {main_contract_handle.contract_name} is a library; the Prover " + f"cannot verify it directly. Generating a harness.", + "INFO", + "Harnesser", + ) + + result = ensure_library_harness( + project_root=project_root, + library=main_contract_handle, + solc=solc, + certora_run_command=certora_run_command, + validate=validate, + ) + + harness_handle = result.harness + + scene = list(contract_handles) + if main_contract_handle not in scene: + scene.append(main_contract_handle) + if harness_handle not in scene: + scene.append(harness_handle) + + logger.log( + f"Verifying {harness_handle.contract_name} instead of {main_contract_handle.contract_name}: " + f"{result.coverage['wrapped']}/{result.coverage['total']} library function(s) exposed, " + f"{result.coverage['skipped']} skipped", + "INFO", + "Harnesser", + ) + return harness_handle, scene, result + + +def swap_library_main_contract_paths( + project_root: Path, + relative_path: str, + contract_name: str, + solc: Optional[str] = None, +) -> Tuple[str, str]: + """Path/name form of the swap, for callers that carry the target as two strings. + + Returns the pair unchanged when the target is not a library, so it is safe on every + run. + """ + handle = ContractHandle(contract_name=contract_name, source_file=relative_path) + swapped, _, result = swap_library_main_contract( + project_root=project_root, main_contract_handle=handle, contract_handles=[handle], solc=solc + ) + if result is None: + return relative_path, contract_name + return swapped.source_file, swapped.contract_name diff --git a/certora_autosetup/setup/setup_prover.py b/certora_autosetup/setup/setup_prover.py index 117656d7..855522ae 100644 --- a/certora_autosetup/setup/setup_prover.py +++ b/certora_autosetup/setup/setup_prover.py @@ -17,11 +17,12 @@ import traceback from dataclasses import dataclass from pathlib import Path -from typing import TYPE_CHECKING, Any, Dict, Iterable, Iterator, List, Optional, Sequence, Set, Tuple +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Set, Tuple if TYPE_CHECKING: from certora_autosetup.setup.setup_summaries import SummarySetup +from certora_autosetup.harnesser.swap import library_behind_harness from certora_autosetup.build_systems.base import BuildSystemConfig from certora_autosetup.parsers.build_system_detector import BuildSystem, BuildSystemDetector from certora_autosetup.parsers.foundry import FoundryContractExtractor @@ -42,13 +43,12 @@ from certora_autosetup.setup.solidity_utils import extract_definitions_from_solidity from certora_autosetup.solidity_ast import ( AstDump, - ContractDefinition, - FileAsts, build_parent_graph_json, - iter_nodes_of_type, + iter_contract_declarations, stream_raw_units, ) from packaging.version import Version +from certora_autosetup.utils.build_json import build_json_path, contract_source_file, iter_contracts from certora_autosetup.utils.config_manager import convert_solc_version_to_certora_format from certora_autosetup.cache.cache_fs import cache_path, get_fs from certora_autosetup.utils.file_utils import atomic_write_json_fsspec @@ -71,53 +71,6 @@ from certora_autosetup.utils.solc_version_resolver import VIA_IR_MIN_VERSION from certora_autosetup.utils.types import ContractHandle, ContractKind, TypeParseMode, parse_type_descriptor -@dataclass(frozen=True) -class _ContractDeclView: - """Uniform view of a ContractDefinition for declaration/inheritance scans, whether - it came from the typed AST or from the raw fallback of an unparsable source.""" - - source_path: str - node_id: Optional[int] - name: str - abstract: bool - contract_kind: str - linearized_base_ids: List[int] - - -def _iter_contract_declarations(units: Iterable[FileAsts]) -> Iterator[_ContractDeclView]: - """Every contract declaration in a stream of compilation units (typically - ``AstDump.stream_units(...)``, so the multi-GB dump is never fully in memory): - typed where the models parsed, completed by a raw flat-map sweep for anything - the typed walk could not reach (a solc surprise cannot hide contracts from - setup; Vyper sources contribute nothing — no ContractDefinition nodes).""" - for file_asts in units: - for source in file_asts.sources.values(): - yield from _unit_contract_declarations(source) - - -def _unit_contract_declarations(source) -> Iterator[_ContractDeclView]: - for node in iter_nodes_of_type(source, ContractDefinition): - if isinstance(node, ContractDefinition): - yield _ContractDeclView( - source_path=source.source_path, - node_id=node.id, - name=node.name, - abstract=node.abstract, - contract_kind=node.contractKind, - linearized_base_ids=list(node.linearizedBaseContracts), - ) - else: - yield _ContractDeclView( - source_path=source.source_path, - node_id=node.get("id"), - name=node.get("name") or "", - abstract=bool(node.get("abstract", False)), - contract_kind=node.get("contractKind", "contract"), - linearized_base_ids=[ - i for i in node.get("linearizedBaseContracts", []) if isinstance(i, int) - ], - ) - class CompilationAnalysisError(Exception): """Raised when compilation analysis fails.""" @@ -890,8 +843,8 @@ def _build_declared_contracts_by_file(self) -> Dict[str, Set[str]]: ast_path = self._build_dir / FILE_BUILD_ASTS if self._build_dir else None if not ast_path or not ast_path.exists(): return {} - for decl in _iter_contract_declarations(AstDump.stream_units(ast_path)): - if decl.contract_kind != "interface" and decl.name: + for decl in iter_contract_declarations(AstDump.stream_units(ast_path)): + if decl.contract_kind is not ContractKind.INTERFACE and decl.name: rel = self.scope.get_relative_path(Path(decl.source_path)) contracts_by_file.setdefault(rel, set()).add(decl.name) return contracts_by_file @@ -915,40 +868,32 @@ def generate_all_methods_json(self, build_data: Dict) -> None: all_methods: list = [] method_counts: dict = {} # For calculating overload counts - # Iterate through all objects in the build data - for key, obj in build_data.items(): - if isinstance(obj, dict) and "contracts" in obj: - # Each contract object has a 'contracts' array with actual contract data - for contract in obj.get("contracts", []): - # Get the originating contract name (the main compilation unit) - originating_contract = contract.get("name", "") - - # Process regular methods - if isinstance(contract, dict) and "allMethods" in contract: - for method in contract["allMethods"]: - self._process_method_info( - method, - methods_by_sig, - all_methods, - method_counts, - originating_contract, - is_internal=False, - ) + for contract in iter_contracts(build_data): + # Get the originating contract name (the main compilation unit) + originating_contract = contract.get("name", "") + + # Process regular methods + for method in contract.get("allMethods", []): + self._process_method_info( + method, + methods_by_sig, + all_methods, + method_counts, + originating_contract, + is_internal=False, + ) - # Process internal functions - if isinstance(contract, dict) and "internalFunctions" in contract: - # Internal functions are stored with IDs as keys - for func_id, func_data in contract["internalFunctions"].items(): - if "method" in func_data: - method = func_data["method"] - self._process_method_info( - method, - methods_by_sig, - all_methods, - method_counts, - originating_contract, - is_internal=True, - ) + # Internal functions are stored with IDs as keys + for func_id, func_data in contract.get("internalFunctions", {}).items(): + if "method" in func_data: + self._process_method_info( + func_data["method"], + methods_by_sig, + all_methods, + method_counts, + originating_contract, + is_internal=True, + ) # Write the processed data to all_methods.json output_path = Path(".certora_internal/all_methods.json") @@ -992,106 +937,102 @@ def generate_all_user_defined_types_json(self, build_data: Dict) -> int: all_user_defined_types = [] # Process user-defined types from each contract - for key, obj in build_data.items(): - if isinstance(obj, dict) and "contracts" in obj: - # Each contract object has a 'contracts' array with actual contract data - for contract in obj.get("contracts", []): - if isinstance(contract, dict) and "solidityTypes" in contract: - for type_info in contract.get("solidityTypes", []): - if isinstance(type_info, dict): - type_name = None - qualified_name = None - base_type = None - enum_members = [] - struct_members = [] - - # Handle UserDefinedValueType - if type_info.get("type") == "UserDefinedValueType": - type_name = type_info.get("valueTypeName") - containing_contract = type_info.get( - "containingContract" - ) - - if containing_contract and type_name: - qualified_name = ( - f"{containing_contract}.{type_name}" - ) - elif type_name: - # Use canonicalId to match _qualify_user_defined_type logic - qualified_name = self._qualify_from_canonical_id( - type_info, str(type_name), contract - ) - - # Get the base type - value_type = type_info.get( - "valueTypeAliasedName", {} - ) - if value_type.get("type") == "Primitive": - base_type = value_type.get("primitiveName") - - # Handle UserDefinedStruct - elif type_info.get("type") == "UserDefinedStruct": - type_name = type_info.get("structName") - containing_contract = type_info.get( - "containingContract" - ) - - if containing_contract and type_name: - qualified_name = ( - f"{containing_contract}.{type_name}" - ) - elif type_name: - # Use canonicalId to match _qualify_user_defined_type logic - qualified_name = self._qualify_from_canonical_id( - type_info, str(type_name), contract - ) - - base_type = "struct" - # Extract struct members - struct_members = type_info.get("structMembers", []) - - # Handle UserDefinedEnum - elif type_info.get("type") == "UserDefinedEnum": - type_name = type_info.get("enumName") - containing_contract = type_info.get( - "containingContract" - ) - - if containing_contract and type_name: - qualified_name = ( - f"{containing_contract}.{type_name}" - ) - elif type_name: - # Use canonicalId to match _qualify_user_defined_type logic - qualified_name = self._qualify_from_canonical_id( - type_info, str(type_name), contract - ) - - base_type = "uint8" - # Extract enum members - enum_members = type_info.get("enumMembers", []) - - # Add to collection if we found a valid type - if type_name and qualified_name: - user_type_info = { - "typeName": type_name, - "qualifiedName": qualified_name, - "baseType": base_type, - "typeCategory": type_info.get("type"), - "containingContract": type_info.get( - "containingContract" - ), - "main_contract": contract.get("name"), - "canonicalId": type_info.get("canonicalId", ""), - } - - # Add enum members for UserDefinedEnum - if type_info.get("type") == "UserDefinedEnum": - user_type_info["enumMembers"] = enum_members - # Add struct members for UserDefinedStruct - if type_info.get("type") == "UserDefinedStruct": - user_type_info["structMembers"] = struct_members - all_user_defined_types.append(user_type_info) + for contract in iter_contracts(build_data): + for type_info in contract.get("solidityTypes", []): + if isinstance(type_info, dict): + type_name = None + qualified_name = None + base_type = None + enum_members = [] + struct_members = [] + + # Handle UserDefinedValueType + if type_info.get("type") == "UserDefinedValueType": + type_name = type_info.get("valueTypeName") + containing_contract = type_info.get( + "containingContract" + ) + + if containing_contract and type_name: + qualified_name = ( + f"{containing_contract}.{type_name}" + ) + elif type_name: + # Use canonicalId to match _qualify_user_defined_type logic + qualified_name = self._qualify_from_canonical_id( + type_info, str(type_name), contract + ) + + # Get the base type + value_type = type_info.get( + "valueTypeAliasedName", {} + ) + if value_type.get("type") == "Primitive": + base_type = value_type.get("primitiveName") + + # Handle UserDefinedStruct + elif type_info.get("type") == "UserDefinedStruct": + type_name = type_info.get("structName") + containing_contract = type_info.get( + "containingContract" + ) + + if containing_contract and type_name: + qualified_name = ( + f"{containing_contract}.{type_name}" + ) + elif type_name: + # Use canonicalId to match _qualify_user_defined_type logic + qualified_name = self._qualify_from_canonical_id( + type_info, str(type_name), contract + ) + + base_type = "struct" + # Extract struct members + struct_members = type_info.get("structMembers", []) + + # Handle UserDefinedEnum + elif type_info.get("type") == "UserDefinedEnum": + type_name = type_info.get("enumName") + containing_contract = type_info.get( + "containingContract" + ) + + if containing_contract and type_name: + qualified_name = ( + f"{containing_contract}.{type_name}" + ) + elif type_name: + # Use canonicalId to match _qualify_user_defined_type logic + qualified_name = self._qualify_from_canonical_id( + type_info, str(type_name), contract + ) + + base_type = "uint8" + # Extract enum members + enum_members = type_info.get("enumMembers", []) + + # Add to collection if we found a valid type + if type_name and qualified_name: + user_type_info = { + "typeName": type_name, + "qualifiedName": qualified_name, + "baseType": base_type, + "typeCategory": type_info.get("type"), + "containingContract": type_info.get( + "containingContract" + ), + "main_contract": contract.get("name"), + "canonicalId": type_info.get("canonicalId", ""), + } + + # Add enum members for UserDefinedEnum + if type_info.get("type") == "UserDefinedEnum": + user_type_info["enumMembers"] = enum_members + # Add struct members for UserDefinedStruct + if type_info.get("type") == "UserDefinedStruct": + user_type_info["structMembers"] = struct_members + all_user_defined_types.append(user_type_info) # Write user-defined types to JSON file types_output_path = PATH_ALL_USER_DEFINED_TYPES_JSON @@ -1197,39 +1138,37 @@ def generate_bytes_mappings_json(self, build_data: Dict) -> None: """ bytes_mappings_list = [] - # Iterate through all contracts in the build data - for contract_data in build_data.values(): - for contract in contract_data.get('contracts', []): - contract_name = contract.get('name') - source_file = contract.get('file') - - if not contract_name or not source_file: - continue - - # solc <0.5.13 doesn't emit native storageLayout output — the key is present - # but explicitly null rather than absent, so .get(key, {}) alone doesn't catch it. - storage_layout = contract.get('storageLayout') or {} - bytes_mapping_fields = [] - - # Check each storage field - for storage_item in storage_layout.get('storage', []): - descriptor = storage_item.get('descriptor', {}) - - # Check if this is a mapping with bytes key - if descriptor.get('type') == 'Mapping': - key_type = descriptor.get('mappingKeyType', {}) - if key_type.get('type') == 'PackedBytes': - field_name = storage_item.get('label', '') - if field_name: - bytes_mapping_fields.append(field_name) - - # Add to list if we found any bytes mapping fields - if bytes_mapping_fields: - bytes_mappings_list.append({ - "contract_name": contract_name, - "source_file": source_file, - "bytes_mapping_fields": bytes_mapping_fields - }) + for contract in iter_contracts(build_data): + contract_name = contract.get('name') + source_file = contract.get('file') + + if not contract_name or not source_file: + continue + + # solc <0.5.13 doesn't emit native storageLayout output — the key is present + # but explicitly null rather than absent, so .get(key, {}) alone doesn't catch it. + storage_layout = contract.get('storageLayout') or {} + bytes_mapping_fields = [] + + # Check each storage field + for storage_item in storage_layout.get('storage', []): + descriptor = storage_item.get('descriptor', {}) + + # Check if this is a mapping with bytes key + if descriptor.get('type') == 'Mapping': + key_type = descriptor.get('mappingKeyType', {}) + if key_type.get('type') == 'PackedBytes': + field_name = storage_item.get('label', '') + if field_name: + bytes_mapping_fields.append(field_name) + + # Add to list if we found any bytes mapping fields + if bytes_mapping_fields: + bytes_mappings_list.append({ + "contract_name": contract_name, + "source_file": source_file, + "bytes_mapping_fields": bytes_mapping_fields + }) # Write through fsspec so it lands on the cache prefix (S3 in SaaS, local in CLI) # and the autosetup cache-hit path can read it back. The sole same-run reader @@ -1305,7 +1244,7 @@ def _extract_inheritance_and_abstract_from_ast(self, ast_file_path: Optional[Pat # Stream the (multi-GB) .asts.json once, keeping only the slim per-contract # views needed below so the dump is never fully materialized. - declarations = list(_iter_contract_declarations(AstDump.stream_units(ast_file_path))) + declarations = list(iter_contract_declarations(AstDump.stream_units(ast_file_path))) # Build ID to contract name mapping once id_to_name = { @@ -1317,7 +1256,7 @@ def _extract_inheritance_and_abstract_from_ast(self, ast_file_path: Optional[Pat if not decl.name: continue # Check if abstract or interface - if decl.abstract or decl.contract_kind == "interface": + if decl.abstract or decl.contract_kind is ContractKind.INTERFACE: abstract_contracts.add(decl.name) self.log(f"Identified {'abstract' if decl.abstract else 'interface'}: {decl.name}", "DEBUG") @@ -1356,85 +1295,75 @@ def _extract_contract_infos_from_build( seen_contracts = set() # Discover contracts and create ContractInfo objects - for contract_key, contract_data in build_data.items(): - if ( - not isinstance(contract_data, dict) - or "contracts" not in contract_data - ): + for contract in iter_contracts(build_data): + methods = contract.get("methods", []) + if not methods: continue - for contract in contract_data.get("contracts", []): - if not isinstance(contract, dict): - continue - - methods = contract.get("methods", []) - if not methods: - continue - - contract_name = contract.get("name", "Unknown") - - # Skip if already processed - if contract_name in seen_contracts: - continue - seen_contracts.add(contract_name) - - # Get source file directly from contract object (canonical source) - source_file_str = contract.get("original_file") or contract.get("file") - - # Fall back to method inspection only if contract-level fields are missing - if not source_file_str: - fallback_source = None - for method in methods: - original_file = method.get("originalFile") - if original_file: - fallback_source = original_file - - # Prefer file that matches contract name (where contract is actually defined) - if original_file.endswith(f"/{contract_name}.sol") or original_file.endswith(f"\\{contract_name}.sol"): - source_file_str = original_file - break - if not source_file_str and fallback_source: - source_file_str = fallback_source - - # Final fallback - if not source_file_str: - source_file_str = "unknown.sol" - - # Determine contract kind (basic heuristic) - is_library = any( - method.get("isLibrary", False) for method in methods - ) - kind = ContractKind.LIBRARY if is_library else ContractKind.CONTRACT + contract_name = contract.get("name", "Unknown") + + # Skip if already processed + if contract_name in seen_contracts: + continue + seen_contracts.add(contract_name) + + # Get source file directly from contract object (canonical source) + source_file_str = contract_source_file(contract) - # Extract constructor params - ctor_params = None + # Fall back to method inspection only if contract-level fields are missing + if not source_file_str: + fallback_source = None for method in methods: - if method.get("name", "") == "constructor": - params = [] - for arg, param_name in zip( - method.get("fullArgs", []), method.get("paramNames", []) - ): - type_desc = arg.get("typeDesc", {}) - sol_type = parse_type_descriptor(type_desc, TypeParseMode.SOLIDITY) - location = arg.get("location", "") - if location in ("memory", "calldata", "storage"): - sol_type = f"{sol_type} {location}" - params.append((sol_type, param_name)) - if params: - ctor_params = params + original_file = method.get("originalFile") + if original_file: + fallback_source = original_file + + # Prefer file that matches contract name (where contract is actually defined) + if original_file.endswith(f"/{contract_name}.sol") or original_file.endswith(f"\\{contract_name}.sol"): + source_file_str = original_file break + if not source_file_str and fallback_source: + source_file_str = fallback_source - # Create contract info with inheritance - contract_info = ContractInfo( - name=contract_name, - kind=kind, - source_file=Path(source_file_str), - inherits_from=[], # added later via _extract_inheritance_from_ast() - artifact_path=build_json_path, - constructor_params=ctor_params, - ) + # Final fallback + if not source_file_str: + source_file_str = "unknown.sol" + + # Determine contract kind (basic heuristic) + is_library = any( + method.get("isLibrary", False) for method in methods + ) + kind = ContractKind.LIBRARY if is_library else ContractKind.CONTRACT + + # Extract constructor params + ctor_params = None + for method in methods: + if method.get("name", "") == "constructor": + params = [] + for arg, param_name in zip( + method.get("fullArgs", []), method.get("paramNames", []) + ): + type_desc = arg.get("typeDesc", {}) + sol_type = parse_type_descriptor(type_desc, TypeParseMode.SOLIDITY) + location = arg.get("location", "") + if location in ("memory", "calldata", "storage"): + sol_type = f"{sol_type} {location}" + params.append((sol_type, param_name)) + if params: + ctor_params = params + break + + # Create contract info with inheritance + contract_info = ContractInfo( + name=contract_name, + kind=kind, + source_file=Path(source_file_str), + inherits_from=[], # added later via _extract_inheritance_from_ast() + artifact_path=build_json_path, + constructor_params=ctor_params, + ) - contract_infos.append(contract_info) + contract_infos.append(contract_info) return contract_infos @@ -1515,17 +1444,16 @@ def getASTParentGraphPath(self) -> Path: def process_certora_build_json(self) -> bool: """Process .certora_build.json to extract method information.""" - build_json_path = Path(".certora_internal/latest/.certora_build.json") - self._build_dir = build_json_path.parent - - self.log(f"Processing build json: {build_json_path.resolve()}") - - if not build_json_path.exists(): - self.log(f"Build JSON not found at: {build_json_path}", "ERROR") + build_json = build_json_path(Path(".")) + if not build_json: + self.log("Build JSON not found under .certora_internal", "ERROR") return False + self._build_dir = build_json.parent + + self.log(f"Processing build json: {build_json.resolve()}") try: - with open(build_json_path, "r") as f: + with open(build_json, "r") as f: build_data = json.load(f) # Generate all_methods.json @@ -1556,7 +1484,7 @@ def process_certora_build_json(self) -> bool: self.generate_ast_graph(asts_target) # Generate signature database (uses the ast file copied before) - self.generate_signature_database_json(build_json_path) + self.generate_signature_database_json(build_json) # Generate bytes mappings JSON self.generate_bytes_mappings_json(build_data) @@ -1568,7 +1496,12 @@ def process_certora_build_json(self) -> bool: self.log(f"Traceback: {traceback.format_exc()}", "ERROR") return False - def run_setup_summaries(self, contract_files: List[str], main_contract: str) -> bool: + def run_setup_summaries( + self, + contract_files: List[str], + main_contract: str, + excluded_library: Optional[str] = None, + ) -> bool: """ Run setup_summaries to detect and configure library summaries. On success, the constructed ``SummarySetup`` is stored on ``self.summary_setup`` so @@ -1592,6 +1525,7 @@ def run_setup_summaries(self, contract_files: List[str], main_contract: str) -> include_dependencies=True, enable_llm=not self.skip_llm, custom_recipe=None, + excluded_library=excluded_library, ) if configured: # Summarize the initial scene (main + additional contracts); call resolution @@ -1663,8 +1597,15 @@ def setup_prover( # recipe analysis scans only the files passed directly. If contract A imports and # uses B, and B uses mulDiv from PRBMath, scanning just A would miss the mulDiv # call from B. Including all contract files ensures we catch transitive usage. + # A harness is the verification target precisely so the library it wraps can be + # verified, so that library is the one contract a summary must never replace. The + # manifest is what names it: AutoProver's pipeline swaps in an earlier process, and by + # the time we run the main contract is simply not a library any more. + excluded_library = library_behind_harness(Path.cwd(), main_contract_handle) success_summaries = self.run_setup_summaries( - [ch.source_file for ch in surviving_contracts], main_contract_name + [ch.source_file for ch in surviving_contracts], + main_contract_name, + excluded_library=excluded_library.contract_name if excluded_library else None, ) if not success_summaries: raise SummarySetupError("Setup summaries generation failed") diff --git a/certora_autosetup/setup/setup_summaries.py b/certora_autosetup/setup/setup_summaries.py index 75ecf764..e3f2e3f4 100755 --- a/certora_autosetup/setup/setup_summaries.py +++ b/certora_autosetup/setup/setup_summaries.py @@ -61,32 +61,11 @@ from certora_autosetup.parsers.spec_imports import parse_imports_from_spec from certora_autosetup.setup.summary_resolver import resolve_summary_specs from certora_autosetup.setup.signature_types import InheritanceGraph +from certora_autosetup.utils.cvl_keywords import escape_reserved -# CVL grammar keyword terminals that cannot double as an identifier. A Solidity parameter whose name -# equals one of these is lexed as that keyword inside a methods{} entry, which is a syntax error; such -# names are suffixed with "_" before emission (see _cvl_safe_param_name). -# -# This deliberately EXCLUDES the terminals listed under the `usable_keywords` production in cvl.cup -# (exists, forall, sum, usum, using, as, import, use, builtin, override, sig, description, invariant, -# preserved, weak, strong, onTransactionBoundary, old, hook, unresolved). The grammar accepts those -# wherever an identifier is expected, so a parameter named after one parses fine and must NOT be -# mangled. Note: uppercase "UNRESOLVED" is a distinct summary keyword and remains reserved. -CVL_RESERVED_WORDS = frozenset({ - "ALL", "ALWAYS", "ASSERT_FALSE", "AUTO", "CONSTANT", "Create", "DELETE", "DISPATCH", "DISPATCHER", - "HAVOC_ALL", "HAVOC_ECF", "NONDET", "PER_CALLEE_CONSTANT", "STORAGE", "Sload", "Sstore", "Tload", - "Tstore", "UNRESOLVED", "assert", "assuming", "at", "axiom", "default", "definition", "else", - "event", "expect", "fallback", "false", "filtered", "function", "ghost", "good_description", - "havoc", "if", "in", "indexed", "lastReverted", "lastStorage", "links", "mapping", "methods", - "new", "norevert", "persistent", "require", "requireInvariant", "reset_storage", "return", - "returns", "revert", "rule", "satisfy", "sort", "true", "void", "with", "withrevert", "xor", -}) -def _cvl_safe_param_name(name: str) -> str: - """The parameter name with a trailing "_" if it equals a CVL reserved word, otherwise unchanged.""" - return f"{name}_" if name in CVL_RESERVED_WORDS else name - try: from dotenv import load_dotenv @@ -227,8 +206,8 @@ class DecimalSummary(BaseModel): @property def summary_line(self) -> str: - params = ", ".join([ f"{_pprint_type(p.ty)} {_cvl_safe_param_name(p.name)}" for p in self.param_list ]) - return f"function _.{self.method_name}({params}) internal => {self.cvl_function_name}({_cvl_safe_param_name(self.amount_parameter)}) expect {self.return_type.ty_name};" + params = ", ".join([ f"{_pprint_type(p.ty)} {escape_reserved(p.name)}" for p in self.param_list ]) + return f"function _.{self.method_name}({params}) internal => {self.cvl_function_name}({escape_reserved(self.amount_parameter)}) expect {self.return_type.ty_name};" @property def cvl_function(self) -> str: @@ -270,7 +249,7 @@ class NondetSummary(BaseModel): @property def summary_line(self) -> str: - params = ", ".join([f"{_pprint_type(p.ty)} {_cvl_safe_param_name(p.name)}" for p in self.param_list]) + params = ", ".join([f"{_pprint_type(p.ty)} {escape_reserved(p.name)}" for p in self.param_list]) if self.return_type is not None: return_types = ", ".join([ _pprint_type(ty) for ty in self.return_type @@ -382,6 +361,10 @@ def __init__(self, verbose: int = 0, inheritance_graph: InheritanceGraph | None # summary attached and therefore should be added to the scene. self.matched_functions: Set[str] = set() + # A library whose own code is the verification target, so it is the one thing + # that must never be summarized. None on every ordinary run. + self.excluded_library: Optional[str] = None + # Every contract name that has entered the verification scene so far # (initial main + additional + call-resolution batches). Drives the # scene-wide Math.Rounding classification: qualifier contracts must be @@ -717,6 +700,26 @@ def find_all_library_files( log_func=self.log, ) + def _excluded_library_keys(self) -> Set[str]: + """Curated keys that would summarize the library under verification.""" + if self.excluded_library is None: + return set() + return { + key + for key, info in self.function_summaries.items() + if self.excluded_library in (info.get("library_names") or ()) + } + + def _excluded_library_methods(self) -> Set[Tuple[str, str]]: + """``(contract, method)`` pairs the LLM must leave alone, in its skip-set shape.""" + if self.excluded_library is None: + return set() + return { + (self.excluded_library, m["name"]) + for m in self.methods_parser.get_all_methods() + if m.get("contractName") == self.excluded_library and m.get("name") + } + def copy_summaries_folder(self, matched_function_keys: Iterable[str]) -> Path: """Copy only the bundled summary files referenced by matched curated keys into ``certora/specs/summaries/``. @@ -1280,7 +1283,7 @@ def is_array_type(sol_type: str) -> bool: params = [] for i, param_type in enumerate(param_types): param_name = param_names[i] if i < len(param_names) and param_names[i] else f"" - param_name = _cvl_safe_param_name(param_name) + param_name = escape_reserved(param_name) location = locations[i] if i < len(locations) else "" cvl_type, classification = classify_solidity_type(param_type) @@ -2475,6 +2478,19 @@ async def on_contracts_entered_scene(self, contract_names: List[str], main_contr # single -> mixed) without re-matching oz_Math_mulDiv itself, so # scene-sensitive templates are re-materialized whenever they have ever # matched (materialization is idempotent, aggregator imports dedup). + # A summary replaces the code it summarizes, so summarizing the library under + # verification would have every rule assert against the summary instead of the code + # the run exists to check. Subtracted here rather than skipped inside the matcher: the + # matcher also returns the (contract, method) tuples that become per_contract_skip + # below, and those still have to shield the same methods from the LLM step. + excluded = self._excluded_library_keys() + if excluded & curated_keys: + self.log( + f"Not summarizing {self.excluded_library} — it is the library under " + f"verification; dropped curated {sorted(excluded & curated_keys)}" + ) + curated_keys -= excluded + rerender_keys = curated_keys | (self.matched_functions & SCENE_SENSITIVE_TEMPLATE_KEYS) if rerender_keys: # Publish for downstream consumers (autosetup's library-scene filter) and for @@ -2489,11 +2505,16 @@ async def on_contracts_entered_scene(self, contract_names: List[str], main_contr # 2. LLM analysis per contract, skipping curated-covered methods. if self._enable_llm: + excluded_methods = self._excluded_library_methods() for name in contract_names: await self.analyze_contract( name, self._llm_contract_files, - methods_to_skip=per_contract_skip.get(name), + # The LLM would otherwise summarize the library under verification on its + # own: its non-linear-ops recipe targets exactly the internal pure + # functions an arithmetic library is made of, and unlike BitMaps that + # failure is silent — the run comes back green having proved nothing. + methods_to_skip=set(per_contract_skip.get(name) or ()) | excluded_methods, custom_recipe=self._custom_recipe, ) @@ -2518,6 +2539,7 @@ def configure( include_dependencies: bool = False, enable_llm: bool = False, custom_recipe: Optional[str] = None, + excluded_library: Optional[str] = None, ) -> bool: """Capture the configuration for a summarization run; perform no summarization. @@ -2554,6 +2576,7 @@ def configure( self.main_contract = main_contract self.additional_names = [split_contract_spec(ac)[1] for ac in (additional_contracts or [])] + self.excluded_library = excluded_library self.log(f"Main contract for analysis: {main_contract}") if self.additional_names: diff --git a/certora_autosetup/setup/signature_manager.py b/certora_autosetup/setup/signature_manager.py index b4fa6162..1a118517 100644 --- a/certora_autosetup/setup/signature_manager.py +++ b/certora_autosetup/setup/signature_manager.py @@ -18,6 +18,7 @@ compute_signature_selector, ) from certora_autosetup.cache.cache_fs import cache_path, get_fs +from certora_autosetup.utils.build_json import iter_contracts from certora_autosetup.utils.constants import DIR_CERTORA_INTERNAL, DIR_SIGNATURE_STATE from certora_autosetup.utils.logger import logger from certora_autosetup.utils.types import ( @@ -71,123 +72,113 @@ def extract_signatures_from_build( signatures = {} - for contract_key, contract_data in build_data.items(): - if ( - not isinstance(contract_data, dict) - or "contracts" not in contract_data - ): + for contract in iter_contracts(build_data): + methods = contract.get("methods", []) + if not methods: continue - for contract in contract_data.get("contracts", []): - if not isinstance(contract, dict): - continue + # Get contract name from first method + contract_name = methods[0].get("contractName", "Unknown") + + for method in methods: + method_name = method.get("name", "") - methods = contract.get("methods", []) - if not methods: + if method_name == "constructor": continue - # Get contract name from first method - contract_name = methods[0].get("contractName", "Unknown") + # Get the sighash directly from Certora (this is the correct ABI selector) + sighash_str = str(method.get("sighash", "0")) - for method in methods: - method_name = method.get("name", "") + # Skip methods with zero sighash (internal/constructor methods) + if sighash_str == "0": + continue - if method_name == "constructor": - continue + certora_selector = self._convert_sighash_to_selector(sighash_str) - # Get the sighash directly from Certora (this is the correct ABI selector) - sighash_str = str(method.get("sighash", "0")) + # Build canonical and internal parameter types + canonical_param_types = [] + internal_param_types = [] + type_descs = [] - # Skip methods with zero sighash (internal/constructor methods) - if sighash_str == "0": - continue + for arg in method.get("fullArgs", []): + type_desc = arg.get("typeDesc", {}) + type_descs.append(type_desc) + # Get canonical type (MarketId -> bytes32, InternalUserData -> (address,uint256,bool)) + canonical_type = parse_type_descriptor(type_desc, TypeParseMode.CANONICAL) + canonical_param_types.append(canonical_type) - certora_selector = self._convert_sighash_to_selector(sighash_str) + # Get internal type (keeps MarketId as MarketId, InternalUserData as InternalUserData) + internal_type = parse_type_descriptor(type_desc, TypeParseMode.INTERNAL) + internal_param_types.append(internal_type) - # Build canonical and internal parameter types - canonical_param_types = [] - internal_param_types = [] - type_descs = [] + # Build signatures + canonical_signature = ( + f"{method_name}({','.join(canonical_param_types)})" + ) + internal_type_signature = ( + f"{method_name}({','.join(internal_param_types)})" + ) - for arg in method.get("fullArgs", []): - type_desc = arg.get("typeDesc", {}) - type_descs.append(type_desc) - # Get canonical type (MarketId -> bytes32, InternalUserData -> (address,uint256,bool)) - canonical_type = parse_type_descriptor(type_desc, TypeParseMode.CANONICAL) - canonical_param_types.append(canonical_type) + # Use Certora's sighash directly - it's already computed correctly! + canonical_selector = certora_selector - # Get internal type (keeps MarketId as MarketId, InternalUserData as InternalUserData) - internal_type = parse_type_descriptor(type_desc, TypeParseMode.INTERNAL) - internal_param_types.append(internal_type) + # For internal selector: if signatures differ, compute it; otherwise reuse canonical + if canonical_signature == internal_type_signature: + # No user-defined types, selectors are identical + internal_selector = canonical_selector + else: + # Different signatures due to user-defined types, compute internal selector + internal_selector = compute_signature_selector(internal_type_signature) + if not internal_selector or internal_selector == "0x00000000": + # Fallback to canonical selector if computation fails + logger.debug( + f"Internal selector computation failed for {internal_type_signature}, using canonical" + ) + internal_selector = canonical_selector - # Build signatures - canonical_signature = ( - f"{method_name}({','.join(canonical_param_types)})" - ) - internal_type_signature = ( - f"{method_name}({','.join(internal_param_types)})" + if not canonical_selector: + logger.warning( + f"Failed to get valid selectors for: {canonical_signature} / {internal_type_signature}" ) + continue - # Use Certora's sighash directly - it's already computed correctly! - canonical_selector = certora_selector + # Generate dispatcher entry name with contract-qualified types + dispatcher_entry_name = self._generate_dispatcher_entry_name( + method_name, type_descs, contract_name + ) - # For internal selector: if signatures differ, compute it; otherwise reuse canonical - if canonical_signature == internal_type_signature: - # No user-defined types, selectors are identical - internal_selector = canonical_selector - else: - # Different signatures due to user-defined types, compute internal selector - internal_selector = compute_signature_selector(internal_type_signature) - if not internal_selector or internal_selector == "0x00000000": - # Fallback to canonical selector if computation fails - logger.debug( - f"Internal selector computation failed for {internal_type_signature}, using canonical" - ) - internal_selector = canonical_selector - - if not canonical_selector: - logger.warning( - f"Failed to get valid selectors for: {canonical_signature} / {internal_type_signature}" - ) - continue + # Get state mutability info + state_mutability = method.get("stateMutability", "nonpayable") + is_view = state_mutability in ["view", "pure"] + is_pure = state_mutability == "pure" + + # Create signature info object + signature_info = { + "signature": canonical_signature, + "selector": canonical_selector, + "internal_type_signature": internal_type_signature, + "internal_type_selector": internal_selector, + "dispatcher_entry_name": dispatcher_entry_name, + "is_view": is_view, + "is_pure": is_pure, + "source_file": method.get("originalFile", ""), + } - # Generate dispatcher entry name with contract-qualified types - dispatcher_entry_name = self._generate_dispatcher_entry_name( - method_name, type_descs, contract_name - ) + # Store by canonical selector, accumulating all implementing contracts + if canonical_selector in signatures: + signatures[canonical_selector]["contracts"].add(contract_name) + else: + signature_info["contracts"] = {contract_name} + signatures[canonical_selector] = signature_info - # Get state mutability info - state_mutability = method.get("stateMutability", "nonpayable") - is_view = state_mutability in ["view", "pure"] - is_pure = state_mutability == "pure" - - # Create signature info object - signature_info = { - "signature": canonical_signature, - "selector": canonical_selector, - "internal_type_signature": internal_type_signature, - "internal_type_selector": internal_selector, - "dispatcher_entry_name": dispatcher_entry_name, - "is_view": is_view, - "is_pure": is_pure, - "source_file": method.get("originalFile", ""), - } - - # Store by canonical selector, accumulating all implementing contracts - if canonical_selector in signatures: - signatures[canonical_selector]["contracts"].add(contract_name) + # Also store by internal selector for dispatcher lookup + if internal_selector != canonical_selector: + if internal_selector in signatures: + signatures[internal_selector]["contracts"].add(contract_name) else: - signature_info["contracts"] = {contract_name} - signatures[canonical_selector] = signature_info - - # Also store by internal selector for dispatcher lookup - if internal_selector != canonical_selector: - if internal_selector in signatures: - signatures[internal_selector]["contracts"].add(contract_name) - else: - internal_info = dict(signature_info) - internal_info["contracts"] = {contract_name} - signatures[internal_selector] = internal_info + internal_info = dict(signature_info) + internal_info["contracts"] = {contract_name} + signatures[internal_selector] = internal_info logger.info(f"Extracted {len(signatures)} function signatures") return signatures diff --git a/certora_autosetup/solidity_ast/__init__.py b/certora_autosetup/solidity_ast/__init__.py index 8c870d12..acd789f4 100644 --- a/certora_autosetup/solidity_ast/__init__.py +++ b/certora_autosetup/solidity_ast/__init__.py @@ -18,6 +18,9 @@ "TypeDescriptions", "Visibility", "StateMutability", "Mutability", "StorageLocation", # loader "AstDump", "FileAsts", "SourceAst", "iter_nodes_of_type", "stream_raw_units", + # contracts + "ContractDeclView", "iter_contract_declarations", "parse_only_declarations", + "unit_contract_declarations", "to_contract_kind", # traversal "iter_children", "walk", "find_all", "build_node_index", "build_parent_map", "build_parent_graph_json", @@ -104,6 +107,13 @@ TupleExpression, UnaryOperation, ) +from .contracts import ( + ContractDeclView, + iter_contract_declarations, + parse_only_declarations, + to_contract_kind, + unit_contract_declarations, +) from .loader import AstDump, FileAsts, SourceAst, iter_nodes_of_type, stream_raw_units from .statements import ( Block, diff --git a/certora_autosetup/solidity_ast/contracts.py b/certora_autosetup/solidity_ast/contracts.py new file mode 100644 index 00000000..c8d9037a --- /dev/null +++ b/certora_autosetup/solidity_ast/contracts.py @@ -0,0 +1,79 @@ +"""Contract declarations and their kind, from either shape of solc AST. + +:class:`ContractDeclView` is the uniform view. :func:`iter_contract_declarations` reads it +from the typed models over an ``AstDump`` stream. :func:`parse_only_declarations` reads it +from the raw nodes of a ``stopAfter: "parsing"`` AST, which the typed models reject: solc +emits no analysis-phase fields (``scope``, ``linearizedBaseContracts``, +``fullyImplemented``) at that stage. +""" + +from dataclasses import dataclass +from typing import Any, Iterable, Iterator, Optional + +from certora_autosetup.utils.types import ContractKind + +from .declarations import ContractDefinition +from .loader import FileAsts, SourceAst, iter_nodes_of_type + + +@dataclass(frozen=True) +class ContractDeclView: + """Uniform view of a ContractDefinition for declaration/inheritance scans, whether + it came from the typed AST or from raw nodes the models could not reach.""" + + source_path: str + node_id: Optional[int] + name: str + abstract: bool + contract_kind: ContractKind + linearized_base_ids: list[int] + + +def to_contract_kind(value: Any) -> ContractKind: + """``contractKind`` as an enum. Anything solc did not give us is UNKNOWN.""" + return ContractKind(value) if value in {kind.value for kind in ContractKind} else ContractKind.UNKNOWN + + +def iter_contract_declarations(units: Iterable[FileAsts]) -> Iterator[ContractDeclView]: + """Every contract declaration in a stream of compilation units (typically + ``AstDump.stream_units(...)``, so the multi-GB dump is never fully in memory): + typed where the models parsed, completed by a raw flat-map sweep for anything + the typed walk could not reach (a solc surprise cannot hide contracts from + setup; Vyper sources contribute nothing — no ContractDefinition nodes).""" + for file_asts in units: + for source in file_asts.sources.values(): + yield from unit_contract_declarations(source) + + +def unit_contract_declarations(source: SourceAst) -> Iterator[ContractDeclView]: + for node in iter_nodes_of_type(source, ContractDefinition): + yield _view(node, source.source_path) + + +def parse_only_declarations(ast: dict, source_path: str) -> Iterator[ContractDeclView]: + """Declarations in a ``stopAfter: "parsing"`` AST node, which stays raw (see module + docstring). Solidity declares contracts only at file scope, so the top level is all + of them.""" + for node in ast.get("nodes", []): + if isinstance(node, dict) and node.get("nodeType") == "ContractDefinition": + yield _view(node, source_path) + + +def _view(node: ContractDefinition | dict[str, Any], source_path: str) -> ContractDeclView: + if isinstance(node, ContractDefinition): + return ContractDeclView( + source_path=source_path, + node_id=node.id, + name=node.name, + abstract=node.abstract, + contract_kind=to_contract_kind(node.contractKind), + linearized_base_ids=list(node.linearizedBaseContracts), + ) + return ContractDeclView( + source_path=source_path, + node_id=node.get("id"), + name=node.get("name") or "", + abstract=bool(node.get("abstract", False)), + contract_kind=to_contract_kind(node.get("contractKind")), + linearized_base_ids=[i for i in node.get("linearizedBaseContracts", []) if isinstance(i, int)], + ) diff --git a/certora_autosetup/utils/build_json.py b/certora_autosetup/utils/build_json.py new file mode 100644 index 00000000..7e989b1f --- /dev/null +++ b/certora_autosetup/utils/build_json.py @@ -0,0 +1,53 @@ +"""Reading ``.certora_build.json``, certoraRun's record of what it actually compiled. + +The file is keyed by compilation unit; each unit holds a ``contracts`` list, and a +contract reached through several units appears once per unit. +""" + +import json +from pathlib import Path +from typing import Any, Dict, Iterator, Optional + +from certora_autosetup.utils.constants import DIR_CERTORA_INTERNAL + +BUILD_JSON_NAME = ".certora_build.json" + +#: Where certoraRun writes it under the run directory it reports as ``latest``. +BUILD_JSON_RELPATH = Path(DIR_CERTORA_INTERNAL) / "latest" / BUILD_JSON_NAME + + +def build_json_path(project_root: Path) -> Optional[Path]: + """The build json of the most recent run under ``project_root``, or None.""" + latest = project_root / BUILD_JSON_RELPATH + if latest.exists(): + return latest + # `latest` is normally a symlink to the timestamped run dir; fall back to the + # newest run dir by name (they sort chronologically) if it is absent. + candidates = sorted(project_root.glob(f"{DIR_CERTORA_INTERNAL}/*/{BUILD_JSON_NAME}")) + return candidates[-1] if candidates else None + + +def load_build_json(path: Path) -> Dict[str, Any]: + return json.loads(path.read_text()) + + +def iter_contracts(build_data: Dict[str, Any]) -> Iterator[Dict[str, Any]]: + """Every contract record across all compilation units in the build. + + A contract reached through several units appears once per unit; callers that need + a single record must disambiguate themselves. + """ + for obj in build_data.values(): + if isinstance(obj, dict): + for contract in obj.get("contracts", []): + if isinstance(contract, dict): + yield contract + + +def contract_source_file(contract: Dict[str, Any]) -> str: + """Where the contract was written, preferring the pre-instrumentation path. + + ``file`` points into the instrumented tree certora compiles; ``original_file`` is + the project path the user knows, and is what a conf entry has to name. + """ + return contract.get("original_file") or contract.get("file") or "" diff --git a/certora_autosetup/utils/contract_linker.py b/certora_autosetup/utils/contract_linker.py index c099f260..d0c2aa52 100644 --- a/certora_autosetup/utils/contract_linker.py +++ b/certora_autosetup/utils/contract_linker.py @@ -33,22 +33,29 @@ class ContractLink: def render_wrapper_contract( harness_name: str, - parent_name: str, + parent_name: Optional[str], pragma_line: str, import_lines: List[str], ctor_forward: Optional[Tuple[str, List[str]]], body_blocks: Optional[List[str]] = None, header_comment_lines: Optional[List[str]] = None, ) -> str: - """Render the source of a ``contract is `` wrapper. + """Render the source of a ``contract `` wrapper. - Emits the SPDX header, the pragma (omitted when empty), the import lines, - an optional constructor forwarding to the parent, and optional extra body blocks. + Emits the SPDX header, the pragma (omitted when empty), the import lines, an + optional constructor forwarding to the parent, and optional extra body blocks. + + ``parent_name`` names the contract to inherit from; None emits a standalone + ``contract {`` — a library harness holds the library at arm's + length (a library cannot be a base contract) rather than extending it. ``ctor_forward`` is a ``(params_source, arg_names)`` pair; None means the parent needs no constructor arguments and the implicit default constructor - suffices. + suffices. It requires a ``parent_name`` to forward to. """ + if ctor_forward is not None and parent_name is None: + raise ValueError("ctor_forward requires a parent_name to forward to") + body_parts: List[str] = [] if ctor_forward is not None: params_src, arg_names = ctor_forward @@ -57,6 +64,11 @@ def render_wrapper_contract( ) body_parts.extend(body_blocks or []) + declaration = ( + f"contract {harness_name} {{" if parent_name is None + else f"contract {harness_name} is {parent_name} {{" + ) + lines = [ "// SPDX-License-Identifier: UNLICENSED", *([pragma_line] if pragma_line else []), @@ -64,7 +76,7 @@ def render_wrapper_contract( *import_lines, "", *(header_comment_lines or []), - f"contract {harness_name} is {parent_name} {{", + declaration, *body_parts, "}", "", diff --git a/certora_autosetup/utils/cvl_keywords.py b/certora_autosetup/utils/cvl_keywords.py new file mode 100644 index 00000000..c6a64ce6 --- /dev/null +++ b/certora_autosetup/utils/cvl_keywords.py @@ -0,0 +1,50 @@ +"""CVL grammar keywords, and the escape for a Solidity name that collides with one. + +Anything we generate for a spec — a summary's parameter names, a harness wrapper's function +names — has to be nameable in CVL. The grammar reserves words that are perfectly legal +Solidity identifiers, and a mechanically-generated name hits them routinely: `at` appears 41 +times across OpenZeppelin and Solady, `sort` 4. Such a name compiles as Solidity and then makes +the spec unparseable. + +The escape is a trailing underscore, matching OpenZeppelin's hand-written Certora harness +(`at_`), so generated specs read like the human-written ones. +""" + +from typing import FrozenSet + +#: Keyword terminals that cannot double as an identifier. A name equal to one of these is lexed +#: as the keyword inside a methods{} entry, which is a syntax error. +CVL_RESERVED_WORDS: FrozenSet[str] = frozenset({ + "ALL", "ALWAYS", "ASSERT_FALSE", "AUTO", "CONSTANT", "Create", "DELETE", "DISPATCH", "DISPATCHER", + "HAVOC_ALL", "HAVOC_ECF", "NONDET", "PER_CALLEE_CONSTANT", "STORAGE", "Sload", "Sstore", "Tload", + "Tstore", "UNRESOLVED", "assert", "assuming", "at", "axiom", "default", "definition", "else", + "event", "expect", "fallback", "false", "filtered", "function", "ghost", "good_description", + "havoc", "if", "in", "indexed", "lastReverted", "lastStorage", "links", "mapping", "methods", + "new", "norevert", "persistent", "require", "requireInvariant", "reset_storage", "return", + "returns", "revert", "rule", "satisfy", "sort", "true", "void", "with", "withrevert", "xor", +}) + +#: The terminals under the `usable_keywords` production in cvl.cup. The grammar accepts these +#: wherever an identifier is expected, so a name equal to one of them parses fine and must NOT +#: be mangled — renaming it would only produce a spec that reads worse. +#: +#: Verified rather than assumed: a contract with functions named `exists`, `sum`, `old`, `forall` +#: and `invariant`, declared in a methods block and called from a rule, typechecks; the same +#: shape with `havoc` or `rule` is a syntax error. +#: +#: Note uppercase "UNRESOLVED" is a distinct summary keyword and stays reserved above. +CVL_USABLE_KEYWORDS: FrozenSet[str] = frozenset({ + "as", "builtin", "description", "exists", "forall", "hook", "import", "invariant", + "old", "onTransactionBoundary", "override", "preserved", "sig", "strong", "sum", + "unresolved", "use", "using", "usum", "weak", +}) + + +def escape_reserved(name: str) -> str: + """Rename `name` if CVL reserves it, else return it unchanged. + + Apply this before collision mangling, not after: renaming afterwards could turn a distinct + name into one already taken — a library declaring both `at` and `at_`, and `at_` is in + active use in OpenZeppelin's own harness. + """ + return f"{name}_" if name in CVL_RESERVED_WORDS else name diff --git a/certora_autosetup/utils/enhanced_config_manager.py b/certora_autosetup/utils/enhanced_config_manager.py index 06b7aae6..e0712ad2 100644 --- a/certora_autosetup/utils/enhanced_config_manager.py +++ b/certora_autosetup/utils/enhanced_config_manager.py @@ -13,14 +13,14 @@ from collections import Counter from dataclasses import dataclass from pathlib import Path -from typing import Any, Dict, Generic, List, Optional, TypeVar +from typing import Any, Dict, Generic, List, Optional, Set, Tuple, TypeVar from packaging.version import Version from certora_autosetup.parsers.spec_imports import parse_imports_from_spec from certora_autosetup.utils.config_manager import certora_format_to_raw_version from certora_autosetup.utils.constants import DEFAULT_SOLC_VERSION, SolcConvention -from certora_autosetup.utils.contract_utils import parse_contract_files +from certora_autosetup.utils.contract_utils import parse_contract_files, split_contract_spec from certora_autosetup.utils.logger import logger from certora_autosetup.utils.solc_version_resolver import ( parse_pragma_constraint, @@ -238,7 +238,19 @@ def create_config( # Add contract files (use normalized paths) TODO: what about the additional files? we expect them to come already normalized now normalized = self.normalize_paths(contract_handles) - conf_template["files"] = [c.to_config_str() for c in normalized] + additional_files + # The two sources legitimately overlap — a contract can be both in the scene and named + # as an additional contract — and certoraRun rejects a `files` list with a repeat. + # Deduped on the (file, contract) pair rather than the string, since to_config_str + # drops the name when it matches the file stem and gives one contract two spellings. + seen: Set[Tuple[str, str]] = set() + files: List[str] = [] + for entry in [c.to_config_str() for c in normalized] + additional_files: + key = split_contract_spec(entry) + if key in seen: + continue + seen.add(key) + files.append(entry) + conf_template["files"] = files # Set verification target (use relative path for spec file) normalized_spec = self._normalize_path(spec_file, context="Spec file") diff --git a/certora_autosetup/utils/paths.py b/certora_autosetup/utils/paths.py index 930be449..e712308a 100644 --- a/certora_autosetup/utils/paths.py +++ b/certora_autosetup/utils/paths.py @@ -144,3 +144,19 @@ def resolve_autosetup_prover_usage_file(project_root: Path) -> Path | None: """Locate the ``prover_usage.json`` the most recent autosetup run wrote under ``project_root`` (``None`` if absent). See :func:`_resolve_autosetup_reports_file`.""" return _resolve_autosetup_reports_file(project_root, FILE_PROVER_USAGE) + + +def same_source_file(candidate: str, wanted: str) -> bool: + """Whether two build-reported paths name the same source file. + + The build reports the same file as a project-relative path, as an absolute one, or + under the instrumented ``.certora_sources`` copy, depending on how it was reached, so + equality is decided on the longest common suffix of path components. Only the trailing + components are compared, which is what makes every one of those prefixes harmless. + """ + if not candidate or not wanted: + return False + cand_parts = Path(candidate).parts + want_parts = Path(wanted).parts + depth = min(len(cand_parts), len(want_parts)) + return cand_parts[-depth:] == want_parts[-depth:] diff --git a/composer/pipeline/cli.py b/composer/pipeline/cli.py index 406f1ff5..b611a8e8 100644 --- a/composer/pipeline/cli.py +++ b/composer/pipeline/cli.py @@ -50,6 +50,7 @@ if TYPE_CHECKING: from sentence_transformers import SentenceTransformer +from certora_autosetup.harnesser.swap import swap_library_main_contract_paths from composer.spec.util import fs_forbidden_read import hashlib @@ -292,6 +293,15 @@ async def cli_pipeline[P: enum.Enum, H]( relative_path = str(full_contract_path.relative_to(project_root)) + # The Prover instantiates no parametric methods against a library, so verifying one + # directly comes back vacuous with no error. Swap in a generated harness here, ahead + # of SourceFields: every later phase — component analysis, CVL authoring, the conf's + # verify target — has to agree on one contract name, and AutoSetup keys its results + # by that name too. + relative_path, contract_name = swap_library_main_contract_paths( + project_root, relative_path, contract_name + ) + # Set up services tiered = get_provider_for(tiered=args) diff --git a/composer/spec/source/harness.py b/composer/spec/source/harness.py index 4270276a..b881a1d7 100644 --- a/composer/spec/source/harness.py +++ b/composer/spec/source/harness.py @@ -37,6 +37,7 @@ from composer.diagnostics.timing import get_run_summary from composer.spec.graph_builder import run_to_completion, bind_standard +from certora_autosetup.harnesser.run import is_generated_library_harness from composer.spec.source.autosetup import run_autosetup, read_autosetup_usage, read_autosetup_prover_usage, SetupFailure, SetupSuccess from composer.spec.service_host import ServiceHost from composer.spec.context import WorkflowContext, SourceCode, CacheKey @@ -657,6 +658,14 @@ async def run_and_apply_part1( for c in res.transitive_closure: if c.harness_definition is not None: tgt = Path(source.project_root) / c.path + if is_generated_library_harness(tgt): + # The library harness is the verification target itself, generated + # mechanically from the library's compiled API. Overwriting it with an + # LLM-authored wrapper would replace the contract the whole run is about. + _logger.warning( + "Refusing to overwrite generated library harness %s", c.path + ) + continue tgt.parent.mkdir(parents=True, exist_ok=True) tgt.write_text(c.harness_definition.harness_source) return res diff --git a/composer/spec/source/munge/compile_check.py b/composer/spec/source/munge/compile_check.py index 4f5fd3c4..e30b9929 100644 --- a/composer/spec/source/munge/compile_check.py +++ b/composer/spec/source/munge/compile_check.py @@ -14,6 +14,7 @@ from typing import Any from graphcore.tools.vfs import VFSState, VFSAccessor +from certora_autosetup.utils.build_json import build_json_path from composer.prover.core import BUILD_TIMEOUT_S, run_prover_inner @@ -48,16 +49,6 @@ def _config_paths(config: dict[str, Any]) -> set[str]: return {str(entry).split(":", 1)[0] for entry in config.get("files", [])} -def _find_build_json(folder: Path) -> Path | None: - latest = folder / ".certora_internal" / "latest" / ".certora_build.json" - if latest.exists(): - return latest - # `latest` is normally a symlink to the timestamped run dir; fall back to the - # newest run dir by name (they sort chronologically) if it's absent. - candidates = sorted(folder.glob(".certora_internal/*/.certora_build.json")) - return candidates[-1] if candidates else None - - def _scrape_touched(build_json: Path) -> set[str]: """Union the ``srclist`` values across every SDC in ``.certora_build.json``. Each srclist is solc's ``sources`` map for one compilation unit — the input @@ -142,7 +133,7 @@ async def check_edits_compile( if isinstance(result, dict) and result.get("sort") == "failure": return BuildFailed(reason=f"{result.get('exc_str', '')}\n{stdout}".strip()) - build_json = _find_build_json(folder) + build_json = build_json_path(folder) if build_json is None: return BuildFailed(reason=f"build produced no .certora_build.json\n{stdout}".strip())