#!/usr/bin/env python3 """qls-check: the reference checker for Stalwart UNSAT certificates. python qls-check.py [--cnf ] [--json] Written from the certificate specification (https://api.stalwart.it/certificate-spec/) alone; section numbers below are that document's. One file, the standard library only, no network. It reports one of three outcomes: verified exit 0 the derivation replays to 0 = 1, or the exhaustion's refutation record proves the variety empty failed exit 1 the certificate does not establish what it claims not_self_contained exit 2 it is honest about a step it could not derive; what was established, and what was not, is named A check establishes that a derivation or a refutation is valid. It never establishes that a policy is safe, and its output says so. MIT licence: see LICENSE beside this file. """ from __future__ import annotations import argparse import hashlib import json import os import re import sys from typing import Dict, FrozenSet, List, Optional, Set, Tuple VERIFIED, FAILED, NOT_SELF_CONTAINED = "verified", "failed", "not_self_contained" EXIT = {VERIFIED: 0, FAILED: 1, NOT_SELF_CONTAINED: 2} SCOPE = ("This establishes that a derivation or a refutation is valid. It does not " "establish that the policy is safe, nor that the equations are the policy.") Monomial = Tuple[int, ...] Row = Tuple[FrozenSet[Monomial], int] # (monomials, right-hand side) _VARIABLE = re.compile(r"x(0|[1-9][0-9]*)\Z") _SHA256 = re.compile(r"[0-9a-f]{64}\Z") #: Section 5: the sources a leaf may carry and be a premise rather than an assertion. PREMISE_SOURCES = ("original-S1", "original-S2") #: Section 3: the two kinds of certificate. CERTIFICATE_TYPES = ("phase0", "algebraic") #: The `status` each external `origin_class` carries; the others carry verified_unsat. STATUS_OF = {"external_proof": "proof_unsat", "external_oracle": "oracle_unsat"} #: The source of the row Phase 0 asserts its refutation with. PHASE0_SOURCE = "cryptominisat-phase0" class Failed(Exception): """The certificate does not establish what it claims; the message says why.""" class Result: def __init__(self, outcome: str, reason: str, details: Optional[dict] = None) -> None: self.outcome, self.reason, self.details = outcome, reason, details or {} @property def exit_code(self) -> int: return EXIT[self.outcome] def as_dict(self) -> dict: return {"outcome": self.outcome, "reason": self.reason, "details": self.details, "scope": SCOPE} # --------------------------------------------------------------- section 4: grammar def parse_equation(text: object) -> Row: """An equation string as (monomials, right-hand side); raises Failed outside the grammar.""" if not isinstance(text, str) or " = " not in text: raise Failed(f"the equation {text!r} has no ' = ' separating its right-hand side") left, right = text.rsplit(" = ", 1) if right not in ("0", "1"): raise Failed(f"the equation {text!r} has a right-hand side that is not 0 or 1") rhs = int(right) terms: set = set() if left == "0": return frozenset(), rhs if not left: raise Failed(f"the equation {text!r} has no left-hand side") for term in left.split(" + "): if term == "1": rhs ^= 1 continue terms ^= {parse_monomial(term, f"the equation {text!r}")} return frozenset(terms), rhs def parse_monomial(term: object, where: str) -> Monomial: """A term other than `1`, as its variables ascending; raises Failed outside the grammar.""" if not isinstance(term, str) or not term: raise Failed(f"{where} has an empty term") factors = set() for factor in term.split("*"): match = _VARIABLE.match(factor) if match is None: raise Failed(f"{where} has a factor {factor!r} that is not a variable") factors.add(int(match.group(1))) return tuple(sorted(factors)) def parse_term(term: object, where: str) -> Monomial: """A term including `1`, which is the empty monomial.""" return () if term == "1" else parse_monomial(term, where) def add(left: Row, right: Row) -> Row: return left[0] ^ right[0], left[1] ^ right[1] def multiply(row: Row, variable: int) -> Row: out: set = set() for monomial in row[0]: out ^= {tuple(sorted(set(monomial) | {variable}))} if row[1]: out ^= {(variable,)} return frozenset(out), 0 def is_contradiction(row: Row) -> bool: return not row[0] and row[1] == 1 def holds(row: Row, point: str) -> bool: """Whether the row's equation holds at a point given as one binary digit per variable.""" left = 0 for monomial in row[0]: left ^= int(all(point[v] == "1" for v in monomial)) return left == row[1] # --------------------------------------------------------------- the document def certificate_of(document: object) -> dict: """Section 1: the document whole, or its `certificate` member.""" if not isinstance(document, dict): raise Failed("the document is not a JSON object") if "operations" not in document and isinstance(document.get("certificate"), dict): return document["certificate"] return document def integer(value: object) -> bool: return isinstance(value, int) and not isinstance(value, bool) def variable_of(text: object, what: str) -> int: match = _VARIABLE.match(text) if isinstance(text, str) else None if match is None: raise Failed(f"{what} {text!r} is not a variable") return int(match.group(1)) def asserted(step: dict) -> bool: return step.get("oracle") is True or step.get("internal_assertion") is True def check_invariants(certificate: dict, unbound_cnf: bool = False) -> None: """Section 10, step 3.""" operations = certificate.get("operations") if operations is not None and not isinstance(operations, list): raise Failed("`operations` is not a list") for at, step in enumerate(operations or []): if not isinstance(step, dict): raise Failed(f"operation {at} is not an object") if step.get("op") == "INSERT" and "equation" not in step and not asserted(step): raise Failed(f"operation {at} is an INSERT with no equation that is not flagged as asserted") kind = certificate.get("certificate_type") if kind not in CERTIFICATE_TYPES: raise Failed(f"certificate_type {kind!r} is neither phase0 nor algebraic") final = certificate.get("final_step") phase0_final = isinstance(final, dict) and final.get("op") == "INSERT" and final.get("source") == PHASE0_SOURCE if (kind == "phase0") != phase0_final: raise Failed(f"certificate_type is {kind} and the final step " f"{'is' if phase0_final else 'is not'} Phase 0's asserted contradiction") origin = certificate.get("origin_class") status = certificate.get("status") if isinstance(status, str) and isinstance(origin, str): expected = STATUS_OF.get(origin, "verified_unsat") if status != expected: raise Failed(f"status is {status} and origin_class is {origin}, which is {expected}") if certificate.get("origin_class") == "derived": if any(isinstance(step, dict) and asserted(step) for step in operations or []): raise Failed("origin_class is derived and a row is flagged as asserted") present = ["proof_log" in certificate, "phase0_system" in certificate, certificate.get("origin_class") == "external_proof"] if unbound_cnf and present == [True, False, True]: return # a document issued before phase0_system was written, read as the reader asked if any(present) and not all(present): raise Failed("proof_log, phase0_system and origin_class external_proof do not occur together" + (" (a certificate issued before phase0_system was written is read with --unbound-cnf)" if present == [True, False, True] else "")) # --------------------------------------------------------------- section 10, step 5 def replay_operations(certificate: dict) -> Tuple[Dict[int, Row], Set[int]]: """Executes `operations`; returns every row by id, and the ids of the premise leaves.""" operations = certificate.get("operations") if not isinstance(operations, list): raise Failed("the certificate carries no `operations` list") rows: Dict[int, Row] = {} premises: Set[int] = set() for at, step in enumerate(operations): where = f"operation {at}" identifier, op = step.get("id"), step.get("op") if not integer(identifier): raise Failed(f"{where} has no integer id") if op == "INSERT": if "equation" not in step: raise Failed(f"{where} is an asserted row, and nothing can be replayed from it") if not asserted(step): if step.get("source") not in PREMISE_SOURCES: raise Failed(f"{where} is a leaf from {step.get('source')!r}, which is neither an " "original equation nor flagged as asserted") premises.add(identifier) row = parse_equation(step["equation"]) elif op == "ADD": parents = step.get("parents") if not isinstance(parents, list) or len(parents) != 2: raise Failed(f"{where} is an ADD without exactly two parents") row = add(row_named(rows, parents[0], where), row_named(rows, parents[1], where)) elif op == "MULTIPLY": base = row_named(rows, step.get("base_id"), where) row = multiply(base, variable_of(step.get("multiplier"), f"{where}'s multiplier")) else: raise Failed(f"{where} has an unknown op {op!r}") if is_contradiction(row): raise Failed(f"{where} already derives 0 = 1, so the final step was not withheld") if identifier in rows: raise Failed(f"{where} reuses id {identifier}") rows[identifier] = row return rows, premises def row_named(rows: Dict[int, Row], identifier: object, where: str) -> Row: if not integer(identifier) or identifier not in rows: raise Failed(f"{where} names a row that is not defined before it") return rows[identifier] def replay(certificate: dict) -> Dict[str, int]: """Executes `operations`, then `final_step`; raises Failed unless it reaches 0 = 1.""" rows, _ = replay_operations(certificate) final = certificate.get("final_step") if not isinstance(final, dict): raise Failed("the certificate withholds no final step") op, inputs = final.get("op"), final.get("inputs") if op == "INSERT": raise Failed("the final step is an asserted INSERT in a certificate that claims a derivation") if not isinstance(inputs, list): raise Failed("the final step names no inputs") if op == "ADD" and len(inputs) == 2: result = add(row_named(rows, inputs[0], "the final step"), row_named(rows, inputs[1], "the final step")) elif op == "MULTIPLY" and len(inputs) == 1: result = multiply(row_named(rows, inputs[0], "the final step"), variable_of(final.get("multiplier"), "the final step's multiplier")) else: raise Failed(f"the final step {op!r} does not name the inputs its op takes") if not is_contradiction(result): raise Failed(f"the final step produces {len(result[0])} monomial(s) with right-hand side " f"{result[1]}, not 0 = 1") return {"operations": len(certificate["operations"])} # --------------------------------------------------------------- section 11 def check_refutation(certificate: dict) -> Dict[str, int]: """Checks an exhaustion's refutation record; raises Failed unless it proves the variety empty.""" record = certificate.get("refutation") if not isinstance(record, dict): raise Failed("origin_class is exhaustion and there is no refutation record") rows, premises = replay_operations(certificate) final = certificate.get("final_step") if not (isinstance(final, dict) and final.get("op") == "INSERT" and final.get("internal_assertion") is True): raise Failed("the final step is not the contradiction the exhaustion asserts") n = record.get("variable_count") if not integer(n) or n < 0: raise Failed("the refutation record names no variable_count") for monomials, _ in rows.values(): for monomial in monomials: if any(v >= n for v in monomial): raise Failed(f"the derivation names a variable outside the record's {n}") candidates = record.get("candidates") if not integer(candidates) or candidates < 1: raise Failed("the refutation record names no positive candidate count") if record.get("unsubstantiated", 0) not in (0, None): raise Failed(f"the refutation record reports {record['unsubstantiated']} unsubstantiated candidate(s)") if "closure_gap" in record: raise Failed(f"the refutation record reports a gap in its closure: {record['closure_gap']}") # Step 1: N, holding 1. members: Set[Monomial] = set() for entry in record.get("normal_set") if isinstance(record.get("normal_set"), list) else []: monomial = parse_term(entry, "a normal-set member") if any(v >= n for v in monomial): raise Failed(f"the normal set names a variable outside the record's {n}") if monomial in members: raise Failed(f"the normal set lists {entry!r} twice") members.add(monomial) if () not in members: raise Failed("the normal set does not hold 1") # Step 2: one derived relation per border monomial, reducing it over N. border: Dict[Monomial, Row] = {} relations = record.get("border_relations") if not isinstance(relations, list): raise Failed("the refutation record carries no border_relations") for entry in relations: if not isinstance(entry, dict): raise Failed("a border relation is not an object") name = entry.get("monomial") monomial = parse_monomial(name, "a border monomial") if monomial in members: raise Failed(f"the border monomial {name} is in the normal set") row = row_named(rows, entry.get("operation"), f"the relation for {name}") if monomial not in row[0] or any(m != monomial and m not in members for m in row[0]): raise Failed(f"the relation cited for {name} is not {name} plus members of the normal set") if monomial in border: raise Failed(f"the border monomial {name} has two relations") border[monomial] = row # The closure: x_i * t is in N or a border monomial, for every member and variable. for t in members: for i in range(n): if i in t: continue product = tuple(sorted(t + (i,))) if product not in members and product not in border: spelled = "*".join(f"x{v}" for v in product) raise Failed(f"the product {spelled} is neither in the normal set nor a border monomial, " "so the normal set does not bound the variety") # Step 3: |N| distinct points, each in V(J) and each falsifying a premise. points = record.get("points") if not isinstance(points, list) or len(points) != candidates or candidates != len(members): listed = len(points) if isinstance(points, list) else 0 raise Failed(f"the record claims {candidates} candidate(s), lists {listed} and bounds the variety " f"by {len(members)}; the points are all of it only when the three agree") seen: Set[str] = set() for at, point in enumerate(points): where = f"point {at}" bits = point.get("assignment") if isinstance(point, dict) else None if not isinstance(bits, str) or len(bits) != n or set(bits) - {"0", "1"}: raise Failed(f"{where} carries no assignment of exactly {n} binary digits") if bits in seen: raise Failed(f"{where} repeats an earlier point") seen.add(bits) if not all(holds(row, bits) for row in border.values()): raise Failed(f"{where} does not satisfy every border relation, so it is not a point of V(J)") falsifies = point.get("falsifies") cited = falsifies.get("operation") if isinstance(falsifies, dict) else None if cited not in premises: raise Failed(f"{where} cites no premise leaf as the row it falsifies") if holds(rows[cited], bits): raise Failed(f"{where} satisfies the premise it cites") return {"points": len(points), "normal_set": len(members), "border_relations": len(border)} # --------------------------------------------------------------- section 12 def parse_dimacs(text: str) -> Tuple[int, List[List[int]]]: original, header, declared, clauses, current = None, False, 0, [], [] for line in text.splitlines(): words = line.split() if not words: continue if words[0] == "c": if len(words) >= 3 and words[1] == "original_variables" and words[2].isdigit(): original = int(words[2]) continue if words[0] == "p": if len(words) != 4 or words[1] != "cnf": raise Failed("the CNF has a malformed 'p cnf' header") header, declared = True, int(words[3]) continue if not header: raise Failed("the CNF has a clause before its 'p cnf' header") for word in words: literal = int(word) if literal == 0: clauses.append(current) current = [] else: current.append(literal) if not header or current: raise Failed("the CNF has no header, or a clause not ended by 0") if original is None: raise Failed("the CNF does not name its original variables") if len(clauses) != declared: raise Failed(f"the CNF's header declares {declared} clauses and it has {len(clauses)}") return original, clauses def decode_cnf(original: int, clauses: List[List[int]]) -> Tuple[Dict[int, Monomial], List[Tuple[List[int], int]]]: """Section 12: the definitions and the equations a CNF encodes, 0-indexed; raises Failed on a discrepancy.""" definitions: Dict[int, Monomial] = {} equations: List[Tuple[List[int], int]] = [] blocks_naming: Dict[int, int] = {} connectors: set = set() chain: Optional[List[int]] = None connector = 0 at = 0 while at < len(clauses): c = clauses[at] if (len(c) == 3 and c[0] < 0 and c[1] < 0 and c[2] > 0 and at + 2 < len(clauses) and clauses[at + 1] == [-c[0], -c[2]] and clauses[at + 2] == [-c[1], -c[2]]): if chain is not None: raise Failed(f"clause {at} of the CNF: a definition inside a chain") x, y, z = -c[0] - 1, -c[1] - 1, c[2] - 1 if z < original or x == y or x >= original or y >= original: raise Failed(f"clause {at} of the CNF: x{z} is not an auxiliary over two distinct system variables") product = tuple(sorted((x, y))) if definitions.get(z, product) != product: raise Failed(f"clause {at} of the CNF: the auxiliary x{z} is defined twice") definitions[z] = product for v in (x, y, z): blocks_naming[v] = blocks_naming.get(v, 0) + 1 at += 3 continue k = len(c) if k == 0: if chain is not None: raise Failed(f"clause {at} of the CNF: the empty clause inside a chain") equations.append(([], 1)) at += 1 continue variables = [abs(literal) for literal in c] size = 1 << (k - 1) block = clauses[at:at + size] if k > 3 or len(set(variables)) != k or len(block) != size: raise Failed(f"clause {at} of the CNF is not a block of this encoding") masks, parities = set(), set() for clause in block: if [abs(literal) for literal in clause] != variables: raise Failed(f"clause {at} of the CNF is not the full set of clauses of one parity block") masks.add(tuple(literal < 0 for literal in clause)) parities.add(sum(literal < 0 for literal in clause) % 2) if len(masks) != size or len(parities) != 1: raise Failed(f"clause {at} of the CNF is not the full set of clauses of one parity block") rhs = 1 if parities.pop() == 0 else 0 for v in variables: blocks_naming[v - 1] = blocks_naming.get(v - 1, 0) + 1 zero = [v - 1 for v in variables] if k == 1: if chain is not None: raise Failed(f"clause {at} of the CNF: a unit clause inside a chain") equations.append((zero, rhs)) elif k == 2: if chain is not None: if variables[0] != connector: raise Failed(f"clause {at} of the CNF: a chain's last link does not start from its connector") equations.append((chain + [zero[1]], rhs)) chain = None else: equations.append((zero, rhs)) else: if rhs: raise Failed(f"clause {at} of the CNF: a chain link of parity 1") if chain is not None: if variables[0] != connector: raise Failed(f"clause {at} of the CNF: a chain link does not start from the previous connector") chain.append(zero[1]) else: chain = zero[:2] connector = variables[2] connectors.add(variables[2] - 1) at += size if chain is not None: raise Failed("the CNF ends inside a chain") for v in connectors: if v < original or v in definitions or blocks_naming.get(v) != 2: raise Failed(f"the connector x{v} is a system variable, a product, or not in exactly two links") for variables, _ in equations: if connectors.intersection(variables): raise Failed("a connector is also a term of a decoded equation") return definitions, equations def compare_cnf(certificate: dict, cnf_text: str) -> int: """Decodes the CNF and compares it with `phase0_system`; returns the equations compared.""" original, clauses = parse_dimacs(cnf_text) definitions, decoded = decode_cnf(original, clauses) system = certificate.get("phase0_system") if not isinstance(system, list): raise Failed("the certificate carries no phase0_system to compare the CNF with") wanted = [] for at, entry in enumerate(system): monomials, rhs = parse_equation(entry.get("equation") if isinstance(entry, dict) else None) # Section 12: the encoding names a product of two variables at most, so a # system equation of higher degree is one no CNF here can encode. if any(len(monomial) > 2 for monomial in monomials): raise Failed(f"phase0_system equation {at} has a monomial of degree above 2") if monomials or rhs: wanted.append((monomials, rhs)) if len(decoded) != len(wanted): raise Failed(f"the CNF decodes to {len(decoded)} equation(s) and phase0_system has {len(wanted)}") for at, ((variables, rhs), (monomials, want_rhs)) in enumerate(zip(decoded, wanted)): if variables and any(v >= original and v not in definitions for v in variables): raise Failed(f"decoded equation {at} carries a variable the encoding gives no meaning") terms = sorted(definitions.get(v, (v,)) for v in variables) if rhs != want_rhs or terms != sorted(monomials) or len(set(terms)) != len(terms): raise Failed(f"decoded equation {at} is not phase0_system's equation {at}") return len(wanted) # --------------------------------------------------------------- section 10 def check(document: object, cnf_text: Optional[str] = None, unbound_cnf: bool = False) -> Result: """Checks one certificate; `cnf_text` is the CNF an external_proof certificate names. `unbound_cnf` reads an external_proof certificate that carries no `phase0_system` -- one issued before the field was written -- by the CNF's digest alone, and the result says the CNF is tied to nothing. Without it such a document is `failed`: nothing in it shows it is old rather than altered. """ try: certificate = certificate_of(document) version = certificate.get("format_version", 1) if version != 1 or not integer(version): raise Failed(f"the certificate declares format version {version!r}; this checker reads version 1") check_invariants(certificate, unbound_cnf) origin = certificate.get("origin_class") if origin == "derived": replayed = replay(certificate) return Result(VERIFIED, f"replayed {replayed['operations']} operation(s), and the withheld " "final step yields 0 = 1", replayed) if origin == "exhaustion": checked = check_refutation(certificate) return Result(VERIFIED, f"the refutation record checks: {checked['border_relations']} derived border " f"relation(s) bound the variety by a normal set of {checked['normal_set']}, and all " f"{checked['points']} of its point(s) falsify a premise", checked) if origin == "external_oracle": oracles = certificate.get("oracle_dependencies") or [] return Result(NOT_SELF_CONTAINED, "the contradiction rests on rows asserted by " + (", ".join(map(str, oracles)) or "an external procedure") + ", which nothing in the certificate checks", {"oracles": oracles}) if origin == "external_proof": log = certificate.get("proof_log") if not isinstance(log, dict): raise Failed("origin_class is external_proof and there is no proof_log") for name in ("cnf_sha256", "frat_sha256"): if not isinstance(log.get(name), str) or not _SHA256.match(log[name]): raise Failed(f"proof_log.{name} is not a lowercase hex SHA-256") if cnf_text is None: raise Failed(f"the CNF {log.get('cnf_file')!r} the proof_log names was not found") digest = hashlib.sha256(cnf_text.encode("utf-8")).hexdigest() if digest != log.get("cnf_sha256"): raise Failed("the CNF's SHA-256 is not the one proof_log records") command = f"frat-rs elab {log.get('frat_file')} {log.get('cnf_file')}" if "phase0_system" not in certificate: # Only with --unbound-cnf: check_invariants refuses this document otherwise. return Result(NOT_SELF_CONTAINED, "UNBOUND: the CNF is the one proof_log records by its digest, " "and nothing here ties it to the policy's equations: the certificate carries no " "phase0_system to decode it against, which a certificate issued before the field " "was written does not. Its refutation is checked by: " + command, {"equations": None, "unbound_cnf": True, "next": command}) compared = compare_cnf(certificate, cnf_text) return Result(NOT_SELF_CONTAINED, f"the CNF decodes to the {compared} equation(s) of phase0_system. " f"Its refutation is checked by: {command}", {"equations": compared, "next": command}) raise Failed(f"origin_class {origin!r} is not one of the four this format defines") except Failed as failure: return Result(FAILED, str(failure)) def check_file(path: str, cnf_path: Optional[str] = None, unbound_cnf: bool = False) -> Result: try: with open(path, "rb") as handle: document = json.loads(handle.read().decode("utf-8")) except (OSError, ValueError) as error: return Result(FAILED, f"{path} could not be read as JSON: {error}") cnf_text = None certificate = document.get("certificate") if isinstance(document, dict) and "operations" not in document else document log = certificate.get("proof_log") if isinstance(certificate, dict) else None if cnf_path is None and isinstance(log, dict) and isinstance(log.get("cnf_file"), str): cnf_path = os.path.join(os.path.dirname(os.path.abspath(path)), os.path.basename(log["cnf_file"])) if cnf_path is not None and os.path.isfile(cnf_path): with open(cnf_path, "rb") as handle: cnf_text = handle.read().decode("utf-8") return check(document, cnf_text, unbound_cnf) def main(argv: Optional[List[str]] = None) -> int: parser = argparse.ArgumentParser(prog="qls-check", description="Check a Stalwart UNSAT certificate.") parser.add_argument("certificate") parser.add_argument("--cnf", help="the CNF an external_proof certificate names; by default beside it") parser.add_argument("--json", action="store_true", help="print the result as JSON") parser.add_argument("--unbound-cnf", action="store_true", help="read an external_proof certificate with no phase0_system, issued before the field " "was written, by the CNF's digest alone") args = parser.parse_args(argv) result = check_file(args.certificate, args.cnf, args.unbound_cnf) if args.json: print(json.dumps(result.as_dict(), indent=2, sort_keys=True)) else: print(f"qls-check: {result.outcome}") print(result.reason) print(SCOPE) return result.exit_code if __name__ == "__main__": sys.exit(main())