Construct KCFGs for Foundry proof obligations

kevm-pyk/src/kevm_pyk/__main__.py

@@ -8,12 +8,14 @@
 
 from pyk.cli_utils import dir_path, file_path
 from pyk.kast import KApply, KDefinition, KFlatModule, KImport, KRequire, KSort
+from pyk.kcfg import KCFG
 from pyk.ktool.krun import _krun
+from pyk.prelude import mlTop
 
 from .gst_to_kore import gst_to_kore
 from .kevm import KEVM
 from .solc_to_k import Contract, contract_to_k, solc_compile
-from .utils import KPrint_make_unparsing, add_include_arg
+from .utils import KCFG_from_claim, KPrint_make_unparsing, add_include_arg, read_kast_flatmodulelist
 
 _LOGGER: Final = logging.getLogger(__name__)
 _LOG_FORMAT: Final = '%(levelname)s %(asctime)s %(name)s - %(message)s'
@@ -177,6 +179,8 @@ def exec_foundry_kompile(
     md_selector: Optional[str],
     regen: bool = False,
     rekompile: bool = False,
+    reparse: bool = False,
+    reinit: bool = False,
     requires: Iterable[str] = (),
     imports: Iterable[str] = (),
     **kwargs,
@@ -189,6 +193,9 @@ def exec_foundry_kompile(
     spec_module = 'FOUNDRY-SPEC'
     foundry_definition_dir = foundry_out / 'kompiled'
     foundry_main_file = foundry_definition_dir / 'foundry.k'
+    kompiled_timestamp = foundry_definition_dir / 'timestamp'
+    parsed_spec = foundry_definition_dir / 'spec.json'
+    kcfgs_file = foundry_definition_dir / 'kcfgs.json'
     requires = ['lemmas/lemmas.k', 'lemmas/int-simplification.k'] + list(requires)
     imports = ['LEMMAS', 'INT-SIMPLIFICATION'] + list(imports)
     if not foundry_definition_dir.exists():
@@ -205,7 +212,7 @@ def exec_foundry_kompile(
                 imports=list(imports),
                 output=fmf,
             )
-    if regen or rekompile or not (foundry_definition_dir / 'timestamp').exists():
+    if regen or rekompile or not kompiled_timestamp.exists():
         KEVM.kompile(
             foundry_definition_dir,
             foundry_main_file,
@@ -216,6 +223,26 @@ def exec_foundry_kompile(
             md_selector=md_selector,
             profile=profile,
         )
+    kevm = KEVM(foundry_definition_dir, main_file=foundry_main_file, profile=profile)
+    if regen or rekompile or reparse or not parsed_spec.exists():
+        prove_args = add_include_arg(includes)
+        kevm.prove(
+            foundry_main_file,
+            spec_module_name=spec_module,
+            dry_run=True,
+            args=(['--emit-json-spec', str(parsed_spec)] + prove_args),
+        )
+    if regen or rekompile or reparse or reinit or not kcfgs_file.exists():
+        cfgs: Dict[str, Dict] = {}
+        for module in read_kast_flatmodulelist(parsed_spec).modules:
+            for claim in module.claims:
+                cfg_label = claim.att["label"]
+                _LOGGER.info(f'Producing KCFG: {cfg_label}')
+                cfgs[cfg_label] = KCFG_from_claim(kevm.definition, claim).to_dict()
+        with open(kcfgs_file, 'w') as kf:
+            kf.write(json.dumps(cfgs))
+            kf.close()
+            _LOGGER.info(f'Wrote file: {kcfgs_file}')
 
 
 def exec_prove(
@@ -267,23 +294,45 @@ def exec_foundry_prove(
     _ignore_arg(kwargs, 'definition_dir', f'--definition: {kwargs["definition_dir"]}')
     _ignore_arg(kwargs, 'spec_module', f'--spec-module: {kwargs["spec_module"]}')
     definition_dir = foundry_out / 'kompiled'
-    spec_file = definition_dir / 'foundry.k'
-    spec_module = 'FOUNDRY-SPEC'
-    claims = [Contract.contract_test_to_claim_id(_t) for _t in tests]
-    exclude_claims = [Contract.contract_test_to_claim_id(_t) for _t in exclude_tests]
-    exec_prove(
-        definition_dir,
-        profile,
-        spec_file,
-        includes=includes,
-        debug_equations=debug_equations,
-        bug_report=bug_report,
-        depth=depth,
-        claims=claims,
-        exclude_claims=exclude_claims,
-        spec_module=spec_module,
-        **kwargs,
-    )
+    use_directory = foundry_out / 'specs'
+    use_directory.mkdir(parents=True, exist_ok=True)
+    kevm = KEVM(definition_dir, profile=profile, use_directory=use_directory)
+    kcfgs_file = definition_dir / 'kcfgs.json'
+    kcfgs: Dict[str, KCFG] = {}
+    _LOGGER.info(f'Reading file: {kcfgs_file}')
+    with open(kcfgs_file, 'r') as kf:
+        kcfgs = {k: KCFG.from_dict(v) for k, v in json.loads(kf.read()).items()}
+    tests = [Contract.contract_test_to_claim_id(_t) for _t in tests]
+    exclude_tests = [Contract.contract_test_to_claim_id(_t) for _t in exclude_tests]
+    claims = list(kcfgs.keys())
+    _unfound_kcfgs: List[str] = []
+    if len(tests) > 0:
+        kcfgs = {k: kcfg for k, kcfg in kcfgs.items() if k in tests}
+    for _t in tests:
+        if _t not in claims:
+            _unfound_kcfgs.append(_t)
+    for _t in exclude_tests:
+        if _t not in claims:
+            _unfound_kcfgs.append(_t)
+        if _t in kcfgs:
+            kcfgs.pop(_t)
+    if _unfound_kcfgs:
+        _LOGGER.error(f'Missing KCFGs for tests: {_unfound_kcfgs}')
+        sys.exit(1)
+    _failed_claims: List[str] = []
+    for kcfg_name, kcfg in kcfgs.items():
+        _LOGGER.info(f'Proving KCFG: {kcfg_name}')
+        edge = kcfg.create_edge(kcfg.get_unique_init().id, kcfg.get_unique_target().id, mlTop(), depth=-1)
+        claim = edge.to_claim()
+        result = kevm.prove_claim(claim, kcfg_name.replace('.', '-'))
+        if type(result) is KApply and result.label.name == '#Top':
+            _LOGGER.info(f'Proved KCFG: {kcfg_name}')
+        else:
+            _LOGGER.error(f'Failed to prove KCFG: {kcfg_name}')
+            _failed_claims.append(kcfg_name)
+    if _failed_claims:
+        _LOGGER.error(f'Failed to prove KCFGs: {_failed_claims}')
+        sys.exit(1)
 
 
 def exec_run(
@@ -482,6 +531,20 @@ def parse(s):
         action='store_true',
         help='Rekompile foundry.k even if kompiled definition already exists.',
     )
+    foundry_kompile.add_argument(
+        '--reparse',
+        dest='reparse',
+        default=False,
+        action='store_true',
+        help='Reparse K specifications even if the parsed spec already exists.',
+    )
+    foundry_kompile.add_argument(
+        '--reinit',
+        dest='reinit',
+        default=False,
+        action='store_true',
+        help='Reinitialize kcfgs even if they already exist.',
+    )
 
     foundry_prove_args = command_parser.add_parser(
         'foundry-prove',

kevm-pyk/src/kevm_pyk/kevm.py

@@ -83,7 +83,7 @@ def _patch_symbol_table(symbol_table: Dict[str, Any]) -> None:
         symbol_table['_|->_']                                         = paren(symbol_table['_|->_'])
         symbol_table['_Map_']                                         = paren(lambda m1, m2: m1 + '\n' + m2)
         symbol_table['_AccountCellMap_']                              = paren(lambda a1, a2: a1 + '\n' + a2)
-        symbol_table['.AccountCellMap']                               = lambda: ''
+        symbol_table['.AccountCellMap']                               = lambda: '.Bag'
         symbol_table['AccountCellMapItem']                            = lambda k, v: v
         symbol_table['_[_:=_]_EVM-TYPES_Memory_Memory_Int_ByteArray'] = lambda m, k, v: m + ' [ '  + k + ' := (' + v + '):ByteArray ]'
         symbol_table['_[_.._]_EVM-TYPES_ByteArray_ByteArray_Int_Int'] = lambda m, s, w: '(' + m + ' [ ' + s + ' .. ' + w + ' ]):ByteArray'

kevm-pyk/src/kevm_pyk/utils.py

@@ -1,8 +1,56 @@
+from pathlib import Path
 from typing import Collection, Iterable, List
 
-from pyk.kast import KDefinition, KFlatModule, KImport, KTerminal, KVariable
-from pyk.kastManip import abstract_term_safely, is_anon_var, split_config_and_constraints, split_config_from, substitute
+from pyk.cfg_manager import instantiate_cell_vars, rename_generated_vars
+from pyk.cterm import CTerm
+from pyk.kast import KApply, KClaim, KDefinition, KFlatModule, KFlatModuleList, KImport, KInner, KVariable, read_kast
+from pyk.kastManip import (
+    abstract_term_safely,
+    bool_to_ml_pred,
+    bottom_up,
+    extract_lhs,
+    extract_rhs,
+    flatten_label,
+    free_vars,
+    is_anon_var,
+    ml_pred_to_bool,
+    remove_generated_cells,
+    split_config_and_constraints,
+    split_config_from,
+    substitute,
+    undo_aliases,
+)
+from pyk.kcfg import KCFG
 from pyk.ktool import KPrint
+from pyk.prelude import Bool, mlAnd
+
+
+def read_kast_flatmodulelist(ifile: Path) -> KFlatModuleList:
+    _flat_module_list = read_kast(ifile)
+    assert isinstance(_flat_module_list, KFlatModuleList)
+    return _flat_module_list
+
+
+def KCFG_from_claim(defn: KDefinition, claim: KClaim) -> KCFG:  # noqa: N802
+    def _make_cterm(_kinner: KInner, _cond: KInner) -> CTerm:
+        _kinner = _kinner if _cond == Bool.true else mlAnd([_kinner, bool_to_ml_pred(_cond)])
+        _kinner = sanitize_config(defn, _kinner)
+        return CTerm(_kinner)
+
+    cfg = KCFG()
+    claim_body = claim.body
+    claim_body = instantiate_cell_vars(defn, claim_body)
+    claim_body = rename_generated_vars(CTerm(claim_body)).kast
+
+    claim_lhs = _make_cterm(extract_lhs(claim_body), claim.requires)
+    init_node = cfg.create_node(claim_lhs)
+    cfg.add_init(init_node.id)
+
+    claim_rhs = _make_cterm(extract_rhs(claim_body), claim.ensures)
+    target_node = cfg.create_node(claim_rhs)
+    cfg.add_target(target_node.id)
+
+    return cfg
 
 
 def KPrint_make_unparsing(_self: KPrint, extra_modules: Iterable[KFlatModule] = ()) -> KPrint:  # noqa: N802
@@ -28,28 +76,33 @@ def abstract_cell_vars(cterm, keep_vars: Collection[KVariable] = ()):
     return substitute(config, subst)
 
 
-def get_productions(definition):
-    return [prod for module in definition.modules for prod in module.productions]
+def sanitize_config(defn: KDefinition, init_term: KInner) -> KInner:
+    def _var_name(vname):
+        new_vname = vname
+        while new_vname.startswith('_') or new_vname.startswith('?'):
+            new_vname = new_vname[1:]
+        return new_vname
+
+    free_vars_subst = {vname: KVariable(_var_name(vname)) for vname in free_vars(init_term)}
 
+    # TODO: This is somewhat hacky. We shouldn't have to touch the config this much.
+    # Likely, the frontend should just be giving us initial states with these already in place.
+    def _remove_cell_map_definedness(_kast):
+        if type(_kast) is KApply:
+            if _kast.label.name.endswith('CellMap:in_keys'):
+                return Bool.false
+            elif _kast.label.name.endswith('CellMapItem'):
+                return _kast.args[1]
+        return _kast
 
-def get_production_for_klabel(definition, klabel):
-    productions = [prod for prod in get_productions(definition) if prod.klabel and prod.klabel == klabel]
-    if len(productions) != 1:
-        raise ValueError(f'Expected 1 production for label {klabel}, not {productions}.')
-    return productions[0]
+    new_term = substitute(init_term, free_vars_subst)
+    new_term = remove_generated_cells(new_term)
+    new_term = bottom_up(_remove_cell_map_definedness, new_term)
 
+    if not (type(new_term) is KApply and new_term.label.name in ['#Top', '#Bottom']):
+        config, constraint = split_config_and_constraints(new_term)
+        constraints = [bool_to_ml_pred(ml_pred_to_bool(c, unsafe=True)) for c in flatten_label('#And', constraint)]
+        new_term = mlAnd([config] + constraints)
+        new_term = undo_aliases(defn, new_term)
 
-def get_label_for_cell_sorts(definition, sort):
-    productions = []
-    for production in get_productions(definition):
-        if production.sort == sort and len(production.items) >= 2:
-            first_arg = production.items[0]
-            if type(first_arg) is KTerminal and not (
-                first_arg.value.startswith('project:')
-                or first_arg.value.startswith('init')
-                or first_arg.value.startswith('get')
-            ):
-                productions.append(production)
-    if len(productions) != 1:
-        raise ValueError(f'Expected 1 production for sort {sort}, not {productions}!')
-    return productions[0].klabel
+    return new_term