support natspec preconditions

.github/workflows/test-pr.yml

@@ -63,6 +63,7 @@ jobs:
       - name: 'Build Kontrol'
         run: |
             docker exec -u github-user kontrol-ci-profile-${GITHUB_SHA} /bin/bash -c 'CXX=clang++-14 uv run kdist --verbose build -j`nproc` kontrol.*'
+            docker exec -u github-user kontrol-ci-profile-${GITHUB_SHA} /bin/bash -c 'CXX=clang++-14 uv run kdist --verbose build -j`nproc` natspec.llvm'
       - name: 'Run profiling'
         run: |
           PROF_ARGS='--numprocesses=8'
@@ -89,6 +90,7 @@ jobs:
       - name: 'Build Kontrol'
         run: |
             docker exec -u github-user kontrol-ci-integration-${GITHUB_SHA} /bin/bash -c 'CXX=clang++-14 uv run kdist --verbose build -j`nproc` kontrol.*'
+            docker exec -u github-user kontrol-ci-integration-${GITHUB_SHA} /bin/bash -c 'CXX=clang++-14 uv run kdist --verbose build -j`nproc` natspec.llvm'
       - name: 'Run integration tests'
         run: |
           TEST_ARGS='-vv --force-sequential -k "not (test_kontrol_cse or test_foundry_minimize_proof or test_kontrol_end_to_end)" --numprocesses=3'
@@ -115,6 +117,7 @@ jobs:
       - name: 'Build Kontrol'
         run: |
             docker exec -u github-user kontrol-ci-integration-${GITHUB_SHA} /bin/bash -c 'CXX=clang++-14 uv run kdist --verbose build -j`nproc` kontrol.*'
+            docker exec -u github-user kontrol-ci-integration-${GITHUB_SHA} /bin/bash -c 'CXX=clang++-14 uv run kdist --verbose build -j`nproc` natspec.llvm'
       - name: 'Run CSE and Minimize tests'
         run: |
           TEST_ARGS='--numprocesses=5 --force-sequential -vv -k "test_kontrol_cse or test_foundry_minimize_proof"'
@@ -141,6 +144,7 @@ jobs:
       - name: 'Build Kontrol'
         run: |
             docker exec -u github-user kontrol-ci-integration-${GITHUB_SHA} /bin/bash -c 'CXX=clang++-14 uv run kdist --verbose build -j`nproc` kontrol.*'
+            docker exec -u github-user kontrol-ci-integration-${GITHUB_SHA} /bin/bash -c 'CXX=clang++-14 uv run kdist --verbose build -j`nproc` natspec.llvm'
       - name: 'Run end-to-end tests'
         run: |
           TEST_ARGS='--numprocesses=6 -vv --force-sequential -k "test_kontrol_end_to_end"'

pyproject.toml

@@ -19,12 +19,16 @@ dependencies = [
 name = "Runtime Verification, Inc."
 email = "[email protected]"
 
+[tool.hatch.build.targets.wheel]
+packages = ["src/kontrol", "src/natspec"]
+
 [project.scripts]
 kontrol = "kontrol.__main__:main"
 kontrol-kdist = "pyk.kdist.__main__:main"
 
 [project.entry-points.kdist]
 kontrol = "kontrol.kdist.plugin"
+natspec = "natspec.kdist.plugin"
 
 [dependency-groups]
 dev = [
@@ -68,4 +72,4 @@ skip-string-normalization = true
 
 [tool.mypy]
 disallow_untyped_defs = true
-exclude = "src/tests/integration/test-data"
+exclude = "src/tests/integration/test-data"

src/kontrol/prove.py

@@ -15,7 +15,7 @@
 from kevm_pyk.utils import KDefinition__expand_macros, abstract_cell_vars, run_prover
 from multiprocess.pool import Pool  # type: ignore
 from pyk.cterm import CTerm, CTermSymbolic
-from pyk.kast.inner import KApply, KSequence, KSort, KVariable, Subst
+from pyk.kast.inner import KApply, KSequence, KSort, KToken, KVariable, Subst
 from pyk.kast.manip import flatten_label, free_vars, set_cell
 from pyk.kast.prelude.bytes import bytesToken
 from pyk.kast.prelude.collections import list_empty, map_empty, map_item, set_empty
@@ -31,9 +31,12 @@
 from pyk.proof import ProofStatus
 from pyk.proof.proof import Proof
 from pyk.proof.reachability import APRFailureInfo, APRProof
-from pyk.utils import hash_str, run_process_2, unique
+from pyk.utils import hash_str, run_process_2, single, unique
 from rich.progress import Progress, SpinnerColumn, TaskID, TextColumn, TimeElapsedColumn
 
+from natspec.natspec import Natspec
+from natspec.utils import GLOBAL_VARIABLES_TO_CELL_NAMES, HEX_LITERAL, ID, NATSPEC_TO_K_OPERATORS
+
 from .foundry import Foundry, KontrolSemantics, foundry_to_xml
 from .options import ConfigType
 from .solc_to_k import Contract, decode_kinner_output
@@ -812,6 +815,7 @@ def _method_to_cfg(
 
         for final_node in final_states:
             new_init_cterm = _update_cterm_from_node(init_cterm, final_node, config_type, keep_vars)
+            new_init_cterm = add_natspec_preconditions(new_init_cterm, method, contract, foundry)
             new_node = cfg.create_node(new_init_cterm)
             if graft_setup_proof:
                 cfg.create_edge(final_node.id, new_node.id, depth=1)
@@ -821,6 +825,7 @@ def _method_to_cfg(
                 )
             new_node_ids.append(new_node.id)
     else:
+        init_cterm = add_natspec_preconditions(init_cterm, method, contract, foundry)
         cfg = KCFG()
         init_node = cfg.create_node(init_cterm)
         new_node_ids = [init_node.id]
@@ -1486,3 +1491,100 @@ def _interpret_proof_failure(
         log = failure_log.print_with_additional_info(status_codes, output_values) + Foundry.help_info()
         for line in log:
             print(replace_k_words(line))
+
+
+def add_natspec_preconditions(
+    cterm: CTerm, method: Contract.Method | Contract.Constructor, contract: Contract, foundry: Foundry
+) -> CTerm:
+    if type(method) is Contract.Method:
+        natspec_preconditions = _create_precondition_constraints(method, cterm, contract)
+
+        for precondition in natspec_preconditions:
+            _LOGGER.info(f'Adding NatSpec precondition: {foundry.kevm.pretty_print(mlEqualsTrue(precondition))}')
+            cterm = cterm.add_constraint(mlEqualsTrue(precondition))
+    return cterm
+
+
+def _create_precondition_constraints(method: Contract.Method, init_cterm: CTerm, contract: Contract) -> list[KApply]:
+    precondition_constraints: list[KApply] = []
+
+    if method.preconditions is not None:
+        natspec = Natspec(kdist.get('natspec.llvm'))
+        for p in method.preconditions:
+            kinner = natspec.decode(input=p.precondition)
+            if type(kinner) is KApply:
+                kontrol_precondition = _kontrol_kast_from_natspec(kinner, method, init_cterm, contract)
+                precondition_constraints.append(kontrol_precondition)
+
+    return precondition_constraints
+
+
+def _kontrol_kast_from_natspec(term: KApply, method: Contract.Method, init_cterm: CTerm, contract: Contract) -> KApply:
+    """Convert a KApply representing a Precondition defined using the Natspec-Grammar to a Kontrol KApply.
+
+    Replace symbol sorts and Id elements with KVariables.
+    """
+    symbol = NATSPEC_TO_K_OPERATORS[term.label]
+    args: list[KInner] = []
+    for arg in term.args:
+        if type(arg) is KApply:
+            # Check for predefined `block` or `msg` value.
+            global_variable = get_global_variable(arg, init_cterm)
+            if global_variable:
+                args.append(global_variable)
+            else:
+                # TODO: Handle IndexAccess and FieldAccess
+                args.append(_kontrol_kast_from_natspec(arg, method, init_cterm, contract))
+        if type(arg) is KToken:
+            if arg.sort == ID:
+                # If it's an Id token, ensure it's either a function parameter or a storage variable
+                function_arg = method.find_arg(arg.token)
+                if function_arg:
+                    args.append(KVariable(function_arg))
+                    continue
+                # Identify storage slot and offset
+                (storage_slot, _slot_offset) = contract.get_storage_slot_by_name(arg.token)
+                target_address = init_cterm.cell('ID_CELL')
+                if storage_slot:
+                    storage_map = _get_account_storage_by_address(init_cterm, target_address)
+                    if storage_map is None:
+                        continue
+                    # TODO: Apply the _slot_offset on the Map lookup.
+                    args.append(KEVM.lookup(storage_map, token(storage_slot)))
+                    continue
+
+                _LOGGER.warning(f'NatSpec precondition: Unknown Id sort element: {arg.token}')
+            elif arg.sort == HEX_LITERAL:
+                args.append(token(int(arg.token, 16)))
+                continue
+            elif arg.sort in [KSort('Int'), KSort('Bool')]:
+                args.append(arg)
+                continue
+            _LOGGER.warning(f'NatSpec precondition: unknown token: {arg.token}')
+    return KApply(symbol, args)
+
+
+def get_global_variable(term: KApply, init_cterm: CTerm) -> KInner | None:
+    """Checks if a KAst term is a Solidity global variable and return the cell value from init_cterm if so."""
+    cell_name = GLOBAL_VARIABLES_TO_CELL_NAMES.get(term)
+    if cell_name is None:
+        return None
+    return init_cterm.cell(cell_name)
+
+
+def _get_account_storage_by_address(cterm: CTerm, target_address: KInner) -> KInner | None:
+    accounts_cell = flatten_label('_AccountCellMap_', cterm.cell('ACCOUNTS_CELL'))
+    for account_wrapped in accounts_cell:
+        assert type(account_wrapped) is KApply
+        acct_id_cell = account_wrapped.terms[0]
+        assert type(acct_id_cell) is KApply and acct_id_cell.label.name == '<acctID>'
+        account_address = single(acct_id_cell.terms)
+        if target_address == account_address:
+            account_cell = account_wrapped.terms[1]
+            assert type(account_cell) is KApply and account_cell.label.name == '<account>'
+            storage_cell = account_cell.terms[3]
+            assert type(storage_cell) is KApply and storage_cell.label.name == '<storage>'
+            return single(storage_cell.terms)
+
+    _LOGGER.warning(f'Account {target_address} not found in the state.')
+    return None

src/kontrol/solc_to_k.py

@@ -342,6 +342,32 @@ def parse_devdoc(tag: str, devdoc: dict | None) -> dict:
     return natspecs
 
 
+def parse_annotations(
+    devdoc: str | None, method: Contract.Method | Contract.Constructor
+) -> tuple[Precondition, ...] | None:
+    """
+    Parse developer documentation (devdoc) to extract user-provided preconditions.
+
+    Returns a tuple of Precondition objects, each representing a single precondition and a method to which it belongs.
+    """
+
+    if devdoc is None:
+        return None
+
+    preconditions: list[Precondition] = []
+    for precondition in devdoc.split(','):
+        # Trim whitespace and skip if empty
+        precondition = precondition.strip()
+        if not precondition:
+            continue
+
+        # Create a Precondition object and add it to the list
+        new_precondition = Precondition(precondition, method)
+        preconditions.append(new_precondition)
+
+    return tuple(preconditions)
+
+
 class StorageField(NamedTuple):
     label: str
     data_type: str
@@ -350,6 +376,19 @@ class StorageField(NamedTuple):
     linked_interface: str | None
 
 
+@dataclass
+class Precondition:
+    precondition: str
+    method: Contract.Method | Contract.Constructor
+
+    def __init__(self, precondition: str, method: Contract.Method | Contract.Constructor):
+        """
+        Initializes a new instance of the Precondition class.
+        """
+        self.precondition = precondition
+        self.method = method
+
+
 @dataclass
 class Contract:
     @dataclass
@@ -361,6 +400,7 @@ class Constructor:
         contract_storage_digest: str
         payable: bool
         signature: str
+        preconditions: tuple[Precondition, ...] | None
 
         def __init__(
             self,
@@ -476,6 +516,7 @@ class Method:
         ast: dict | None
         natspec_values: dict | None
         function_calls: tuple[str, ...] | None
+        preconditions: tuple[Precondition, ...] | None
 
         def __init__(
             self,
@@ -506,6 +547,9 @@ def __init__(
             natspec_tags = ['custom:kontrol-array-length-equals', 'custom:kontrol-bytes-length-equals']
             self.natspec_values = {tag.split(':')[1]: parse_devdoc(tag, devdoc) for tag in natspec_tags}
             self.inputs = tuple(inputs_from_abi(abi['inputs'], self.natspec_values))
+            self.preconditions = (
+                parse_annotations(devdoc.get('custom:kontrol-precondition', None), self) if devdoc is not None else None
+            )
             self.function_calls = tuple(function_calls) if function_calls is not None else None
 
         @property
@@ -571,6 +615,13 @@ def arg_types(self) -> tuple[str, ...]:
                     arg_types.extend([sub_input.type for sub_input in input.flattened()])
             return tuple(arg_types)
 
+        def find_arg(self, name: str) -> str | None:
+            try:
+                idx = single([i.idx for i in self.inputs if i.name == name])
+                return self.arg_names[idx]
+            except ValueError:
+                return None
+
         def up_to_date(self, digest_file: Path) -> bool:
             digest_dict = _read_digest_file(digest_file)
             return digest_dict.get('methods', {}).get(self.qualified_name, {}).get('method', '') == self.digest
@@ -907,6 +958,12 @@ def method_by_name(self) -> dict[str, Contract.Method]:
     def method_by_sig(self) -> dict[str, Contract.Method]:
         return {method.signature: method for method in self.methods}
 
+    def get_storage_slot_by_name(self, target_name: str) -> tuple[int, int] | tuple[None, None]:
+        for field in self.fields:
+            if field.label == target_name:
+                return (field.slot, field.offset)
+        return (None, None)
+
 
 # Helpers
 

src/natspec/kdist/natspec-grammar.md

@@ -0,0 +1,60 @@
+Solidity Natspec Grammar
+------------------------
+
+
+ TODO:
+ ----
+ 1. add support for bitwise operators
+
+```k
+module NATSPEC-SYNTAX
+    imports INT-SYNTAX
+    imports BOOL-SYNTAX
+    imports ID-SYNTAX
+
+    // Literals
+    syntax HexLiteral ::= r"0x[0-9a-fA-F]+" [token, symbol(SolidityHexLiteral)]
+ // ---------------------------------------------------------------------------
+
+    syntax Access ::= Id
+                    | Access "[" Exp "]" [symbol(SolidityIndexAccess)]
+                    | Access "." Id      [symbol(SolidityFieldAccess)]
+ // ------------------------------------------------------------------
+
+    syntax Exp ::= Int | Bool | HexLiteral | Access
+                 | "(" Exp ")"                   [bracket]
+                 // Unary operators (high precedence)
+                 > "!" Exp                       [symbol(SolidityNegation)]
+                 // Power (right associative)
+                 > right:
+                   Exp "**" Exp                  [symbol(SolidityPower)]
+                 // Multiplicative
+                 > left:
+                   Exp "*" Exp                   [symbol(SolidityMultiplication)]
+                 | Exp "/" Exp                   [symbol(SolidityDivision)]
+                 | Exp "%" Exp                   [symbol(SolidityModulo)]
+                 // Additive
+                 > left:
+                   Exp "+" Exp                   [symbol(SolidityAddition)]
+                 | Exp "-" Exp                   [symbol(SoliditySubtraction)]
+                 // Relational
+                 > non-assoc:
+                   Exp "<" Exp                   [symbol(SolidityLT)]
+                 | Exp ">" Exp                   [symbol(SolidityGT)]
+                 | Exp "<=" Exp                  [symbol(SolidityLE)]
+                 | Exp ">=" Exp                  [symbol(SolidityGE)]
+                 // Equality
+                 > non-assoc:
+                   Exp "==" Exp                  [symbol(SolidityEq)]
+                 | Exp "!=" Exp                  [symbol(SolidityNeq)]
+                 // Logical AND
+                 > left:
+                   Exp "&&" Exp                  [symbol(SolidityConjunction)]
+                 // Logical OR
+                 > left:
+                   Exp "||" Exp                  [symbol(SolidityDisjunction)]
+endmodule
+
+module NATSPEC
+endmodule
+```