This demo runs entirely in your browser. The Python checker is
being downloaded and started under Pyodide; nothing you type is sent anywhere.
First load takes a few seconds.
"""certkit -- a certificate format for machine-checked program admission, and an
independent checker for it.
The checker imports nothing outside the Python standard library. It contains no
solver, no search, and no floating-point arithmetic. You are meant to read it.
Quickstart::
from certkit import atom, make_spec, check_certificate
# "if 19 + payload <= record_len then 3 + payload <= record_len"
domain = [atom({"payload": -1}), atom({"payload": 1}, -65535)]
guard = [atom({"payload": 1, "record_len": -1}, 19)]
safety = [atom({"payload": 1, "record_len": -1}, 3)]
spec = make_spec(domain, guard, safety, name="heartbleed")
See ``SPEC.md`` for the on-disk format.
"""
from .atoms import Atom, atom, atom_from_json, atom_to_json, negate
from .cert import (
ACCEPTED,
CERT_SCHEMA,
REFUSED,
SPEC_SCHEMA,
UNVERIFIED,
CheckReport,
check_certificate,
fingerprint,
make_spec,
reconstruct_obligation,
)
from .farkas import FarkasResult, verify_farkas
from .smtlib import (
SmtLibError,
SmtLibUnsupported,
export_obligation,
export_spec,
import_spec,
)
from .sos import SOS_SCHEMA, verify_sos
__version__ = "0.4.1"
__all__ = [
"ACCEPTED",
"export_spec",
"export_obligation",
"import_spec",
"SmtLibError",
"SmtLibUnsupported",
"REFUSED",
"UNVERIFIED",
"Atom",
"atom",
"negate",
"atom_from_json",
"atom_to_json",
"verify_farkas",
"FarkasResult",
"verify_sos",
"SOS_SCHEMA",
"check_certificate",
"CheckReport",
"reconstruct_obligation",
"make_spec",
"fingerprint",
"CERT_SCHEMA",
"SPEC_SCHEMA",
"__version__",
]
"""Linear atoms over the rationals.
An atom is a linear relation in canonical form::
sum(coeff[v] * v) + const {< 0 if strict else <= 0}
Every relation a caller might write (``x <= y``, ``3*a + 2 < b``, ``x >= 0``) is
normalised into this single shape, so the checker has exactly one form to reason
about. That is deliberate: the trusted core stays small because it never has to
case-split on relational operators.
All arithmetic is exact (``fractions.Fraction``). There are no floats anywhere in
this package -- a floating-point rounding error in a proof checker is a soundness
bug, not a precision nuisance.
"""
from __future__ import annotations
from collections.abc import Iterable, Mapping
from fractions import Fraction
from typing import Any
__all__ = ["Atom", "atom", "negate", "atom_from_json", "atom_to_json"]
class Atom:
"""A linear relation ``sum(coeff*var) + const <= 0`` (or ``< 0`` when strict).
Instances are immutable in practice; the checker never mutates an atom it was
given. Construct with :func:`atom` for convenience.
"""
__slots__ = ("coeff", "const", "strict")
def __init__(
self,
coeff: Mapping[str, Fraction],
const: Fraction = Fraction(0),
strict: bool = False,
) -> None:
# Drop zero coefficients so structurally equal atoms compare equal.
self.coeff: dict[str, Fraction] = {
v: Fraction(c) for v, c in coeff.items() if Fraction(c) != 0
}
self.const: Fraction = Fraction(const)
self.strict: bool = bool(strict)
def variables(self) -> Iterable[str]:
return self.coeff.keys()
def __eq__(self, other: Any) -> bool:
if not isinstance(other, Atom):
return NotImplemented
return (
self.coeff == other.coeff and self.const == other.const and self.strict == other.strict
)
def __hash__(self) -> int:
return hash((tuple(sorted(self.coeff.items())), self.const, self.strict))
def __repr__(self) -> str:
terms = " + ".join(f"{c}*{v}" for v, c in sorted(self.coeff.items()))
rel = "<" if self.strict else "<="
body = terms if terms else "0"
if self.const:
body = f"{body} + {self.const}"
return f"Atom({body} {rel} 0)"
def atom(coeff: Mapping[str, Any], const: Any = 0, strict: bool = False) -> Atom:
"""Build an :class:`Atom`, coercing ints/strings/fractions to ``Fraction``."""
return Atom({v: Fraction(c) for v, c in coeff.items()}, Fraction(const), strict)
def negate(a: Atom) -> Atom:
"""Logical negation of an atom.
``not (L <= 0)`` is ``-L < 0``; ``not (L < 0)`` is ``-L <= 0``. Negation flips
every coefficient, the constant, and the strictness flag. This is the only
place strictness inverts, and it is what lets a refutation be expressed as an
infeasible conjunction.
"""
return Atom(
{v: -c for v, c in a.coeff.items()},
-a.const,
not a.strict,
)
def atom_from_json(d: Mapping[str, Any]) -> Atom:
"""Parse an atom from the on-disk JSON form.
Coefficients and the constant are ``[numerator, denominator]`` pairs so a
certificate never depends on decimal parsing.
"""
coeff = {v: Fraction(int(n), int(dd)) for v, (n, dd) in d["coeff"].items()}
n, dd = d["const"]
return Atom(coeff, Fraction(int(n), int(dd)), bool(d["strict"]))
def atom_to_json(a: Atom) -> dict[str, Any]:
"""Serialise an atom to the on-disk JSON form."""
return {
"coeff": {v: [c.numerator, c.denominator] for v, c in sorted(a.coeff.items())},
"const": [a.const.numerator, a.const.denominator],
"strict": a.strict,
}
"""Farkas certificate checking -- the load-bearing ~30 lines of this package.
A *Farkas certificate* witnesses that a conjunction of linear atoms is
unsatisfiable over the rationals. It is a vector of nonnegative multipliers, one
per atom, such that the multiplied atoms sum to a contradiction: every variable
cancels, and the surviving constant is impossible.
Why that is sound, in one paragraph. If every atom ``L_i <= 0`` holds and every
multiplier ``m_i >= 0``, then ``sum(m_i * L_i) <= 0`` necessarily. If the
multipliers are chosen so that all variable coefficients cancel, the left side
collapses to a constant ``c``. So ``c <= 0`` must hold. If instead ``c > 0``, no
assignment can satisfy the conjunction -- the system is infeasible. When any atom
used with a nonzero multiplier is *strict* (``< 0``), the combination is strict
too, so ``c >= 0`` already suffices for the contradiction.
Why rational infeasibility is enough for integer problems: the integers are a
subset of the rationals, so a system with no rational solution has no integer
solution either. The converse does not hold, which is why a failure to find a
Farkas certificate is *not* a proof of satisfiability -- this checker refuses,
it does not certify the negation.
Checking is O(number of nonzero multipliers x atom size) and uses only exact
rational arithmetic. Finding the multipliers is somebody else's problem: this
package deliberately contains no search, no solver, and no LP. That asymmetry is
the point of proof-carrying verification -- the producer may be arbitrarily
clever and arbitrarily untrusted, because the checker is small enough to audit.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from fractions import Fraction
from typing import Any
from .atoms import Atom
__all__ = ["verify_farkas", "FarkasResult"]
class FarkasResult:
"""Outcome of a certificate check, with a reason when it fails.
Truthy when the certificate is valid, so ``if verify_farkas(...):`` reads
naturally, while ``.reason`` explains a rejection.
"""
__slots__ = ("ok", "reason")
def __init__(self, ok: bool, reason: str = "") -> None:
self.ok = ok
self.reason = reason
def __bool__(self) -> bool:
return self.ok
def __repr__(self) -> str:
return f"FarkasResult(ok={self.ok!r}, reason={self.reason!r})"
def verify_farkas(
atoms: Sequence[Atom],
multipliers: Mapping[Any, Any],
) -> FarkasResult:
"""Check a Farkas certificate for the conjunction ``atoms``.
``multipliers`` maps an atom index to a nonnegative rational weight. Indices
may arrive as strings (JSON object keys are always strings), so they are
coerced; anything that is not a valid in-range index is a rejection rather
than an exception. That matters: this function is fed attacker-controlled
input, and a checker that raises on malformed input is a denial-of-service
surface at best and a bypass at worst.
Returns a :class:`FarkasResult` -- truthy exactly when the multipliers prove
the conjunction infeasible.
"""
if not multipliers:
return FarkasResult(False, "empty multiplier vector")
combined: dict[str, Fraction] = {}
const = Fraction(0)
any_strict = False
used = 0
for raw_index, raw_weight in multipliers.items():
# --- coerce and validate the index -------------------------------
try:
index = int(raw_index)
except (TypeError, ValueError):
return FarkasResult(False, f"non-integer atom index {raw_index!r}")
if index < 0 or index >= len(atoms):
return FarkasResult(False, f"atom index {index} out of range")
# --- coerce and validate the weight ------------------------------
try:
weight = _as_fraction(raw_weight)
except (TypeError, ValueError, ZeroDivisionError):
return FarkasResult(False, f"malformed multiplier {raw_weight!r}")
if weight < 0:
return FarkasResult(False, f"negative multiplier {weight} at index {index}")
if weight == 0:
continue # a zero weight contributes nothing and is not "use"
# --- accumulate ---------------------------------------------------
a = atoms[index]
for v, c in a.coeff.items():
combined[v] = combined.get(v, Fraction(0)) + weight * c
const += weight * a.const
any_strict = any_strict or a.strict
used += 1
if used == 0:
return FarkasResult(False, "all multipliers zero")
residual = {v: c for v, c in combined.items() if c != 0}
if residual:
names = ", ".join(sorted(residual))
return FarkasResult(False, f"variables did not cancel: {names}")
# A strict atom makes the combination strict, so const >= 0 already
# contradicts "< 0". Otherwise we need a genuinely positive constant.
if any_strict:
if const >= 0:
return FarkasResult(True)
return FarkasResult(False, f"strict combination needs const >= 0, got {const}")
if const > 0:
return FarkasResult(True)
return FarkasResult(False, f"non-strict combination needs const > 0, got {const}")
def _as_fraction(value: Any) -> Fraction:
"""Coerce a JSON-ish weight to an exact Fraction.
Accepts ``5``, ``"5"``, ``"3/4"``, ``[3, 4]``, and ``Fraction``. Rejects
floats: a multiplier that arrived as a float has already lost exactness, and
silently accepting it would let rounding decide soundness.
"""
if isinstance(value, Fraction):
return value
if isinstance(value, bool):
raise TypeError("bool is not a multiplier")
if isinstance(value, int):
return Fraction(value)
if isinstance(value, (list, tuple)):
if len(value) != 2:
raise ValueError("pair form must be [numerator, denominator]")
return Fraction(int(value[0]), int(value[1]))
if isinstance(value, str):
return Fraction(value)
raise TypeError(f"unsupported multiplier type {type(value).__name__}")
"""Sum-of-squares certificate checking -- the Farkas move one theory up.
Farkas handles *linear* infeasibility. For a *nonlinear* obligation of the form
``target(x) >= 0``, the analogous certificate is a rational sum-of-squares
identity::
scale * target == sum_i q_i^2
with ``scale`` a positive integer and each ``q_i`` a polynomial. Soundness is
immediate: every square is nonnegative and ``scale > 0``, so ``target >= 0``.
As with Farkas, an untrusted producer (typically a semidefinite-programming
relaxation) *finds* the ``q_i``; this checker only re-multiplies them and
verifies the polynomial identity holds exactly over the rationals. A perturbed
certificate fails, because every monomial coefficient must cancel to zero -- there
is no tolerance to tune.
Scope, honestly: sum-of-squares is incomplete. A nonnegative polynomial need not
admit an SOS decomposition (Motzkin's polynomial is the standard witness), and
nonnegativity over the integers or over a non-compact domain is a different
question again. A refusal here means "no certificate was supplied or it did not
check", never "the target is negative".
Polynomials are dicts from exponent tuples to Fractions. The JSON form encodes a
monomial as a comma-joined exponent string over the certificate's declared
variable order: ``"2,0"`` is ``x^2``, ``"1,1"`` is ``x*y``, ``"0,0"`` is the
constant term.
"""
from __future__ import annotations
from collections.abc import Mapping
from fractions import Fraction
from typing import Any
__all__ = ["verify_sos", "SOS_SCHEMA", "parse_polynomial"]
SOS_SCHEMA = "certkit/sos/v1"
Monomial = tuple[int, ...]
Polynomial = dict[Monomial, Fraction]
def _parse_monomial(key: str) -> Monomial:
return tuple(int(e) for e in key.split(","))
def parse_polynomial(d: Mapping[str, Any]) -> Polynomial:
"""Parse ``{"2,0": [3,1], ...}`` into ``{(2,0): Fraction(3,1), ...}``."""
out: Polynomial = {}
for k, v in d.items():
n, dd = v
c = Fraction(int(n), int(dd))
if c != 0:
out[_parse_monomial(k)] = c
return out
def _poly_add(a: Polynomial, b: Polynomial, scale: Fraction = Fraction(1)) -> Polynomial:
out = dict(a)
for m, c in b.items():
out[m] = out.get(m, Fraction(0)) + scale * c
return {m: c for m, c in out.items() if c != 0}
def _poly_mul(a: Polynomial, b: Polynomial) -> Polynomial:
out: Polynomial = {}
for ma, ca in a.items():
for mb, cb in b.items():
# Exponent vectors add under multiplication. Pad to equal arity so a
# certificate cannot smuggle a shape mismatch past us.
width = max(len(ma), len(mb))
ea = ma + (0,) * (width - len(ma))
eb = mb + (0,) * (width - len(mb))
m = tuple(x + y for x, y in zip(ea, eb))
out[m] = out.get(m, Fraction(0)) + ca * cb
return {m: c for m, c in out.items() if c != 0}
def verify_sos(cert: Mapping[str, Any]) -> bool:
"""Re-check a rational sum-of-squares certificate.
Returns ``True`` only when the schema tag matches, ``scale`` is a positive
integer, the square list is non-empty with no zero polynomial, and the
identity ``scale*target == sum q_i^2`` holds exactly.
"""
if not isinstance(cert, Mapping) or cert.get("schema") != SOS_SCHEMA:
return False
squares = cert.get("squares")
if not isinstance(squares, list) or not squares:
return False
try:
scale = int(cert["scale"])
target = parse_polynomial(cert["target"])
qs = [parse_polynomial(q) for q in squares]
except Exception:
# Malformed input is a rejection, never a traceback.
return False
if scale <= 0:
return False
if any(not q for q in qs):
# An empty polynomial is the zero polynomial; allowing it would let a
# certificate pad its square list without contributing anything.
return False
sos: Polynomial = {}
for q in qs:
sos = _poly_add(sos, _poly_mul(q, q))
scaled_target = {m: scale * c for m, c in target.items()}
residual = _poly_add(scaled_target, sos, scale=Fraction(-1))
return len(residual) == 0
"""Certificate container, obligation reconstruction, and the top-level check.
The security-relevant idea in this module is **reconstruction**. A naive checker
reads the atoms out of the certificate and verifies the multipliers against
*those* atoms. That is close to worthless: a certificate that carries an easy,
unrelated system together with a valid refutation of it would pass, while proving
nothing about the program you care about.
So the checker here ignores any atoms the certificate carries and rebuilds the
obligation from an independently supplied specification:
system := domain AND guard AND NOT(safety)
If that system is infeasible, then within the domain the guard implies safety --
which is the property actually being claimed. The certificate is only permitted
to supply the *multipliers*.
The certificate is additionally bound to the specification by a fingerprint over
the canonical serialisation of the spec. This detects drift and accidental
mismatch. It is explicitly **not** a defence against a deliberate forger, who
would simply recompute the fingerprint over an edited spec -- soundness rests on
a human having audited the spec's relations, which are small enough to read.
"""
from __future__ import annotations
import hashlib
import json
from collections.abc import Mapping, Sequence
from typing import Any
from .atoms import Atom, atom_from_json, atom_to_json, negate
from .farkas import FarkasResult, verify_farkas
__all__ = [
"CERT_SCHEMA",
"SPEC_SCHEMA",
"ACCEPTED",
"REFUSED",
"UNVERIFIED",
"fingerprint",
"reconstruct_obligation",
"check_certificate",
"CheckReport",
]
CERT_SCHEMA = "certkit/farkas/v1"
SPEC_SCHEMA = "certkit/spec/v1"
#: Every obligation was refuted **and** the certificate was bound to this spec.
ACCEPTED = "ACCEPTED"
#: At least one obligation was not refuted.
REFUSED = "REFUSED"
#: The arithmetic checked out, but a required precondition was never established
#: -- so no claim is being made. This is not a pass.
UNVERIFIED = "UNVERIFIED"
def fingerprint(spec: Mapping[str, Any]) -> str:
"""SHA-256 over the canonical JSON of a specification body."""
payload = json.dumps(spec, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
def _parse_spec_atoms(spec: Mapping[str, Any]) -> tuple[list[Atom], list[Atom]]:
"""Parse a spec's atoms once: the ``domain ++ guard`` prefix, and safety.
Every obligation shares the same prefix and differs only in which safety
conjunct is negated onto the end. Parsing is the expensive part -- each atom
builds a Fraction per coefficient -- so :func:`check_certificate` parses once
here and reuses the result across all conjuncts, rather than re-parsing the
whole spec per obligation.
"""
prefix: list[Atom] = [atom_from_json(a) for a in spec.get("domain", [])]
prefix += [atom_from_json(a) for a in spec.get("guard", [])]
safety: list[Atom] = [atom_from_json(a) for a in spec.get("safety", [])]
return prefix, safety
def reconstruct_obligation(spec: Mapping[str, Any]) -> list[Atom]:
"""Rebuild ``domain AND guard AND NOT(safety)`` from a specification.
``safety`` is a conjunction; its negation is a disjunction, which is not
expressible as a single atom list. The spec therefore names which safety
conjunct is under refutation via ``safety_index``, and one obligation is
formed per conjunct. Callers wanting the whole property iterate over the
indices -- see :func:`check_certificate`.
"""
prefix, safety = _parse_spec_atoms(spec)
idx = int(spec.get("safety_index", 0))
if not safety:
raise ValueError("spec has no safety conjuncts")
if idx < 0 or idx >= len(safety):
raise ValueError(f"safety_index {idx} out of range")
return [*prefix, negate(safety[idx])]
class CheckReport:
"""Aggregate result across every safety conjunct of a specification.
Two things must hold before this reports success, and they are tracked
separately because they can fail separately:
``obligations_ok``
every safety conjunct was refuted by the supplied multipliers.
``binding_verified``
the certificate was cryptographically bound to *this* spec.
``ok`` is the conjunction. A certificate whose arithmetic is impeccable but
which was never bound to the spec proves nothing about the spec, so it
reports :data:`UNVERIFIED` -- never ``ACCEPTED``, and never silently ``ok``.
"""
__slots__ = ("obligations", "reason", "obligations_ok", "binding_verified")
def __init__(
self,
ok: bool,
obligations: list[dict[str, Any]],
reason: str = "",
*,
binding_verified: bool = True,
) -> None:
self.obligations_ok = ok
self.binding_verified = binding_verified
self.obligations = obligations
self.reason = reason
@property
def ok(self) -> bool:
"""True only when the proof checks *and* it is bound to this spec."""
return self.obligations_ok and self.binding_verified
@property
def verdict(self) -> str:
""":data:`ACCEPTED`, :data:`REFUSED`, or :data:`UNVERIFIED`."""
if not self.obligations_ok:
return REFUSED
if not self.binding_verified:
return UNVERIFIED
return ACCEPTED
def __bool__(self) -> bool:
return self.ok
def to_dict(self) -> dict[str, Any]:
return {
"verdict": self.verdict,
"ok": self.ok,
"obligations_ok": self.obligations_ok,
"binding_verified": self.binding_verified,
"reason": self.reason,
"obligations": self.obligations,
}
def __repr__(self) -> str:
return (
f"CheckReport(verdict={self.verdict!r}, n={len(self.obligations)}, "
f"reason={self.reason!r})"
)
def _repr_html_(self) -> str:
"""Rendering for Jupyter. Imported lazily so `cert` stays standalone.
The renderer gives UNVERIFIED its own colour rather than folding it into
pass or fail -- a notebook is exactly where a verdict is most likely to
be skimmed rather than read.
"""
from .notebook import report_html
return report_html(self)
def check_certificate(
spec: Mapping[str, Any],
cert: Mapping[str, Any],
*,
require_fingerprint: bool = True,
) -> CheckReport:
"""Verify that ``cert`` proves the guard implies safety over the domain.
One obligation is checked per safety conjunct. Every obligation must be
refuted, **and** the certificate must be bound to this spec, for the report
to be ``ok``.
Passing ``require_fingerprint=False`` skips the binding check. It does not
make the result an acceptance: the report comes back with
``binding_verified=False`` and ``verdict == UNVERIFIED``, because a
certificate that was never tied to this spec has not been shown to say
anything about this spec. The escape hatch is for authoring workflows where
the fingerprint has not been computed yet, not for relaxing the verdict.
"""
# A spec or certificate that is not a mapping at all -- None, a list, a bare
# string, a number -- used to raise AttributeError out of the checker. The
# JavaScript implementation guarded this and the Python one did not, which is
# exactly the kind of divergence the differential vectors exist to find; the
# stress suite found it first. A checker fed attacker-controlled input
# refuses; it does not raise.
if not isinstance(spec, Mapping):
return CheckReport(False, [], f"spec must be an object, got {type(spec).__name__}")
if not isinstance(cert, Mapping):
return CheckReport(False, [], f"certificate must be an object, got {type(cert).__name__}")
if cert.get("schema") != CERT_SCHEMA:
return CheckReport(False, [], f"unexpected certificate schema {cert.get('schema')!r}")
if spec.get("schema") != SPEC_SCHEMA:
return CheckReport(False, [], f"unexpected spec schema {spec.get('schema')!r}")
body = {k: v for k, v in spec.items() if k != "fingerprint"}
actual = fingerprint(body)
binding_verified = True
binding_note = ""
if require_fingerprint:
declared = spec.get("fingerprint")
if declared is not None and declared != actual:
return CheckReport(False, [], "spec fingerprint does not match its own body")
bound = cert.get("spec_fingerprint")
if bound is None:
return CheckReport(False, [], "certificate is not bound to a spec fingerprint")
if bound != actual:
return CheckReport(False, [], "certificate is bound to a different spec")
else:
binding_verified = False
binding_note = (
"TRUST ANCHOR ABSENT: fingerprint checking was disabled, so this certificate "
"was never tied to the supplied spec. The multipliers below were checked "
"against the spec's own reconstructed atoms, but nothing establishes that "
"this certificate was issued for this spec. Not an acceptance."
)
safety = spec.get("safety", [])
if not isinstance(safety, Sequence) or isinstance(safety, (str, bytes)):
return CheckReport(False, [], f"spec 'safety' must be a list, got {type(safety).__name__}")
if not safety:
return CheckReport(False, [], "spec has no safety conjuncts")
per_obligation = cert.get("obligations")
if not isinstance(per_obligation, list) or len(per_obligation) != len(safety):
return CheckReport(
False,
[],
f"certificate supplies {len(per_obligation) if isinstance(per_obligation, list) else 0}"
f" obligation(s); spec requires {len(safety)}",
)
# Parse the spec's atoms once rather than once per obligation. Every
# obligation is `domain ++ guard ++ [negate(safety[i])]`, so only the last
# element differs; re-parsing the shared prefix k times made the check
# quadratic in the number of safety conjuncts.
#
# A spec is attacker-controlled input like any other, so a malformed atom --
# a zero denominator, an Infinity coefficient, a string where an object
# belongs -- is a refusal with a reason, never a traceback out of the
# checker. Parsing up front means one bad atom fails every obligation, which
# is correct: the spec as a whole could not be rebuilt.
try:
prefix, safety_atoms = _parse_spec_atoms(body)
except (ValueError, TypeError, KeyError, AttributeError, ArithmeticError) as exc:
reason = f"malformed spec atom: {type(exc).__name__}: {exc}"
return CheckReport(
False,
[{"index": i, "ok": False, "reason": reason} for i in range(len(safety))],
reason,
binding_verified=binding_verified,
)
# One list, reused across obligations: only the final slot changes. Safe
# because verify_farkas reads the sequence and never retains it.
obligation: list[Atom] = [*prefix, negate(safety_atoms[0])]
results: list[dict[str, Any]] = []
all_ok = True
for i in range(len(safety)):
obligation[-1] = negate(safety_atoms[i])
atoms = obligation
entry = per_obligation[i]
multipliers = entry.get("multipliers") if isinstance(entry, Mapping) else None
if not isinstance(multipliers, Mapping):
results.append({"index": i, "ok": False, "reason": "missing multipliers"})
all_ok = False
continue
res: FarkasResult = verify_farkas(atoms, multipliers)
results.append({"index": i, "ok": bool(res), "reason": res.reason})
all_ok = all_ok and bool(res)
if not all_ok:
reason = "one or more obligations failed"
if binding_note:
reason = f"{reason}; also: {binding_note}"
else:
reason = binding_note
return CheckReport(all_ok, results, reason, binding_verified=binding_verified)
def make_spec(
domain: Sequence[Atom],
guard: Sequence[Atom],
safety: Sequence[Atom],
name: str = "unnamed",
) -> dict[str, Any]:
"""Build a specification dict with its fingerprint filled in."""
body: dict[str, Any] = {
"schema": SPEC_SCHEMA,
"name": name,
"domain": [atom_to_json(a) for a in domain],
"guard": [atom_to_json(a) for a in guard],
"safety": [atom_to_json(a) for a in safety],
}
body["fingerprint"] = fingerprint(dict(body.items()))
return body
"""SMT-LIB 2 bridge: emit an obligation a solver can check, and read one back.
SMT-LIB is the lingua franca of this field. Anything that already runs z3, cvc5
or CBMC speaks it, and until certkit did too, "bring your own producer" was an
instruction with no path attached. This module is that path:
certkit export --spec s.json --smtlib -o obligation.smt2 # -> z3 obligation.smt2
certkit import --smtlib f.smt2 -o s.json # <- back to a spec
**Export** is total: every certkit spec has an SMT-LIB rendering, because the
certkit fragment is a strict subset of QF_LIA. The emitted script asserts the
obligation `domain AND guard AND NOT(safety[i])` and asks `check-sat`. A solver
answering `unsat` agrees with an `ACCEPTED` certificate; `sat` means the guard
really does admit a forbidden state, and the model is a counterexample.
**Import is partial, and loudly so.** SMT-LIB is a large language and certkit
handles quantifier-free linear integer arithmetic and nothing else. Rather than
silently dropping what it does not understand -- which would produce a spec that
proves something *weaker* than the file said, the worst possible failure here --
the reader raises :class:`SmtLibUnsupported` naming the construct it refused.
A partial importer that guesses is worse than no importer, because everything
downstream would then be proving the wrong theorem, correctly.
"""
from __future__ import annotations
import re
from fractions import Fraction
from typing import Any
from .atoms import Atom, atom, atom_to_json
from .cert import SPEC_SCHEMA, fingerprint
__all__ = [
"SmtLibError",
"SmtLibUnsupported",
"export_spec",
"export_obligation",
"import_spec",
"parse_sexpr",
]
class SmtLibError(ValueError):
"""The SMT-LIB text could not be read."""
class SmtLibUnsupported(SmtLibError):
"""A construct outside certkit's fragment. Names what it refused."""
# --------------------------------------------------------------------------- #
# export
# --------------------------------------------------------------------------- #
def _term(a: Atom, *, negated: bool = False) -> str:
"""Render one atom as an SMT-LIB assertion body.
An atom is ``sum(c*v) + k <= 0`` (or ``< 0``). Coefficients are exact
rationals; when any denominator is not 1 we scale the whole atom by the LCM
of the denominators, which preserves the relation exactly because the scale
is positive. That keeps the emitted script in integer arithmetic, so a
solver can use QF_LIA rather than QF_LRA.
"""
denominators = [c.denominator for c in a.coeff.values()] + [a.const.denominator]
scale = 1
for d in denominators:
scale = scale * d // _gcd(scale, d)
parts = []
for v in sorted(a.coeff):
c = a.coeff[v] * scale
assert c.denominator == 1
n = int(c)
if n == 0:
continue
parts.append(v if n == 1 else f"(* {_numeral(n)} {v})")
const = a.const * scale
assert const.denominator == 1
k = int(const)
if not parts:
lhs = _numeral(k)
elif k:
lhs = f"(+ {' '.join(parts)} {_numeral(k)})"
elif len(parts) == 1:
lhs = parts[0]
else:
lhs = f"(+ {' '.join(parts)})"
op = "<" if a.strict else "<="
body = f"({op} {lhs} 0)"
return f"(not {body})" if negated else body
def _numeral(n: int) -> str:
"""Render an integer as an SMT-LIB 2 term.
``-1`` is not a numeral in SMT-LIB 2 -- the grammar's numerals are
non-negative, and a leading minus makes it a *symbol*. z3 accepts it as an
extension; a standards-strict cvc5 reports `Symbol '-1' not declared as a
variable` and refuses the whole script. The portable spelling is ``(- 1)``.
"""
return str(n) if n >= 0 else f"(- {-n})"
def _gcd(a: int, b: int) -> int:
while b:
a, b = b, a % b
return abs(a) or 1
def _variables(atoms: list[Atom]) -> list[str]:
seen: set[str] = set()
for a in atoms:
seen.update(v for v, c in a.coeff.items() if c != 0)
return sorted(seen)
def export_obligation(
domain: list[Atom],
guard: list[Atom],
safety: list[Atom],
index: int,
*,
name: str = "obligation",
) -> str:
"""Emit `domain AND guard AND NOT(safety[index])` as an SMT-LIB 2 script."""
if not safety:
raise SmtLibError("a spec needs at least one safety conjunct")
if not 0 <= index < len(safety):
raise SmtLibError(f"safety_index {index} out of range (0..{len(safety) - 1})")
all_atoms = list(domain) + list(guard) + [safety[index]]
lines = [
f"; certkit obligation {index} of {len(safety)} -- {name}",
";",
"; unsat => the guard implies this safety conjunct over the declared domain.",
"; sat => it does not, and the model is a counterexample.",
";",
"; Integers, not machine words: this says nothing about wraparound.",
"(set-logic QF_LIA)",
"",
]
for v in _variables(all_atoms):
lines.append(f"(declare-const {v} Int)")
lines.append("")
for label, group in (("domain", domain), ("guard", guard)):
if group:
lines.append(f"; {label}")
lines += [f"(assert {_term(a)})" for a in group]
lines.append("")
lines.append(f"; negated safety conjunct {index}")
lines.append(f"(assert {_term(safety[index], negated=True)})")
# `(get-model)` is left commented out on purpose. It is an error after
# `unsat` -- which is the *success* case here -- and z3 then exits non-zero
# with `model is not available`, so the proved case would look like a broken
# run to anyone piping this into CI. Uncomment it when you get `sat` and want
# the counterexample.
lines += [
"",
"(check-sat)",
"; (get-model) ; uncomment after a `sat` result to see the counterexample",
"",
]
return "\n".join(lines)
def export_spec(spec: dict[str, Any]) -> list[str]:
"""One SMT-LIB script per safety conjunct, in spec order."""
from .cert import _parse_spec_atoms
prefix, safety = _parse_spec_atoms(spec)
n_domain = len(spec.get("domain", []) or [])
domain, guard = prefix[:n_domain], prefix[n_domain:]
name = spec.get("name", "unnamed")
return [export_obligation(domain, guard, safety, i, name=name) for i in range(len(safety))]
# --------------------------------------------------------------------------- #
# import -- restricted, and explicit about it
# --------------------------------------------------------------------------- #
_TOKEN = re.compile(r"\(|\)|[^\s()]+")
def parse_sexpr(text: str) -> list[Any]:
"""Tokenise SMT-LIB into nested lists. Comments and strings are stripped."""
text = re.sub(r";[^\n]*", "", text)
stack: list[list[Any]] = [[]]
for tok in _TOKEN.findall(text):
if tok == "(":
stack.append([])
elif tok == ")":
if len(stack) == 1:
raise SmtLibError("unbalanced ')' in SMT-LIB input")
done = stack.pop()
stack[-1].append(done)
else:
stack[-1].append(tok)
if len(stack) != 1:
raise SmtLibError("unbalanced '(' in SMT-LIB input")
return stack[0]
_REL_FLIP = {"<=": ">=", "<": ">", ">=": "<=", ">": "<", "=": "="}
def _linear(node: Any, sign: Fraction = Fraction(1)) -> tuple[dict[str, Fraction], Fraction]:
"""Evaluate a term into (coefficients, constant). Refuses anything nonlinear."""
if isinstance(node, str):
try:
return {}, sign * Fraction(node)
except ValueError:
pass
if not re.fullmatch(r"[A-Za-z_][A-Za-z_0-9]*", node):
raise SmtLibUnsupported(f"cannot read the term {node!r} as a variable or number")
return {node: sign}, Fraction(0)
if not node:
raise SmtLibError("empty term")
head, *rest = node
if head == "+":
coeff: dict[str, Fraction] = {}
const = Fraction(0)
for r in rest:
c, k = _linear(r, sign)
for v, x in c.items():
coeff[v] = coeff.get(v, Fraction(0)) + x
const += k
return coeff, const
if head == "-":
if len(rest) == 1:
return _linear(rest[0], -sign)
coeff, const = _linear(rest[0], sign)
for r in rest[1:]:
c, k = _linear(r, -sign)
for v, x in c.items():
coeff[v] = coeff.get(v, Fraction(0)) + x
const += k
return coeff, const
if head == "*":
# Exactly one non-constant factor, or it is not linear. Each factor is
# evaluated rather than pattern-matched on being a bare numeral, because
# a constant can arrive as a term: `(- 1)` is how SMT-LIB 2 spells -1,
# and treating it as the variable part made `(* (- 1) x)` look nonlinear.
factor = sign
var_part: Any = None
for r in rest:
coeff_r, const_r = _linear(r)
if not coeff_r:
factor *= const_r
continue
if var_part is not None:
raise SmtLibUnsupported(
"nonlinear multiplication: certkit handles linear arithmetic only"
)
var_part = r
if var_part is None:
return {}, factor
return _linear(var_part, factor)
raise SmtLibUnsupported(f"unsupported operator {head!r}; certkit handles + - * only")
def _atom_from_assert(node: Any) -> Atom:
"""Turn one `(assert ...)` body into a canonical atom."""
negated = False
if isinstance(node, list) and node and node[0] == "not":
if len(node) != 2:
raise SmtLibError("(not ...) takes exactly one argument")
negated, node = True, node[1]
if not isinstance(node, list) or len(node) != 3:
raise SmtLibUnsupported(
"each assertion must be a single binary comparison such as (<= (+ x 1) y)"
)
rel, lhs, rhs = node
if rel == "=":
raise SmtLibUnsupported(
"an equality is two atoms; assert the two inequalities you mean instead"
)
if rel not in _REL_FLIP:
raise SmtLibUnsupported(f"unsupported relation {rel!r}; use <=, <, >= or >")
lc, lk = _linear(lhs)
rc, rk = _linear(rhs)
if rel in (">=", ">"):
lc, rc, lk, rk = rc, lc, rk, lk
rel = "<=" if rel == ">=" else "<"
coeff = dict(lc)
for v, x in rc.items():
coeff[v] = coeff.get(v, Fraction(0)) - x
const = lk - rk
strict = rel == "<"
if negated:
coeff = {v: -x for v, x in coeff.items()}
const, strict = -const, not strict
coeff = {v: x for v, x in coeff.items() if x != 0}
if not coeff:
raise SmtLibUnsupported(
"an assertion with no variables is a constant claim, not a relation"
)
return atom(coeff, const, strict=strict)
_KNOWN_COMMANDS = {
"set-logic",
"set-info",
"set-option",
"declare-const",
"declare-fun",
"assert",
"check-sat",
"get-model",
"get-value",
"exit",
"push",
"pop",
"echo",
}
def import_spec(text: str, *, name: str = "imported") -> dict[str, Any]:
"""Read an SMT-LIB script into a certkit spec.
Every assertion becomes a **safety** conjunct, because the file states what
must hold and carries no notion of which relations are assumptions. Move
atoms into ``domain`` and ``guard`` yourself -- that is a modelling decision,
and guessing it would silently change what gets proved.
Raises :class:`SmtLibUnsupported`, naming the construct, for anything outside
quantifier-free linear integer arithmetic.
"""
forms = parse_sexpr(text)
asserts: list[Atom] = []
declared: set[str] = set()
for form in forms:
if not isinstance(form, list) or not form:
raise SmtLibError(f"expected a command, got {form!r}")
cmd = form[0]
if cmd == "assert":
if len(form) != 2:
raise SmtLibError("(assert ...) takes exactly one argument")
asserts.append(_atom_from_assert(form[1]))
elif cmd in ("declare-const", "declare-fun"):
if cmd == "declare-fun" and len(form) >= 3 and form[2]:
raise SmtLibUnsupported(
f"uninterpreted function {form[1]!r}: certkit has no function symbols"
)
sort = form[-1]
if sort not in ("Int",):
raise SmtLibUnsupported(
f"sort {sort!r} for {form[1]!r}: certkit handles Int only "
"(no Real, Bool, or BitVec)"
)
declared.add(form[1])
elif cmd in ("forall", "exists"):
raise SmtLibUnsupported("quantifiers: certkit's fragment is quantifier-free")
elif cmd not in _KNOWN_COMMANDS:
raise SmtLibUnsupported(f"unsupported command {cmd!r}")
if not asserts:
raise SmtLibError("no assertions found, so there is nothing to prove")
undeclared = sorted({v for a in asserts for v in a.coeff} - declared)
if undeclared:
raise SmtLibError(
f"assertion mentions undeclared variable(s): {', '.join(undeclared)}. "
"Add (declare-const <name> Int)."
)
body: dict[str, Any] = {
"schema": SPEC_SCHEMA,
"name": name,
"domain": [],
"guard": [],
"safety": [atom_to_json(a) for a in asserts],
}
body["fingerprint"] = fingerprint(dict(body.items()))
return body
"""One place that turns a :class:`CheckReport` into whatever your CI reads.
Every output format is a projection of the same verdict, so they live together:
the moment SARIF lives in the Action and JSON lives in the CLI, they drift, and
a refusal reason ends up worded one way in the log and another in the Security
tab. The oracle for this module is that no format may state a verdict the report
does not carry.
Formats:
text what a person reads in a terminal
json the report verbatim, for scripts
sarif SARIF 2.1.0, for GitHub code scanning and anything else that reads it
junit JUnit XML, which almost every CI system renders natively
markdown a table, for PR comments and job summaries
`UNVERIFIED` is deliberately not collapsed into either pass or fail anywhere. It
is its own level in SARIF, its own failure type in JUnit, and its own word in the
text and markdown. A format that rendered it as "passed" would undo the whole
point of having a third verdict.
"""
from __future__ import annotations
import json
from typing import Any
from xml.sax.saxutils import escape, quoteattr
from .cert import ACCEPTED, UNVERIFIED, CheckReport
__all__ = ["FORMATS", "render", "sarif_document", "sarif_result", "SARIF_RULES"]
FORMATS = ("text", "json", "sarif", "junit", "markdown")
SARIF_RULES: list[dict[str, Any]] = [
{
"id": "certkit/refused",
"name": "CertificateRefused",
"shortDescription": {"text": "A proof certificate did not check out"},
"fullDescription": {
"text": (
"The supplied multipliers do not refute the obligation rebuilt from this "
"spec. The safety property is NOT PROVEN over the declared domain. That "
"is not the same as the property being false -- certkit refuses, it never "
"certifies the negation."
)
},
"defaultConfiguration": {"level": "error"},
"help": {
"text": "Run `certkit explain --spec <spec> --cert <cert>` to see the arithmetic."
},
},
{
"id": "certkit/unverified",
"name": "CertificateUnverified",
"shortDescription": {"text": "A certificate was not bound to its spec"},
"fullDescription": {
"text": (
"The multipliers checked out, but fingerprint verification was disabled, "
"so nothing establishes that this certificate was issued for this spec. "
"This is not an acceptance."
)
},
"defaultConfiguration": {"level": "error"},
"help": {"text": "Re-run without --no-fingerprint, or bind the certificate to the spec."},
},
]
def _rule_for(verdict: str) -> str:
return "certkit/unverified" if verdict == UNVERIFIED else "certkit/refused"
def sarif_result(verdict: str, location: str, message: str) -> dict[str, Any]:
"""One SARIF finding, anchored to the spec file it came from."""
return {
"ruleId": _rule_for(verdict),
"level": "error",
"message": {"text": message},
"locations": [
{
"physicalLocation": {
"artifactLocation": {"uri": location},
"region": {"startLine": 1},
}
}
],
}
def sarif_document(results: list[dict[str, Any]], verdict: str | None = None) -> dict[str, Any]:
"""Wrap findings in a SARIF 2.1.0 run.
Only the parts GitHub actually consumes are emitted; the full schema is
enormous and most of it would be noise.
``verdict`` is recorded as a run property. An ACCEPTED run has no results --
that is what SARIF means by a clean run -- but a file that does not say which
verdict produced it cannot be told apart from a run that never happened. The
stress suite found this: "the artifact is verdict-preserving" was true for
the two failing verdicts and silently false for the passing one.
"""
run: dict[str, Any] = {
"tool": {
"driver": {
"name": "certkit",
"informationUri": "https://github.com/nickharris808/certkit",
"rules": SARIF_RULES,
}
},
"results": results,
}
if verdict is not None:
run["properties"] = {"certkit.verdict": verdict}
return {
"$schema": "https://json.schemastore.org/sarif-2.1.0.json",
"version": "2.1.0",
"runs": [run],
}
def _reason_of(report: CheckReport) -> str:
detail = "; ".join(o["reason"] for o in report.obligations if o.get("reason"))
return detail or report.reason or report.verdict
def _text(report: CheckReport, name: str) -> str:
lines = [f"{report.verdict}: {name}"]
if report.reason:
lines.append(f" reason: {report.reason}")
for ob in report.obligations:
mark = "ok " if ob["ok"] else "FAIL"
extra = f" -- {ob['reason']}" if ob.get("reason") else ""
lines.append(f" [{mark}] obligation {ob['index']}{extra}")
if not report.binding_verified and report.obligations_ok:
lines.append(
" NOTE: every obligation was refuted, but with no trust anchor this "
"is UNVERIFIED, not ACCEPTED. Re-run without --no-fingerprint."
)
return "\n".join(lines)
def _markdown(report: CheckReport, name: str) -> str:
head = ["| item | verdict | detail |", "|---|---|---|"]
mark = report.verdict if report.verdict == ACCEPTED else f"**{report.verdict}**"
head.append(f"| `{name}` | {mark} | {_reason_of(report) or 'all obligations discharged'} |")
return "\n".join(head)
def _junit(report: CheckReport, name: str) -> str:
"""JUnit XML. UNVERIFIED is a distinct failure type, never a pass."""
failed = 0 if report.ok else 1
body = ""
if not report.ok:
kind = "unverified" if report.verdict == UNVERIFIED else "refused"
body = (
f"\n <failure type={quoteattr(kind)} "
f"message={quoteattr(_reason_of(report))}>"
f"{escape(report.verdict)}: {escape(_reason_of(report))}</failure>\n "
)
# The verdict travels even on a pass. A JUnit file that only names the
# verdict when it is bad cannot be distinguished from one produced by a run
# that checked nothing.
if not body:
body = f"\n <system-out>{escape(report.verdict)}</system-out>\n "
return (
'<?xml version="1.0" encoding="UTF-8"?>\n'
f'<testsuites tests="1" failures="{failed}">\n'
f' <testsuite name="certkit" tests="1" failures="{failed}">\n'
f' <testcase classname="certkit" name={quoteattr(name)}>{body}</testcase>\n'
" </testsuite>\n"
"</testsuites>"
)
def render(report: CheckReport, fmt: str = "text", *, name: str = "<unnamed>") -> str:
"""Render a report in ``fmt``. Raises on an unknown format rather than guessing."""
if fmt not in FORMATS:
raise ValueError(f"unknown format {fmt!r}; choose one of: {', '.join(FORMATS)}")
if fmt == "json":
return json.dumps(report.to_dict(), indent=2)
if fmt == "sarif":
results = (
[]
if report.ok
else [sarif_result(report.verdict, name, f"{report.verdict}: {_reason_of(report)}")]
)
return json.dumps(sarif_document(results, report.verdict), indent=2)
if fmt == "junit":
return _junit(report, name)
if fmt == "markdown":
return _markdown(report, name)
return _text(report, name)
"""HTML rendering for notebooks, with the verdict discipline intact.
Jupyter calls ``_repr_html_()`` on whatever a cell evaluates to, so this is the
form most people will first see a result in. That makes it a place where a
verdict can quietly become something friendlier than the truth: a green tick, a
boolean, a "passed" badge. It does not, here.
Three colours for three verdicts, and ``UNVERIFIED`` gets its own -- neither the
green of an acceptance nor the red of a refusal, because it is neither. It is the
tool declining to answer, and a renderer that showed it as either would undo the
reason the third verdict exists.
The rendering imports nothing: no template engine, no CSS framework, no
JavaScript. It is `html.escape` and an f-string, so a hostile spec name cannot
inject markup into your notebook.
"""
from __future__ import annotations
from html import escape
from typing import Any
from .cert import ACCEPTED, REFUSED, UNVERIFIED
__all__ = ["report_html", "COLOURS"]
#: Border/background per verdict. UNVERIFIED is deliberately not a shade of
#: either of the others.
COLOURS: dict[str, tuple[str, str]] = {
ACCEPTED: ("#0f7b3f", "#eaf6ee"),
REFUSED: ("#a41b1b", "#fbeaea"),
UNVERIFIED: ("#8a5a00", "#fff6e5"),
}
_MEANING = {
ACCEPTED: "Every obligation was refuted, and the certificate is bound to this spec.",
REFUSED: (
"At least one obligation was not refuted. <b>This is not a proof that the "
"guard is unsound</b> — only that this certificate does not establish it."
),
UNVERIFIED: (
"The arithmetic checked out, but a required precondition was never "
"established, so <b>no claim is being made</b>. This is not a pass."
),
}
def report_html(report: Any, *, name: str = "") -> str:
"""Render a :class:`~certkit.cert.CheckReport` as standalone HTML.
``name`` is the spec name, shown escaped. Nothing here trusts its input:
a spec called ``<script>`` renders as text.
"""
verdict = report.verdict
fg, bg = COLOURS.get(verdict, ("#333333", "#f2f2f2"))
rows = []
for o in report.obligations:
mark = "ok" if o.get("ok") else "FAIL"
reason = escape(str(o.get("reason") or ""))
rows.append(
f'<tr><td style="padding:2px 10px 2px 0">obligation {int(o.get("index", 0))}</td>'
f'<td style="padding:2px 10px 2px 0"><code>{mark}</code></td>'
f'<td style="padding:2px 0"><small>{reason}</small></td></tr>'
)
table = (
f'<table style="border-collapse:collapse;margin-top:6px">{"".join(rows)}</table>'
if rows
else ""
)
title = escape(name) if name else "certificate"
reason = escape(report.reason or "")
reason_html = f'<div style="margin-top:6px"><small>{reason}</small></div>' if reason else ""
return (
f'<div style="border-left:4px solid {fg};background:{bg};padding:10px 14px;'
f"font-family:ui-monospace,SFMono-Regular,Menlo,monospace;font-size:13px;"
f'color:#111;max-width:46em">'
f'<div style="font-weight:700;color:{fg}">{escape(verdict)}</div>'
f"<div>{title}</div>"
f'<div style="margin-top:6px;font-family:system-ui,sans-serif">{_MEANING.get(verdict, "")}</div>'
f"{table}{reason_html}"
f"</div>"
)
"""A Language Server for certkit specifications.
certkit lsp # speaks LSP over stdio
Verification that only happens in CI arrives after you have stopped thinking
about the guard. This moves it to the moment you write it: open a
``*.spec.json`` in any editor with an LSP client and the diagnostics appear as
you type.
What it reports, and nothing beyond it:
* **Structure** -- a spec that does not match the published JSON Schema, with the
offending field named.
* **Binding** -- when a ``*.cert.json`` sits beside the spec, the verdict of
actually checking it, with the binding check on. A certificate carrying no
fingerprint is refused here exactly as ``certkit check`` refuses it: the editor
and the gate must not disagree about what a file means.
* **Modelling smells** -- a variable that appears in the guard or the safety
property but is bounded by nothing in the domain. That is not an error and is
not reported as one. An unbounded variable makes the claim *stronger* (the
obligation must hold for every value it could take) and therefore harder to
prove, which is worth knowing when a guard you believe is correct will not
verify.
What it does **not** do is guess. There is no producer here, so it never offers
to "fix" a guard, and it never reports a spec as correct -- the absence of
diagnostics means nothing was found wrong with the *file*, which is a much
smaller claim than the guard being right. That distinction is the whole reason
the checker exists, and an editor is exactly where it would be tempting to blur.
The implementation is standard library only: a JSON-RPC framing loop over stdin
and stdout. No pygls, no dependency, and small enough to read in one sitting --
the same property the checker itself is built around.
"""
from __future__ import annotations
import json
import re
import sys
from pathlib import Path
from typing import Any, BinaryIO
from urllib.parse import unquote, urlparse
from .cert import ACCEPTED, check_certificate
from .schemas import load_schema
__all__ = ["diagnostics_for", "uri_to_path", "serve", "main", "ERROR", "WARNING", "INFORMATION"]
ERROR = 1
WARNING = 2
INFORMATION = 3
HINT = 4
def uri_to_path(uri: str) -> Path | None:
"""``file://`` URI to a path. Returns ``None`` for anything else.
A server that guessed at an unfamiliar scheme would end up reading a file the
editor never mentioned.
On Windows an editor sends ``file:///c%3A/x/a.spec.json``, whose parsed path
is ``/c:/x/a.spec.json`` -- a leading slash before the drive letter that
makes the path invalid. It is stripped here. The Windows CI matrix is what
surfaced this; a POSIX-only test would have passed forever.
"""
parsed = urlparse(uri)
if parsed.scheme != "file":
return None
path = unquote(parsed.path)
if re.match(r"^/[A-Za-z]:", path):
path = path[1:]
return Path(path)
def _line_of(text: str, needle: str) -> int:
"""The zero-based line where ``needle`` first appears, or 0.
Diagnostics need a position and JSON has no line information once parsed.
Searching the raw text is approximate; being one line off is better than
dropping the diagnostic, and it never changes what is reported.
"""
for i, line in enumerate(text.splitlines()):
if needle in line:
return i
return 0
def _diagnostic(line: int, severity: int, message: str, code: str) -> dict[str, Any]:
return {
"range": {
"start": {"line": line, "character": 0},
"end": {"line": line, "character": 200},
},
"severity": severity,
"source": "certkit",
"code": code,
"message": message,
}
def _validate_structure(spec: Any, text: str) -> list[dict[str, Any]]:
"""Shape checks that do not need a JSON Schema library.
`jsonschema` is a development dependency of this package and must not become
a runtime one, so the structural rules the schema encodes are re-stated here
for the handful of fields an editor can usefully complain about.
"""
out: list[dict[str, Any]] = []
if not isinstance(spec, dict):
return [_diagnostic(0, ERROR, "a spec must be a JSON object", "spec/not-object")]
schema_id = load_schema("certkit/spec/v1").get("properties", {}).get("schema", {})
expected = schema_id.get("const", "certkit/spec/v1")
if spec.get("schema") != expected:
out.append(
_diagnostic(
_line_of(text, '"schema"'),
ERROR,
f"expected schema {expected!r}, found {spec.get('schema')!r}",
"spec/schema",
)
)
for field in ("domain", "guard", "safety"):
value = spec.get(field)
if value is None:
if field not in spec:
# Absent is fine for domain and guard (an empty conjunction), and
# fatal for safety.
if field == "safety":
out.append(
_diagnostic(
0, ERROR, "a spec needs at least one safety conjunct", "spec/safety"
)
)
continue
# Present but null is not an empty list. The checker refuses it, so
# the editor must not show a clean file. The stress suite found this:
# `"guard": null` produced no diagnostic at all.
out.append(
_diagnostic(
_line_of(text, f'"{field}"'),
ERROR,
f"'{field}' is null. Use [] for an empty conjunction, or remove the key.",
f"spec/{field}",
)
)
continue
if not isinstance(value, list):
out.append(
_diagnostic(
_line_of(text, f'"{field}"'),
ERROR,
f"'{field}' must be a list of atoms, found {type(value).__name__}",
f"spec/{field}",
)
)
continue
for i, atom in enumerate(value):
if not isinstance(atom, dict) or not isinstance(atom.get("coeff"), dict):
out.append(
_diagnostic(
_line_of(text, f'"{field}"'),
ERROR,
f"{field}[{i}] is not an atom: it needs a 'coeff' object",
f"spec/{field}/atom",
)
)
continue
const = atom.get("const")
if const is not None and not (isinstance(const, list) and len(const) == 2):
out.append(
_diagnostic(
_line_of(text, f'"{field}"'),
ERROR,
f"{field}[{i}] 'const' must be [numerator, denominator]",
f"spec/{field}/const",
)
)
for var, pair in atom["coeff"].items():
out.extend(_pair_problems(pair, text, var, f"{field}[{i}] coefficient for {var!r}"))
if const is not None:
out.extend(_pair_problems(const, text, f'"{field}"', f"{field}[{i}] 'const'"))
if not spec.get("safety"):
out.append(
_diagnostic(0, ERROR, "a spec needs at least one safety conjunct", "spec/safety")
)
return out
def _pair_problems(pair: Any, text: str, needle: str, label: str) -> list[dict[str, Any]]:
"""Validate one ``[numerator, denominator]`` pair.
Checking only the *shape* was not enough, and the stress suite is what showed
it: ``[NaN, 1]``, ``[Infinity, 1]``, ``[[[1]], 1]`` and ``[1, 0]`` are all
two-element lists, so the editor reported a clean file for four specs the
checker refuses outright. An editor that disagrees with the gate is worse
than an editor with no diagnostics.
"""
if not (isinstance(pair, list) and len(pair) == 2):
return [
_diagnostic(
_line_of(text, needle),
ERROR,
f"{label} must be [numerator, denominator]",
"spec/pair",
)
]
numerator, denominator = pair
for part, what in ((numerator, "numerator"), (denominator, "denominator")):
if isinstance(part, bool) or not isinstance(part, int):
shown = "a nested list" if isinstance(part, list) else repr(part)
return [
_diagnostic(
_line_of(text, needle),
ERROR,
f"{label}: the {what} must be a whole number, found {shown}. "
"Rationals travel as exact integer pairs precisely so that no "
"verdict depends on float formatting.",
"spec/pair",
)
]
if denominator == 0:
return [
_diagnostic(
_line_of(text, needle),
ERROR,
f"{label} has a zero denominator",
"spec/zero-denominator",
)
]
return []
def _unbounded_variables(spec: dict[str, Any]) -> set[str]:
"""Variables the guard or safety mentions that the domain never constrains."""
def names(field: str) -> set[str]:
found: set[str] = set()
for atom in spec.get(field) or []:
if isinstance(atom, dict) and isinstance(atom.get("coeff"), dict):
found |= {v for v, c in atom["coeff"].items() if c != [0, 1]}
return found
return (names("guard") | names("safety")) - names("domain")
def diagnostics_for(text: str, path: Path | None = None) -> list[dict[str, Any]]:
"""Every diagnostic for one spec document.
An empty list means nothing was found wrong with the *file*. It is not a
statement that the guard is right, and this server never makes one.
"""
try:
spec = json.loads(text)
except json.JSONDecodeError as exc:
return [_diagnostic(max(exc.lineno - 1, 0), ERROR, f"invalid JSON: {exc.msg}", "json")]
out = _validate_structure(spec, text)
if any(d["severity"] == ERROR for d in out):
# Structure first. Reporting a verdict for a spec that does not parse
# would be a verdict about something other than what the author wrote.
return out
for var in sorted(_unbounded_variables(spec)):
out.append(
_diagnostic(
_line_of(text, var),
INFORMATION,
f"{var!r} is used but never bounded by the domain, so the obligation must "
"hold for every value it could take -- including values your program cannot "
"produce. That makes the claim stronger, not weaker, and it makes the "
"obligation harder to refute. If a guard you believe is correct will not "
"verify, a missing domain bound is the first thing to check.",
"spec/unbounded",
)
)
cert_path = _certificate_beside(path)
if cert_path is not None:
try:
cert = json.loads(cert_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
out.append(
_diagnostic(0, WARNING, f"cannot read {cert_path.name}: {exc}", "cert/unreadable")
)
return out
report = check_certificate(spec, cert)
if report.verdict == ACCEPTED:
out.append(
_diagnostic(
0,
INFORMATION,
f"{cert_path.name} verifies against this spec ({len(report.obligations)} "
"obligation(s)). This says the certificate proves the guard implies safety "
"over the declared domain -- not that the domain describes your program.",
"cert/accepted",
)
)
else:
for o in report.obligations:
if not o["ok"]:
out.append(
_diagnostic(
_line_of(text, '"safety"'),
WARNING,
f"{cert_path.name} does not refute obligation {o['index']}: "
f"{o['reason']}. This is not a proof that the guard is wrong.",
"cert/refused",
)
)
if not report.obligations:
out.append(
_diagnostic(0, WARNING, f"{cert_path.name}: {report.reason}", "cert/refused")
)
return out
def _certificate_beside(path: Path | None) -> Path | None:
if path is None:
return None
name = path.name
if name.endswith(".spec.json"):
candidate = path.with_name(name[: -len(".spec.json")] + ".cert.json")
else:
candidate = path.with_suffix(".cert.json")
return candidate if candidate.is_file() else None
# --------------------------------------------------------------------------- #
# the protocol
# --------------------------------------------------------------------------- #
def _read_message(stream: BinaryIO) -> dict[str, Any] | None:
"""Read one LSP message. Returns ``None`` at end of input."""
length = 0
while True:
line = stream.readline()
if not line:
return None
line = line.strip()
if not line:
break
if line.lower().startswith(b"content-length:"):
try:
length = int(line.split(b":", 1)[1])
except ValueError:
return None
if length <= 0:
return None
body = stream.read(length)
try:
return json.loads(body.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError):
# A malformed frame is dropped rather than crashing the server; the
# editor is free to send another.
return {}
def _write_message(stream: BinaryIO, payload: dict[str, Any]) -> None:
body = json.dumps(payload).encode("utf-8")
stream.write(f"Content-Length: {len(body)}\r\n\r\n".encode("ascii"))
stream.write(body)
stream.flush()
CAPABILITIES = {
"textDocumentSync": {"openClose": True, "change": 1, "save": True},
"positionEncoding": "utf-16",
}
def serve(stdin: BinaryIO, stdout: BinaryIO) -> int:
"""Run the server loop until the client closes the connection."""
documents: dict[str, str] = {}
def publish(uri: str) -> None:
text = documents.get(uri, "")
path = uri_to_path(uri)
_write_message(
stdout,
{
"jsonrpc": "2.0",
"method": "textDocument/publishDiagnostics",
"params": {"uri": uri, "diagnostics": diagnostics_for(text, path)},
},
)
while True:
message = _read_message(stdin)
if message is None:
return 0
if not message:
continue
method = message.get("method")
msg_id = message.get("id")
params = message.get("params") or {}
if method == "initialize":
_write_message(
stdout,
{
"jsonrpc": "2.0",
"id": msg_id,
"result": {
"capabilities": CAPABILITIES,
"serverInfo": {"name": "certkit-lsp", "version": _version()},
},
},
)
elif method == "shutdown":
_write_message(stdout, {"jsonrpc": "2.0", "id": msg_id, "result": None})
elif method == "exit":
return 0
elif method == "textDocument/didOpen":
doc = params.get("textDocument", {})
documents[doc.get("uri", "")] = doc.get("text", "")
publish(doc.get("uri", ""))
elif method == "textDocument/didChange":
uri = params.get("textDocument", {}).get("uri", "")
changes = params.get("contentChanges") or []
if changes:
# Full-document sync (the capability advertised above), so the
# last change carries the whole file.
documents[uri] = changes[-1].get("text", "")
publish(uri)
elif method == "textDocument/didSave":
publish(params.get("textDocument", {}).get("uri", ""))
elif method == "textDocument/didClose":
documents.pop(params.get("textDocument", {}).get("uri", ""), None)
elif msg_id is not None:
# An unimplemented request must be answered, or the editor waits
# forever. Answering "method not found" is the honest reply.
_write_message(
stdout,
{
"jsonrpc": "2.0",
"id": msg_id,
"error": {"code": -32601, "message": f"method not found: {method}"},
},
)
def _version() -> str:
from . import __version__
return __version__
def main(argv: Any = None) -> int:
return serve(sys.stdin.buffer, sys.stdout.buffer)
if __name__ == "__main__":
raise SystemExit(main())
"""Render a refutation as arithmetic a human can follow.
A Farkas certificate is a vector of numbers. That is wonderful for a machine and
close to useless for a person: ``{"2": 1, "3": 1}`` does not look like a proof,
it looks like a hash. But the underlying argument is genuinely simple, and a
reader who is asked to trust the format deserves to see it worked out:
take each atom, multiply it by its weight, add them up, observe that every
variable cancels, observe that the surviving constant is impossible.
That is the whole proof, and this module prints exactly those steps with the
real numbers in them. It re-derives everything from the spec and the multipliers
rather than trusting any narrative the certificate carries, so the explanation
cannot disagree with the verdict.
Nothing here is part of the trusted path: :func:`certkit.verify_farkas` decides,
this only describes. If the two ever disagree, the checker is right.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from fractions import Fraction
from typing import Any
from .atoms import Atom
from .cert import reconstruct_obligation
from .farkas import _as_fraction, verify_farkas
__all__ = ["format_atom", "explain_obligation", "explain_certificate"]
def _num(x: Fraction) -> str:
"""A rational, printed the way a person writes it."""
return str(x.numerator) if x.denominator == 1 else f"{x.numerator}/{x.denominator}"
def format_atom(a: Atom) -> str:
"""Render an atom as the inequality it means, e.g. ``payload - record_len + 19 <= 0``."""
parts: list[str] = []
for v in sorted(a.coeff):
c = a.coeff[v]
if c == 0:
continue
sign = "-" if c < 0 else "+"
mag = abs(c)
term = v if mag == 1 else f"{_num(mag)}*{v}"
parts.append(f"{sign} {term}" if parts else (f"-{term}" if c < 0 else term))
if a.const or not parts:
sign = "-" if a.const < 0 else "+"
parts.append(f"{sign} {_num(abs(a.const))}" if parts else _num(a.const))
return f"{' '.join(parts)} {'<' if a.strict else '<='} 0"
def explain_obligation(
atoms: Sequence[Atom],
multipliers: Mapping[Any, Any],
*,
index: int = 0,
) -> str:
"""Show the weighted sum that refutes ``atoms``, term by term."""
lines: list[str] = []
result = verify_farkas(atoms, multipliers)
lines.append(f"Obligation {index}: is this system satisfiable?")
lines.append("")
for i, a in enumerate(atoms):
lines.append(f" [{i}] {format_atom(a)}")
lines.append("")
if not result:
lines.append(f" The supplied multipliers do NOT refute it: {result.reason}")
lines.append("")
lines.append(" That means 'not proven'. It is not a proof that the system IS")
lines.append(" satisfiable -- certkit refuses, it never certifies the negation.")
return "\n".join(lines)
# Re-derive the combination so the arithmetic shown is the arithmetic checked.
used: list[tuple[int, Fraction, Atom]] = []
for raw_index, raw_weight in multipliers.items():
weight = _as_fraction(raw_weight)
if weight == 0:
continue
used.append((int(raw_index), weight, atoms[int(raw_index)]))
used.sort()
lines.append(" Multiply each atom by its nonnegative weight and add:")
lines.append("")
combined: dict[str, Fraction] = {}
const = Fraction(0)
any_strict = False
for i, weight, a in used:
lines.append(f" {_num(weight)} * [{i}] ({format_atom(a)})")
for v, c in a.coeff.items():
combined[v] = combined.get(v, Fraction(0)) + weight * c
const += weight * a.const
any_strict = any_strict or a.strict
lines.append("")
surviving = {v: c for v, c in combined.items() if c != 0}
cancelled = sorted(v for v, c in combined.items() if c == 0)
if cancelled:
lines.append(f" Every variable cancels: {', '.join(cancelled)} all sum to 0.")
else:
lines.append(" No variables appear in the sum.")
if surviving: # unreachable when verify_farkas said ok; kept honest anyway
lines.append(f" Left over: {surviving!r}")
lines.append("")
relation = "<" if any_strict else "<="
lines.append(f" What remains is: {_num(const)} {relation} 0")
lines.append("")
if any_strict:
lines.append(f" At least one atom was strict, so the sum is strict: {_num(const)} < 0.")
lines.append(
" That is false, so the system has no solution. Within the declared\n"
" domain, the guard implies the safety property."
)
return "\n".join(lines)
def explain_certificate(spec: Mapping[str, Any], cert: Mapping[str, Any]) -> str:
"""Explain every obligation of a spec/certificate pair."""
safety = spec.get("safety") or []
obligations = cert.get("obligations") or []
body = {k: v for k, v in spec.items() if k != "fingerprint"}
out: list[str] = []
name = spec.get("name", "<unnamed>")
out.append(f"certkit explain -- {name}")
out.append("=" * (18 + len(str(name))))
out.append("")
out.append(
"The claim: within the declared domain, the guard implies every safety\n"
"conjunct. Each conjunct becomes one obligation, refuted separately."
)
out.append("")
if not safety:
out.append("This spec has no safety conjuncts, so there is nothing to prove.")
return "\n".join(out)
for i in range(len(safety)):
sub = dict(body)
sub["safety_index"] = i
try:
atoms = reconstruct_obligation(sub)
except (ValueError, TypeError, KeyError, AttributeError, ArithmeticError) as exc:
out.append(f"Obligation {i}: cannot be rebuilt from the spec: {exc}")
out.append("")
continue
entry = obligations[i] if i < len(obligations) else {}
multipliers = entry.get("multipliers") if isinstance(entry, Mapping) else None
if not isinstance(multipliers, Mapping):
out.append(f"Obligation {i}: the certificate supplies no multipliers for it.")
out.append("")
continue
try:
out.append(explain_obligation(atoms, multipliers, index=i))
except (TypeError, ValueError, ArithmeticError) as exc:
out.append(f"Obligation {i}: multipliers are malformed: {exc}")
out.append("")
out.append("-" * 60)
out.append(
"The atoms above were rebuilt from the spec. Any atoms the certificate\n"
"carried were ignored, which is why a certificate proving some easier\n"
"unrelated system cannot pass here."
)
return "\n".join(out)
"""Turn a written-out inequality into a spec, so nobody hand-writes JSON atoms.
The canonical atom form is deliberate -- ``{"coeff": {"payload": [1, 1]}, ...}``
is unambiguous, exactly representable, and easy to check. It is also a miserable
thing to type, and requiring it before a newcomer can run anything is a barrier
with no payoff. This module parses the relation people actually write:
19 + payload <= record_len
0 <= payload
2*offset + len < size
and emits the canonical form. The parser is deliberately small: linear integer
relations only, one comparison per line. Anything it does not understand is an
error naming the offending text, never a guess -- a spec silently parsed into
the wrong relation would be far worse than a rejected one, because everything
downstream would then prove the wrong thing.
"""
from __future__ import annotations
import re
from fractions import Fraction
from .atoms import Atom, atom
from .cert import make_spec
__all__ = ["parse_relation", "build_spec", "RelationSyntaxError"]
class RelationSyntaxError(ValueError):
"""A relation could not be parsed. Carries the offending text."""
_TOKEN = re.compile(
r"""
\s*(?:
(?P<num>\d+)
| (?P<name>[A-Za-z_][A-Za-z_0-9]*)
| (?P<op><=|>=|<|>|==|=)
| (?P<sym>[+\-*])
)\s*
""",
re.VERBOSE,
)
_COMPARISONS = ("<=", ">=", "<", ">", "==", "=")
def _tokenise(text: str) -> list[tuple[str, str]]:
out: list[tuple[str, str]] = []
pos = 0
while pos < len(text):
m = _TOKEN.match(text, pos)
if not m or m.end() == pos:
raise RelationSyntaxError(
f"cannot parse {text[pos : pos + 12]!r} in {text!r}. Supported syntax is a "
f"linear relation such as '19 + payload <= record_len' or '2*n < size'."
)
kind = m.lastgroup or ""
out.append((kind, m.group(kind)))
pos = m.end()
return out
def _parse_side(tokens: list[tuple[str, str]], source: str) -> tuple[dict[str, Fraction], Fraction]:
"""Parse one side of a comparison into (coefficients, constant)."""
coeff: dict[str, Fraction] = {}
const = Fraction(0)
sign = 1
pending: Fraction | None = None
expect_term = True
i = 0
while i < len(tokens):
kind, value = tokens[i]
if kind == "sym" and value in "+-":
if expect_term and pending is None and not coeff and const == 0 and i == 0:
sign = -1 if value == "-" else 1
else:
sign = -1 if value == "-" else 1
expect_term = True
i += 1
continue
if kind == "num":
n = Fraction(int(value))
# "3*x" -- a coefficient; otherwise a bare constant.
if i + 1 < len(tokens) and tokens[i + 1] == ("sym", "*"):
pending = n
i += 2
continue
const += sign * n
sign = 1
expect_term = False
i += 1
continue
if kind == "name":
c = (pending if pending is not None else Fraction(1)) * sign
coeff[value] = coeff.get(value, Fraction(0)) + c
pending = None
sign = 1
expect_term = False
i += 1
continue
raise RelationSyntaxError(f"unexpected {value!r} in {source!r}")
if pending is not None:
raise RelationSyntaxError(f"trailing coefficient with no variable in {source!r}")
return coeff, const
def parse_relation(text: str) -> Atom:
"""Parse ``lhs <op> rhs`` into a canonical atom (``... <= 0`` or ``... < 0``).
``a <= b`` becomes ``a - b <= 0``; ``a >= b`` becomes ``b - a <= 0``. An
equality is rejected rather than silently split, because an atom is a single
inequality and turning ``=`` into two of them would change the shape of the
spec without saying so.
"""
source = text.strip()
if not source:
raise RelationSyntaxError("empty relation")
op = None
for candidate in _COMPARISONS:
idx = source.find(candidate)
if idx != -1 and (op is None or idx < op[1] or len(candidate) > len(op[0])):
if op is None or idx < op[1]:
op = (candidate, idx)
if op is None:
raise RelationSyntaxError(
f"no comparison operator in {source!r}. Write something like "
f"'19 + payload <= record_len'."
)
name, idx = op
# Prefer the two-character form when both match at the same position.
for longer in ("<=", ">=", "=="):
if source[idx : idx + 2] == longer:
name = longer
break
if name in ("==", "="):
raise RelationSyntaxError(
f"equality is not a single atom: {source!r}. Write the two inequalities "
f"you mean, e.g. 'a <= b' and 'b <= a'."
)
lhs_text, rhs_text = source[:idx], source[idx + len(name) :]
lhs = _parse_side(_tokenise(lhs_text), source)
rhs = _parse_side(_tokenise(rhs_text), source)
if name in ("<=", "<"):
left, right = lhs, rhs
else: # >= or > -- flip so the atom is always "<= 0" shaped
left, right = rhs, lhs
coeff: dict[str, Fraction] = dict(left[0])
for v, c in right[0].items():
coeff[v] = coeff.get(v, Fraction(0)) - c
const = left[1] - right[1]
coeff = {v: c for v, c in coeff.items() if c != 0}
if not coeff:
raise RelationSyntaxError(
f"{source!r} has no variables, so it is a constant claim, not a relation."
)
return atom(coeff, const, strict=name in ("<", ">"))
def build_spec(
domain: list[str],
guard: list[str],
safety: list[str],
name: str = "unnamed",
) -> dict:
"""Parse three lists of written relations into a fingerprinted spec."""
def parse_all(items: list[str], label: str) -> list[Atom]:
out = []
for text in items:
try:
out.append(parse_relation(text))
except RelationSyntaxError as exc:
raise RelationSyntaxError(f"in --{label} {text!r}: {exc}") from exc
return out
if not safety:
raise RelationSyntaxError(
"a spec needs at least one --safety relation: the property you want proved."
)
return make_spec(
parse_all(domain, "domain"),
parse_all(guard, "guard"),
parse_all(safety, "safety"),
name=name,
)
"""Machine-readable JSON Schemas for the certkit formats.
`SPEC.md` documents the format in prose, which is good for a reader and useless
for a tool. Anything that wants to *emit* certkit specs -- a solver front end, a
CI plugin, another language's implementation -- needs a definition it can check
itself against. These are it.
The schemas describe **shape**, and that is all. Passing validation means the
JSON is well-formed certkit; it says nothing about whether the relations are
true, whether the multipliers refute anything, or whether the spec describes your
program. `check_certificate` decides those. A schema is not a verdict.
Loading a schema needs no third-party package -- these are data files read with
:mod:`json`, so certkit keeps its zero-dependency property. Validating *against*
them needs a JSON Schema implementation, which is a development dependency here
and never imported at runtime.
"""
from __future__ import annotations
import json
from functools import cache
from pathlib import Path
from typing import Any
__all__ = ["SCHEMA_DIR", "SCHEMA_FILES", "load_schema", "schema_for"]
SCHEMA_DIR = Path(__file__).resolve().parent
#: Format identifier -> schema filename.
SCHEMA_FILES: dict[str, str] = {
"certkit/spec/v1": "certkit-spec-v1.schema.json",
"certkit/farkas/v1": "certkit-farkas-v1.schema.json",
}
@cache # schemas are immutable data; loading one twice is pure waste
def load_schema(name: str) -> dict[str, Any]:
"""Load a schema by format identifier, e.g. ``"certkit/spec/v1"``.
Raises :class:`KeyError` naming the known identifiers rather than returning
something empty, because a caller that silently validated against ``{}``
would believe everything.
"""
try:
filename = SCHEMA_FILES[name]
except KeyError:
known = ", ".join(sorted(SCHEMA_FILES))
raise KeyError(f"no schema for {name!r}; known formats are: {known}") from None
return json.loads((SCHEMA_DIR / filename).read_text(encoding="utf-8"))
def schema_for(document: Any) -> dict[str, Any]:
"""Pick the schema matching a document's own ``schema`` field."""
if not isinstance(document, dict):
raise TypeError(f"expected a JSON object, got {type(document).__name__}")
declared = document.get("schema")
if not isinstance(declared, str):
raise KeyError("document has no 'schema' field, so its format is unknown")
return load_schema(declared)
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://raw.githubusercontent.com/nickharris808/certkit/main/src/certkit/schemas/certkit-spec-v1.schema.json",
"title": "certkit specification, v1",
"description": "The claim: within `domain`, `guard` implies every conjunct of `safety`. Relations are linear over the rationals, with exact [numerator, denominator] coefficients. This schema constrains shape only -- it cannot tell you whether the relations describe your program, which is the human's job.",
"type": "object",
"required": ["schema", "safety"],
"properties": {
"schema": {
"const": "certkit/spec/v1"
},
"name": {
"type": "string",
"description": "A label for humans. Not load-bearing, but it is covered by the fingerprint."
},
"domain": {
"type": "array",
"description": "Bounds on the attacker-controlled inputs. Assumed, not proved.",
"items": { "$ref": "#/$defs/atom" }
},
"guard": {
"type": "array",
"description": "The check the program performs.",
"items": { "$ref": "#/$defs/atom" }
},
"safety": {
"type": "array",
"description": "The property that must hold. A conjunction: one obligation per element.",
"minItems": 1,
"items": { "$ref": "#/$defs/atom" }
},
"safety_index": {
"type": "integer",
"minimum": 0,
"description": "Which safety conjunct a single reconstructed obligation refutes."
},
"fingerprint": {
"$ref": "#/$defs/sha256",
"description": "SHA-256 over the canonical JSON of this object with `fingerprint` removed. Detects drift; it is NOT a defence against a forger who controls the spec."
}
},
"$defs": {
"sha256": {
"type": "string",
"pattern": "^[0-9a-f]{64}$"
},
"rational": {
"description": "An exact rational as [numerator, denominator]; a bare integer is also accepted. The denominator must not be zero.",
"oneOf": [
{ "type": "integer" },
{
"type": "array",
"minItems": 2,
"maxItems": 2,
"items": { "type": "integer" },
"prefixItems": [{ "type": "integer" }, { "type": "integer", "not": { "const": 0 } }]
}
]
},
"atom": {
"type": "object",
"description": "A linear relation in canonical form: sum(coeff*var) + const <= 0, or < 0 when strict.",
"required": ["coeff"],
"properties": {
"coeff": {
"type": "object",
"additionalProperties": { "$ref": "#/$defs/rational" },
"propertyNames": { "pattern": "^[A-Za-z_][A-Za-z_0-9]*$" }
},
"const": { "$ref": "#/$defs/rational" },
"strict": { "type": "boolean" }
}
}
}
}
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://raw.githubusercontent.com/nickharris808/certkit/main/src/certkit/schemas/certkit-farkas-v1.schema.json",
"title": "certkit Farkas certificate, v1",
"description": "Nonnegative multipliers that refute each obligation of a spec. Note what is absent: a certificate carries no atoms. The checker rebuilds every obligation from the spec and lets the certificate supply only weights, so a certificate proving some easier unrelated system cannot pass. Any extra keys here are ignored by the checker.",
"type": "object",
"required": ["schema", "obligations"],
"properties": {
"schema": {
"const": "certkit/farkas/v1"
},
"spec_fingerprint": {
"type": "string",
"pattern": "^[0-9a-f]{64}$",
"description": "Binds this certificate to one spec. Required unless fingerprint checking is explicitly disabled, in which case the result can only ever be UNVERIFIED -- never ACCEPTED."
},
"obligations": {
"type": "array",
"description": "One entry per safety conjunct, in the same order as the spec's `safety` array.",
"items": {
"type": "object",
"required": ["multipliers"],
"properties": {
"multipliers": {
"type": "object",
"description": "Atom index -> nonnegative weight. Indices are positions in the RECONSTRUCTED obligation `domain ++ guard ++ [negate(safety[i])]`, not in anything the certificate carries. JSON object keys are strings, so indices are decimal strings.",
"minProperties": 1,
"propertyNames": { "pattern": "^(0|[1-9][0-9]*)$" },
"additionalProperties": { "$ref": "#/$defs/nonneg_rational" }
}
}
}
}
},
"$defs": {
"nonneg_rational": {
"description": "A nonnegative exact rational. Negative weights would flip an inequality and let you prove anything, so they are rejected by the checker.",
"oneOf": [
{ "type": "integer", "minimum": 0 },
{ "type": "string", "pattern": "^[0-9]+(/[1-9][0-9]*)?$" },
{
"type": "array",
"minItems": 2,
"maxItems": 2,
"prefixItems": [
{ "type": "integer", "minimum": 0 },
{ "type": "integer", "exclusiveMinimum": 0 }
],
"items": { "type": "integer" }
}
]
}
}
}
"""``certkit`` command-line entry point.
Exit codes are part of the contract, because this is meant to run in CI:
0 ACCEPTED -- every obligation refuted, and the certificate is bound to
this spec
1 REFUSED -- at least one obligation not refuted, or malformed input
2 usage error (missing file, unreadable JSON)
3 UNVERIFIED -- the arithmetic checked out but a required precondition was
never established (``--no-fingerprint``). No claim is made.
Note that exit 1 means "not proven", never "proven false". A refusal is a
refusal. Exit 3 is neither: it is the tool declining to certify something it
did not fully check, and CI should treat it as a failure, not a pass.
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Any
from .cert import UNVERIFIED, check_certificate
from .report import FORMATS, render
from .schemas import SCHEMA_FILES, load_schema
from .sos import verify_sos
def _load(path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def example_path(name: str) -> Path:
"""Path to a bundled example file, wherever the package is installed."""
return Path(__file__).resolve().parent / "examples" / name
def _demo() -> int:
"""Run the bundled Heartbleed example: one valid certificate, one forged.
This needs no files on disk and no repository checkout, so the documented
quickstart works immediately after `pip install certkit`.
"""
spec = _load(example_path("heartbleed.spec.json"))
good = _load(example_path("heartbleed.cert.json"))
forged = _load(example_path("heartbleed.forged.json"))
print("certkit demo -- CVE-2014-0160 (Heartbleed) shape")
print(" claim: guard `19 + payload <= record_len`")
print(" implies `3 + payload <= record_len` for payload in [0, 65535]")
print()
ok = check_certificate(spec, good)
print(f" valid certificate -> {ok.verdict}")
for obligation in ok.obligations:
detail = f" -- {obligation['reason']}" if obligation["reason"] else ""
print(
f" obligation {obligation['index']}: "
f"{'ok' if obligation['ok'] else 'FAIL'}{detail}"
)
bad = check_certificate(spec, forged)
print(f" forged certificate -> {bad.verdict}")
for obligation in bad.obligations:
detail = f" -- {obligation['reason']}" if obligation["reason"] else ""
print(
f" obligation {obligation['index']}: "
f"{'ok' if obligation['ok'] else 'FAIL'}{detail}"
)
print()
if ok and not bad:
print(" As expected: the real certificate checks out and the forgery does not.")
return 0
print(" UNEXPECTED: the demo did not behave as documented.")
return 1
def main(argv: Any = None) -> int:
parser = argparse.ArgumentParser(
prog="certkit",
description="Independently re-check a program-admission certificate.",
)
sub = parser.add_subparsers(dest="command", required=True)
p_check = sub.add_parser("check", help="check a Farkas certificate against a spec")
p_check.add_argument("--spec", required=True, type=Path)
p_check.add_argument("--cert", required=True, type=Path)
p_check.add_argument(
"--no-fingerprint",
action="store_true",
help=(
"skip the spec/certificate binding check. The result can then only ever be "
"UNVERIFIED (exit 3), never ACCEPTED -- an unbound certificate has not been "
"shown to be about this spec"
),
)
p_check.add_argument("--json", action="store_true", help="shorthand for --format json")
p_check.add_argument(
"--format",
choices=FORMATS,
default="text",
help=(
"output format. 'sarif' feeds GitHub code scanning; 'junit' is rendered "
"natively by most CI systems; 'markdown' suits a PR comment."
),
)
p_explain = sub.add_parser(
"explain",
help="show the refutation arithmetic in prose (which atoms, which weights, what cancels)",
)
p_explain.add_argument("--spec", required=True, type=Path)
p_explain.add_argument("--cert", required=True, type=Path)
p_init = sub.add_parser(
"init",
help="scaffold a spec from written relations, e.g. '19 + payload <= record_len'",
)
p_init.add_argument(
"--domain",
action="append",
default=[],
metavar="RELATION",
help="a bound on the attacker's inputs, e.g. '0 <= payload'. Repeatable.",
)
p_init.add_argument(
"--guard",
action="append",
default=[],
metavar="RELATION",
help="the check your code performs, e.g. '19 + payload <= record_len'. Repeatable.",
)
p_init.add_argument(
"--safety",
action="append",
default=[],
metavar="RELATION",
help="the property that must hold, e.g. '3 + payload <= record_len'. Repeatable.",
)
p_init.add_argument("--name", default="unnamed")
p_init.add_argument("-o", "--out", type=Path, help="write here instead of stdout")
p_export = sub.add_parser(
"export",
help="emit each obligation as an SMT-LIB 2 script a solver can check",
)
p_export.add_argument("--spec", required=True, type=Path)
p_export.add_argument(
"--smtlib", action="store_true", help="(the only format today; accepted for clarity)"
)
p_export.add_argument(
"-o", "--out", type=Path, help="write here; one file per obligation when there are several"
)
p_import = sub.add_parser(
"import",
help="read an SMT-LIB 2 script into a certkit spec (linear integer fragment only)",
)
p_import.add_argument("--smtlib", required=True, type=Path, metavar="FILE")
p_import.add_argument("--name", default="imported")
p_import.add_argument("-o", "--out", type=Path)
p_schema = sub.add_parser(
"schema",
help="print the JSON Schema for a certkit format, so other tools can emit it",
)
p_schema.add_argument(
"--format",
dest="schema_name",
default="certkit/spec/v1",
choices=sorted(SCHEMA_FILES),
)
p_sos = sub.add_parser("sos", help="check a sum-of-squares certificate")
p_sos.add_argument("--cert", required=True, type=Path)
sub.add_parser(
"lsp",
help="run a language server over stdio: diagnostics for spec files as you type",
)
sub.add_parser(
"demo",
help="run the bundled example (needs no files; works straight from pip install)",
)
args = parser.parse_args(argv)
if args.command == "demo":
return _demo()
if args.command == "lsp":
from .lsp import main as lsp_main
return lsp_main()
if args.command == "check":
try:
spec = _load(args.spec)
cert = _load(args.cert)
except (OSError, json.JSONDecodeError) as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
report = check_certificate(spec, cert, require_fingerprint=not args.no_fingerprint)
fmt = "json" if args.json else args.format
# SARIF anchors findings at a file, so pass the spec path rather than the
# spec's own name -- a code-scanning alert has to point somewhere real.
# `spec` is whatever was in the file, which may not be an object at all.
# check_certificate already refused it; reading .name here would turn a
# clean refusal into a traceback.
spec_name = spec.get("name", "<unnamed>") if isinstance(spec, dict) else "<unnamed>"
label = str(args.spec) if fmt == "sarif" else spec_name
print(render(report, fmt, name=label))
if report.verdict == UNVERIFIED:
return 3
return 0 if report.ok else 1
if args.command == "explain":
try:
spec = _load(args.spec)
cert = _load(args.cert)
except (OSError, json.JSONDecodeError) as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
from .explain import explain_certificate
print(explain_certificate(spec, cert))
# Explaining is not deciding. Exit status still reflects the verdict, so
# this can be dropped into a script without changing its meaning.
report = check_certificate(spec, cert)
if report.verdict == UNVERIFIED:
return 3
return 0 if report.ok else 1
if args.command == "init":
from .scaffold import RelationSyntaxError, build_spec
try:
spec = build_spec(args.domain, args.guard, args.safety, name=args.name)
except RelationSyntaxError as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
text = json.dumps(spec, indent=2)
if args.out:
try:
args.out.write_text(text + "\n", encoding="utf-8")
except OSError as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
n = len(spec["safety"])
print(f"wrote {args.out}")
print(
f"Next: find multipliers that refute each of the {n} obligation(s), then\n"
f" certkit check --spec {args.out} --cert your.cert.json"
)
else:
print(text)
return 0
if args.command == "export":
try:
spec = _load(args.spec)
except (OSError, json.JSONDecodeError) as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
from .smtlib import SmtLibError, export_spec
try:
scripts = export_spec(spec)
except (SmtLibError, ValueError, TypeError, KeyError, ArithmeticError) as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
if not args.out:
print(("\n" + ";" + "-" * 70 + "\n").join(scripts))
return 0
if len(scripts) == 1:
args.out.write_text(scripts[0], encoding="utf-8")
print(f"wrote {args.out}")
else:
stem = args.out.with_suffix("")
for i, s in enumerate(scripts):
path = Path(f"{stem}.{i}{args.out.suffix or '.smt2'}")
path.write_text(s, encoding="utf-8")
print(f"wrote {path}")
print("Check with: z3 <file> (unsat means the guard implies safety)")
return 0
if args.command == "import":
from .smtlib import SmtLibError, import_spec
try:
text = args.smtlib.read_text(encoding="utf-8")
except OSError as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
try:
spec = import_spec(text, name=args.name)
except SmtLibError as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
out = json.dumps(spec, indent=2)
if args.out:
args.out.write_text(out + "\n", encoding="utf-8")
print(f"wrote {args.out}")
print(
f"Every assertion became a safety conjunct ({len(spec['safety'])} of them). "
"Move the assumptions into 'domain' and the check into 'guard' yourself -- "
"an SMT-LIB file does not say which is which."
)
else:
print(out)
return 0
if args.command == "schema":
print(json.dumps(load_schema(args.schema_name), indent=2))
return 0
if args.command == "sos":
try:
cert = _load(args.cert)
except (OSError, json.JSONDecodeError) as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
ok = verify_sos(cert)
print("ACCEPTED" if ok else "REFUSED")
return 0 if ok else 1
return 2
if __name__ == "__main__":
raise SystemExit(main())
"""exploit-counter -- turn "we found no escapes" into an exact integer.
Counts, exactly, how many states a guard admits that a safety property forbids.
When the state space is too large to enumerate, falls back to a *sound* bracket
(Clopper-Pearson) rather than a point guess.
Quickstart::
from certkit import atom
from exploit_counter import over_acceptance
domain = [atom({"p": -1}), atom({"p": 1}, -65535)]
guard = [atom({"p": 1, "r": -1}, 19)]
safety = [atom({"p": 1, "r": -1}, 3)]
box = {"p": (0, 65535), "r": (0, 65535)}
result = over_acceptance(domain, guard, safety, box)
print(result.exact) # 0 -- the guard is sound
"""
from .api import Decision, OverAcceptance, count_models, decide_soundness, over_acceptance
from .approx import ApproxCount, clopper_pearson, monte_carlo_count
from .exact import (
DEFAULT_EXACT_CAP,
BoxError,
DegenerateBoxError,
InvertedBoxError,
UnboundedVariableError,
UnknownVariableError,
box_from_atoms,
box_volume,
count_conjunction,
enumeration_cost,
find_witness,
validate_box,
)
__version__ = "0.3.1"
__all__ = [
"over_acceptance",
"OverAcceptance",
"decide_soundness",
"Decision",
"count_models",
"count_conjunction",
"find_witness",
"box_from_atoms",
"box_volume",
"validate_box",
"enumeration_cost",
"BoxError",
"InvertedBoxError",
"UnboundedVariableError",
"UnknownVariableError",
"DegenerateBoxError",
"monte_carlo_count",
"clopper_pearson",
"ApproxCount",
"DEFAULT_EXACT_CAP",
"__version__",
]
"""Exact model counting for conjunctions of linear atoms over a bounded integer box.
The counting strategy is "enumerate all but one variable, close-form the last".
For a conjunction of linear atoms, once you fix integer values for every variable
except one target variable, each atom becomes a half-line constraint on the
target: ``c*target + rest {<,<=} 0`` where ``rest`` is now a constant. The
intersection of half-lines is an interval, and the number of integer points in an
interval is arithmetic, not search. So the cost is the product of the enumerated
variables' ranges -- and the *widest* variable is the one we close-form, because
that removes the largest factor from the product.
This is why the Heartbleed-shaped count is tractable. The payload variable ranges
over 2^16 values and the record-length variable over 2^32; enumerating payload
and closing the form over record_len costs 65,536 interval computations rather
than 2^48 point tests.
All arithmetic is exact integer/Fraction arithmetic. The result is a count, not
an estimate. When the enumerated product would exceed a cap, this module returns
``None`` rather than running for hours -- callers should fall back to the bounded
approximate leg in :mod:`exploit_counter.approx`.
"""
from __future__ import annotations
import math
from collections.abc import Iterable, Mapping, Sequence
from fractions import Fraction
from certkit import Atom
__all__ = [
"count_conjunction",
"find_witness",
"box_from_atoms",
"box_volume",
"validate_box",
"enumeration_cost",
"BoxError",
"InvertedBoxError",
"UnboundedVariableError",
"UnknownVariableError",
"DegenerateBoxError",
"DEFAULT_EXACT_CAP",
]
# Enumerating more than ~4 million points is where this stops being interactive.
DEFAULT_EXACT_CAP = 4_000_000
Box = dict[str, tuple[int, int]]
# --------------------------------------------------------------------------- #
# refusals
#
# Every one of these exists because the alternative was a confident-looking
# number that nothing had actually verified. A degenerate box counts zero
# escapes because there is nowhere for one to hide; an inverted box counts zero
# because it contains no points at all. Reporting "sound" in either case is an
# assertion the analysis did not earn, so the counter refuses instead.
# --------------------------------------------------------------------------- #
class BoxError(ValueError):
"""The declared box cannot support a trustworthy count."""
class InvertedBoxError(BoxError):
"""A variable's upper bound is below its lower bound, so the box is empty."""
class UnboundedVariableError(BoxError):
"""A variable has no finite bound, so it has no finite model count."""
class UnknownVariableError(BoxError):
"""An atom constrains a variable the box does not declare."""
class DegenerateBoxError(BoxError):
"""The box holds at most one point, so a zero count establishes nothing."""
def box_volume(box: Mapping[str, tuple[int, int]]) -> int:
"""Total number of integer points in the box (the denominator of a rate)."""
volume = 1
for lo, hi in box.values():
if hi < lo:
return 0
volume *= hi - lo + 1
return volume
def atom_variables(atoms: Iterable[Atom]) -> set[str]:
"""Every variable mentioned with a non-zero coefficient."""
seen: set[str] = set()
for a in atoms:
for v, c in a.coeff.items():
if c != 0:
seen.add(v)
return seen
def validate_box(
box: Mapping[str, tuple[int, int]],
atoms: Sequence[Atom] | None = None,
*,
allow_degenerate: bool = True,
) -> None:
"""Refuse a box that cannot carry a meaningful verdict.
Raises rather than returning a flag, because every caller that ignored this
would go on to report a count of zero -- and a zero from an empty box reads
exactly like a zero from a proof.
"""
for name, bounds in box.items():
try:
lo, hi = bounds
except (TypeError, ValueError) as exc:
raise BoxError(f"box entry for {name!r} must be a (lo, hi) pair") from exc
if hi < lo:
raise InvertedBoxError(
f"variable {name!r} has an inverted range [{lo}, {hi}]: the upper bound "
f"is below the lower bound, so the box contains no points at all. "
f"A count over an empty box is vacuously zero and proves nothing. "
f"Did you mean [{hi}, {lo}]?"
)
if atoms is not None:
undeclared = sorted(atom_variables(atoms) - set(box))
if undeclared:
known = ", ".join(sorted(box)) or "<none>"
raise UnknownVariableError(
f"the atoms constrain {', '.join(repr(v) for v in undeclared)}, which the "
f"box does not declare (box declares: {known}). An undeclared variable is "
f"unbounded and has no finite model count. Add a range for it, e.g. "
f"{undeclared[0]}=0:255."
)
if not allow_degenerate:
volume = box_volume(box)
if volume <= 1:
pinned = ", ".join(f"{k}={v[0]}" for k, v in box.items() if v[0] == v[1])
raise DegenerateBoxError(
f"the declared box holds {volume} point(s) ({pinned or 'no variables'}), so "
f"'no escapes found' would be true no matter how unsound the guard is. "
f"Declare a real range for each variable, or pass allow_degenerate=True if "
f"you genuinely mean to ask about a single point."
)
def enumeration_cost(box: Mapping[str, tuple[int, int]]) -> tuple[int, str | None]:
"""Points actually enumerated, and the variable that is closed-form instead.
This -- not the box volume -- is what the exact cap applies to. The counter
enumerates every variable except the widest, so a two-variable box of 2^32
points costs only 2^16 enumerations.
"""
variables = list(box)
if not variables:
return 1, None
widths = {v: max(0, box[v][1] - box[v][0] + 1) for v in variables}
target = max(variables, key=lambda v: widths[v])
product = 1
for v in variables:
if v != target:
product *= widths[v]
return product, target
def box_from_atoms(
variables: Sequence[str],
domain: Iterable[Atom],
default: tuple[int, int] | None = None,
) -> Box:
"""Derive per-variable integer bounds from single-variable domain atoms.
Only atoms mentioning exactly one variable can tighten a box; multi-variable
atoms are relations, not bounds, and are left for the counter to apply.
A variable that no atom bounds is **unbounded**, and this function will not
invent a range for it: it raises :class:`UnboundedVariableError` unless the
caller supplies ``default`` deliberately. Silently defaulting to ``(0, 0)``
was how a forgotten domain atom turned into a confident "sound" verdict.
"""
lows: dict[str, int | None] = dict.fromkeys(variables, None)
highs: dict[str, int | None] = dict.fromkeys(variables, None)
for a in domain:
if len(a.coeff) != 1:
continue
((v, c),) = a.coeff.items()
if v not in lows or c == 0:
continue
bound = Fraction(-a.const) / c
if c > 0: # v <= bound (strict: v < bound)
b = math.ceil(bound) - 1 if a.strict else math.floor(bound)
cur = highs[v]
highs[v] = b if cur is None else min(cur, b)
else: # v >= bound (strict: v > bound)
b = math.floor(bound) + 1 if a.strict else math.ceil(bound)
cur = lows[v]
lows[v] = b if cur is None else max(cur, b)
box: Box = {}
for v in variables:
lo, hi = lows[v], highs[v]
if lo is None or hi is None:
if default is None:
missing = "below" if lo is None else "above"
raise UnboundedVariableError(
f"variable {v!r} is unbounded {missing}: no single-variable domain atom "
f"constrains it. An unbounded variable has no finite model count, so no "
f"count over it would mean anything. Add a bounding atom, or pass an "
f"explicit default=(lo, hi) if you are deliberately choosing a range."
)
lo = default[0] if lo is None else lo
hi = default[1] if hi is None else hi
box[v] = (lo, hi)
return box
def _interval_for(
target: str,
atoms: Sequence[Atom],
assign: Mapping[str, int],
base: tuple[int, int],
) -> tuple[int, int] | None:
"""Intersect all atoms into one integer interval for ``target``.
Every variable other than ``target`` must be present in ``assign``. Returns
``None`` when the intersection is empty.
This is the general rational path. :func:`_compile_atoms` produces an
integer-only specialisation of the same arithmetic for the common case; the
two are checked against each other in the test suite.
"""
lo, hi = base
for a in atoms:
c = a.coeff.get(target, Fraction(0))
rest = a.const
for v, cv in a.coeff.items():
if v == target:
continue
rest += cv * assign[v]
if c == 0:
# The atom does not constrain target; it is a pure feasibility test
# on the current assignment.
violated = (rest >= 0) if a.strict else (rest > 0)
if violated:
return None
continue
bound = Fraction(-rest) / c
if c > 0: # target <= bound (strict: <)
b = math.ceil(bound) - 1 if a.strict else math.floor(bound)
hi = min(hi, b)
else: # target >= bound (strict: >)
b = math.floor(bound) + 1 if a.strict else math.ceil(bound)
lo = max(lo, b)
if lo > hi:
return None
return (lo, hi) if lo <= hi else None
# --------------------------------------------------------------------------- #
# integer specialisation of the inner loop
#
# Profiling the enumeration at its cap (500,000 points, 4 atoms) showed ~70% of
# the runtime inside `fractions.py` -- nine million Fraction constructions for
# half a million points. Real specs are bounds and length relations, so their
# coefficients are integers, and integers are a subset of the rationals: the
# arithmetic below is the same arithmetic, not an approximation of it.
#
# Atoms with a genuinely rational coefficient fall back to `_interval_for`.
# Nothing is rounded on either path, and `test_integer_and_rational_paths_agree`
# checks them against each other on random specs.
# --------------------------------------------------------------------------- #
def _compile_atoms(
atoms: Sequence[Atom], target: str
) -> list[tuple[int, int, tuple[tuple[str, int], ...], bool]] | None:
"""Pre-extract integer coefficients, or ``None`` if any is not an integer.
Returns one tuple per atom: the target's coefficient, the atom's constant,
the other variables' coefficients, and strictness. Hoisting this out of the
loop also removes a dict lookup and a dict iteration per atom per point.
"""
compiled: list[tuple[int, int, tuple[tuple[str, int], ...], bool]] = []
for a in atoms:
if a.const.denominator != 1:
return None
c_target = 0
others: list[tuple[str, int]] = []
for v, cv in a.coeff.items():
if cv.denominator != 1:
return None
if v == target:
c_target = int(cv)
elif cv:
others.append((v, int(cv)))
compiled.append((c_target, int(a.const), tuple(others), a.strict))
return compiled
def _floor_div(a: int, b: int) -> int:
"""floor(a / b) for non-zero b, exactly, in integers."""
if b < 0:
a, b = -a, -b
return a // b
def _ceil_div(a: int, b: int) -> int:
"""ceil(a / b) for non-zero b, exactly, in integers."""
if b < 0:
a, b = -a, -b
return -((-a) // b)
def _interval_for_int(
compiled: Sequence[tuple[int, int, tuple[tuple[str, int], ...], bool]],
assign: Mapping[str, int],
base: tuple[int, int],
) -> tuple[int, int] | None:
"""Integer-only twin of :func:`_interval_for`. Same result, no Fractions."""
lo, hi = base
for c, const, others, strict in compiled:
rest = const
for v, cv in others:
rest += cv * assign[v]
if c == 0:
if (rest >= 0) if strict else (rest > 0):
return None
continue
# target <= -rest/c when c > 0, target >= -rest/c when c < 0.
if c > 0:
b = _ceil_div(-rest, c) - 1 if strict else _floor_div(-rest, c)
if b < hi:
hi = b
else:
b = _floor_div(-rest, c) + 1 if strict else _ceil_div(-rest, c)
if b > lo:
lo = b
if lo > hi:
return None
return (lo, hi)
def count_conjunction(
atoms: Sequence[Atom],
box: Mapping[str, tuple[int, int]],
exact_cap: int = DEFAULT_EXACT_CAP,
) -> int | None:
"""Exact number of integer points in ``box`` satisfying every atom.
Returns ``None`` when the enumeration would exceed ``exact_cap`` -- that is a
refusal to spend unbounded time, not a count of zero. Callers must
distinguish the two.
Raises :class:`BoxError` when the box itself is unusable (inverted range, or
an atom naming a variable the box does not declare). Those were previously
reported as a count of zero or as ``None``, both of which are indistinguishable
from a real answer.
"""
validate_box(box, atoms)
variables = list(box.keys())
if not variables:
# No variables: the empty assignment satisfies the conjunction iff no
# atom is violated by it. validate_box has already established that no
# atom mentions a variable, so every atom here is a bare constant.
for a in atoms:
violated = (a.const >= 0) if a.strict else (a.const > 0)
if violated:
return 0
return 1
widths = {v: box[v][1] - box[v][0] + 1 for v in variables}
target = max(variables, key=lambda v: widths[v])
enum_vars = [v for v in variables if v != target]
product = 1
for v in enum_vars:
product *= widths[v]
if product > exact_cap:
return None
total = 0
assign: dict[str, int] = {}
compiled = _compile_atoms(atoms, target)
base = box[target]
if compiled is not None:
def rec(i: int) -> None:
nonlocal total
if i == len(enum_vars):
iv = _interval_for_int(compiled, assign, base)
if iv is not None:
total += iv[1] - iv[0] + 1
return
v = enum_vars[i]
lo, hi = box[v]
for val in range(lo, hi + 1):
assign[v] = val
rec(i + 1)
else:
def rec(i: int) -> None:
nonlocal total
if i == len(enum_vars):
iv = _interval_for(target, atoms, assign, base)
if iv is not None:
total += iv[1] - iv[0] + 1
return
v = enum_vars[i]
lo, hi = box[v]
for val in range(lo, hi + 1):
assign[v] = val
rec(i + 1)
rec(0)
return total
def find_witness(
atoms: Sequence[Atom],
box: Mapping[str, tuple[int, int]],
exact_cap: int = DEFAULT_EXACT_CAP,
) -> dict[str, int] | None:
"""Return one concrete integer point satisfying every atom, or ``None``.
A count tells you *how many* states escape; a witness tells you *which one*,
which is what a developer can actually act on. Returns the first point found
in enumeration order, so it is deterministic for a given box.
``None`` is ambiguous between "no witness exists" and "the search was capped".
Callers that need to distinguish should first check ``count_conjunction``,
which returns ``None`` specifically for the capped case. An unusable box is
never folded into that ambiguity -- it raises :class:`BoxError`.
"""
validate_box(box, atoms)
variables = list(box.keys())
if not variables:
for a in atoms:
if (a.const >= 0) if a.strict else (a.const > 0):
return None
return {}
widths = {v: box[v][1] - box[v][0] + 1 for v in variables}
target = max(variables, key=lambda v: widths[v])
enum_vars = [v for v in variables if v != target]
product = 1
for v in enum_vars:
product *= widths[v]
if product > exact_cap:
return None
assign: dict[str, int] = {}
compiled = _compile_atoms(atoms, target)
base = box[target]
def rec(i: int) -> dict[str, int] | None:
if i == len(enum_vars):
iv = (
_interval_for_int(compiled, assign, base)
if compiled is not None
else _interval_for(target, atoms, assign, base)
)
if iv is not None:
return {**assign, target: iv[0]}
return None
v = enum_vars[i]
lo, hi = box[v]
for val in range(lo, hi + 1):
assign[v] = val
found = rec(i + 1)
if found is not None:
return found
del assign[v]
return None
return rec(0)
"""Bounded approximate counting with an exact-binomial confidence interval.
When the exact leg refuses (the box is too big to enumerate), you can still get a
*sound* bracket rather than a guess. Sample the box uniformly, count hits, and
invert the binomial to a Clopper-Pearson interval -- then scale by the box volume.
Clopper-Pearson is the deliberate choice over the faster normal-approximation
intervals. It is *conservative*: it over-covers, so the true count is inside the
reported bracket at least as often as the stated confidence. For a risk figure
that a downstream decision depends on, over-covering is the correct failure
direction. A Wald interval is tighter and wrong near zero, which is exactly the
regime a security count lives in.
The zero-hit case is handled exactly: with 0 hits in n draws the upper bound is
``1 - (alpha/2)^(1/n)``, the rule-of-three family. That is why "we sampled and saw
nothing" produces a real bound rather than a claim of zero.
Sampling is seeded and deterministic so a reported figure is reproducible.
"""
from __future__ import annotations
import math
import random
from collections.abc import Mapping, Sequence
from dataclasses import asdict, dataclass
from typing import Any
from certkit import Atom
from .exact import box_volume
__all__ = ["clopper_pearson", "monte_carlo_count", "ApproxCount"]
# --------------------------------------------------------------------------- #
# regularized incomplete beta, and its inverse by bisection
#
# Provenance note. The continued fraction below is the standard one for the
# incomplete beta function, given as equation 26.5.8 in Abramowitz & Stegun,
# *Handbook of Mathematical Functions* (a US Government work, public domain).
# It is evaluated by the modified Lentz method (Lentz, *Applied Optics* 15:668,
# 1976; refined by Thompson & Barnett, *J. Comput. Phys.* 64:490, 1986), which is
# likewise published mathematics.
#
# This is a clean-room implementation written from those recurrences. It is
# deliberately *not* derived from any textbook's source listing, several of which
# are copyrighted and forbid redistribution. If you extend it, keep it that way:
# work from the published recurrence, not from someone else's code.
# --------------------------------------------------------------------------- #
# Terms of the continued fraction, as coefficient pairs.
_CF_MAX_TERMS = 400
_CF_TOLERANCE = 1e-15
_CF_FLOOR = 1e-300 # guards a zero denominator without branching on equality
def _cf_numerator(n: int, a: float, b: float, x: float) -> float:
"""The n-th numerator d_n of the incomplete-beta continued fraction.
Odd and even terms differ; A&S 26.5.8 gives
d_{2m} = +m(b - m)x / ((a + 2m - 1)(a + 2m))
d_{2m+1} = -(a + m)(a + b + m)x / ((a + 2m)(a + 2m + 1))
"""
if n % 2 == 0:
m = n // 2
return m * (b - m) * x / ((a + 2 * m - 1) * (a + 2 * m))
m = (n - 1) // 2
return -(a + m) * (a + b + m) * x / ((a + 2 * m) * (a + 2 * m + 1))
def _betacf(a: float, b: float, x: float) -> float:
"""Evaluate the A&S 26.5.8 continued fraction, bottom-up.
The fraction is
1 / (1 + d_1 / (1 + d_2 / (1 + ... )))
Evaluating it from the deepest term outwards is a two-line recurrence with no
special cases: set ``g = 1`` at the bottom and apply ``g <- 1 + d_n / g``
while walking ``n`` down to 1. The answer is ``1 / g``.
Forward evaluation (Lentz) would let us stop early on convergence, but it
needs guards against vanishing denominators and is fiddlier to get right. A
fixed depth is simpler and fast enough: this runs twice per confidence
interval, not in a hot loop. The depth is well past convergence for the
``a``, ``b`` values a binomial interval produces.
"""
g = 1.0
for n in range(_CF_MAX_TERMS, 0, -1):
d = _cf_numerator(n, a, b, x)
g = 1.0 + d / g
if abs(g) < _CF_FLOOR:
g = _CF_FLOOR
return 1.0 / g
def _betai(a: float, b: float, x: float) -> float:
"""Regularized incomplete beta function I_x(a, b)."""
if x <= 0.0:
return 0.0
if x >= 1.0:
return 1.0
lbeta = (
math.lgamma(a + b) - math.lgamma(a) - math.lgamma(b) + a * math.log(x) + b * math.log1p(-x)
)
front = math.exp(lbeta)
if x < (a + 1.0) / (a + b + 2.0):
return front * _betacf(a, b, x) / a
return 1.0 - front * _betacf(b, a, 1.0 - x) / b
def _beta_ppf(p: float, a: float, b: float) -> float:
"""Inverse of the regularized incomplete beta, by bisection.
Bisection rather than Newton: it is slower but cannot diverge, and this runs
twice per interval, not in a hot loop.
"""
if p <= 0.0:
return 0.0
if p >= 1.0:
return 1.0
lo, hi = 0.0, 1.0
for _ in range(200):
mid = (lo + hi) / 2.0
if _betai(a, b, mid) < p:
lo = mid
else:
hi = mid
return (lo + hi) / 2.0
def clopper_pearson(k: int, n: int, alpha: float = 0.05) -> tuple[float, float]:
"""Exact (Clopper-Pearson) confidence interval for a binomial proportion.
``k`` successes in ``n`` trials at confidence ``1 - alpha``. Conservative by
construction. Returns ``(0.0, 1.0)`` for ``n == 0`` -- no data, no bound.
"""
if n == 0:
return (0.0, 1.0)
if k < 0 or k > n:
raise ValueError("k must satisfy 0 <= k <= n")
a2 = alpha / 2.0
lo = 0.0 if k == 0 else _beta_ppf(a2, k, n - k + 1)
hi = 1.0 if k == n else _beta_ppf(1.0 - a2, k + 1, n - k)
return (lo, hi)
@dataclass
class ApproxCount:
"""A sampled count with a sound bracket.
``ci_low`` and ``ci_high`` bracket the true count with at least ``1 - alpha``
confidence. ``estimate`` is the point estimate and carries no guarantee on
its own -- report the bracket.
"""
estimate: int
ci_low: int
ci_high: int
n_samples: int
n_hits: int
alpha: float
box_volume: int
seed: int
method: str = "monte-carlo-clopper-pearson"
def brackets(self, exact: int) -> bool:
"""True when the reported interval contains a known exact count."""
return self.ci_low <= exact <= self.ci_high
def to_dict(self) -> dict[str, Any]:
return asdict(self)
def monte_carlo_count(
atoms: Sequence[Atom],
box: Mapping[str, tuple[int, int]],
n_samples: int = 200_000,
alpha: float = 0.05,
seed: int = 0,
) -> ApproxCount:
"""Uniform-sample estimate of the model count, with a Clopper-Pearson bracket.
Deterministic for a fixed seed, so a published figure is reproducible.
"""
if n_samples <= 0:
raise ValueError("n_samples must be positive")
rng = random.Random(seed)
variables = list(box.keys())
volume = box_volume(box)
hits = 0
for _ in range(n_samples):
assign = {v: rng.randint(box[v][0], box[v][1]) for v in variables}
ok = True
for a in atoms:
s = a.const
for v, c in a.coeff.items():
s += c * assign[v]
if (s >= 0) if a.strict else (s > 0):
ok = False
break
if ok:
hits += 1
p_lo, p_hi = clopper_pearson(hits, n_samples, alpha)
p_hat = hits / n_samples
return ApproxCount(
estimate=round(p_hat * volume),
ci_low=math.floor(p_lo * volume),
ci_high=math.ceil(p_hi * volume),
n_samples=n_samples,
n_hits=hits,
alpha=alpha,
box_volume=volume,
seed=seed,
)
"""The public counting API: over-acceptance of a guard, exactly or bracketed.
The headline question this package answers is:
"If I remove (or weaken) this guard, exactly how many states become
reachable that should not be?"
That number is the *over-acceptance* of the guard against a safety property, over
a declared domain. It converts "we found no escapes" -- which is unfalsifiable and
therefore unpersuasive -- into a specific integer with a specific meaning.
A guard that is sound has over-acceptance zero. A guard that is *nearly* sound
has a small positive count, and that count is exactly the attacker's search
space. It is also, under a uniform sampling model, the reciprocal of the expected
number of random draws needed to hit the gap -- which is how a blind spot that a
fuzzer will never find gets a defensible number attached to it.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from dataclasses import asdict, dataclass
from typing import Any
from certkit import Atom, negate
from .approx import ApproxCount, monte_carlo_count
from .exact import (
DEFAULT_EXACT_CAP,
box_volume,
count_conjunction,
enumeration_cost,
find_witness,
validate_box,
)
__all__ = [
"count_models",
"over_acceptance",
"OverAcceptance",
"decide_soundness",
"Decision",
]
def count_models(
atoms: Sequence[Atom],
box: Mapping[str, tuple[int, int]],
*,
exact_cap: int = DEFAULT_EXACT_CAP,
n_samples: int = 200_000,
alpha: float = 0.05,
seed: int = 0,
) -> tuple[int | None, ApproxCount | None]:
"""Count satisfying assignments, exactly if affordable else bracketed.
Returns ``(exact, None)`` when the exact leg succeeds, and
``(None, approx)`` when it refuses and the sampled leg is used. Exactly one
element is ever non-``None``, so callers cannot silently treat an estimate as
a count.
"""
exact = count_conjunction(atoms, box, exact_cap=exact_cap)
if exact is not None:
return exact, None
approx = monte_carlo_count(atoms, box, n_samples=n_samples, alpha=alpha, seed=seed)
return None, approx
@dataclass
class OverAcceptance:
"""How many states a guard admits that the safety property forbids."""
exact: int | None
approx: dict[str, Any] | None
domain_volume: int
n_obligations: int
method: str
@property
def is_exact(self) -> bool:
return self.exact is not None
def _repr_html_(self) -> str:
from .notebook import over_acceptance_html
return over_acceptance_html(self)
@property
def is_sound_guard(self) -> bool | None:
"""True when over-acceptance is exactly zero **over a box worth counting**.
``None`` when only a sampled bracket is available -- sampling cannot
establish a zero, only bound it. Callers must not coerce this to False.
``True`` here is an earned claim, not an absence of evidence: an empty or
single-point box cannot produce this object at all, because
:func:`over_acceptance` refuses those boxes up front. A zero over a box
with nowhere for a counterexample to hide is not a proof of soundness,
and it used to be reported as one.
"""
if self.exact is None:
return None
return self.exact == 0
def hit_probability(self) -> float | None:
"""Per-draw probability a uniform sampler lands in the gap."""
if self.exact is None or self.domain_volume == 0:
return None
return self.exact / self.domain_volume
def expected_draws_to_hit(self) -> float | None:
"""Expected uniform draws before the gap is hit; ``None`` if unreachable."""
p = self.hit_probability()
if p is None or p == 0:
return None
return 1.0 / p
def to_dict(self) -> dict[str, Any]:
return asdict(self)
def over_acceptance(
domain: Sequence[Atom],
guard: Sequence[Atom],
safety: Sequence[Atom],
box: Mapping[str, tuple[int, int]],
*,
exact_cap: int = DEFAULT_EXACT_CAP,
n_samples: int = 200_000,
alpha: float = 0.05,
seed: int = 0,
allow_degenerate: bool = False,
) -> OverAcceptance:
"""Count states admitted by ``guard`` but forbidden by ``safety``.
One region is counted per safety conjunct: ``domain AND guard AND NOT
safety[i]``. The regions may overlap, so the total is computed by counting
the union directly when there is one conjunct, and by summing per-conjunct
counts otherwise -- which **over-counts** when conjuncts overlap.
Over-counting is the safe direction for a risk figure: it never reports a
smaller attack surface than exists. The exact union requires
inclusion-exclusion over 2^n terms; for the single-conjunct case that
dominates real guards, the number reported is exact.
Raises :class:`~exploit_counter.exact.BoxError` when the declared box cannot
support a verdict -- inverted bounds, an atom naming an undeclared variable,
or a box holding at most one point. Each of those used to yield
``exact=0, is_sound_guard=True``, which is a soundness claim over a region
that contains nothing. Pass ``allow_degenerate=True`` to ask about a single
point on purpose.
"""
if not safety:
raise ValueError("safety must have at least one conjunct")
base: list[Atom] = list(domain) + list(guard)
# Validate against every atom that will be counted, so an undeclared variable
# is named here rather than surfacing as a KeyError from deep in the recursion.
validate_box(
box,
base + [negate(c) for c in safety],
allow_degenerate=allow_degenerate,
)
volume = box_volume(box)
if len(safety) == 1:
atoms = base + [negate(safety[0])]
exact, approx = count_models(
atoms, box, exact_cap=exact_cap, n_samples=n_samples, alpha=alpha, seed=seed
)
return OverAcceptance(
exact=exact,
approx=approx.to_dict() if approx else None,
domain_volume=volume,
n_obligations=1,
method="exact" if exact is not None else "sampled",
)
total = 0
all_exact = True
for conjunct in safety:
atoms = base + [negate(conjunct)]
exact = count_conjunction(atoms, box, exact_cap=exact_cap)
if exact is None:
all_exact = False
break
total += exact
if all_exact:
return OverAcceptance(
exact=total,
approx=None,
domain_volume=volume,
n_obligations=len(safety),
method="exact-union-upper-bound",
)
# Fall back to sampling the union directly, which needs no correction: a
# sample either violates some conjunct or it does not.
approx = _sample_union(base, safety, box, n_samples=n_samples, alpha=alpha, seed=seed)
return OverAcceptance(
exact=None,
approx=approx.to_dict(),
domain_volume=volume,
n_obligations=len(safety),
method="sampled-union",
)
def _sample_union(
base: Sequence[Atom],
safety: Sequence[Atom],
box: Mapping[str, tuple[int, int]],
*,
n_samples: int,
alpha: float,
seed: int,
) -> ApproxCount:
"""Sample ``base AND NOT(all safety conjuncts)`` as one region."""
import math
import random
from .approx import clopper_pearson
rng = random.Random(seed)
variables = list(box.keys())
volume = box_volume(box)
hits = 0
def holds(a: Atom, assign: Mapping[str, int]) -> bool:
s = a.const
for v, c in a.coeff.items():
s += c * assign[v]
return (s < 0) if a.strict else (s <= 0)
for _ in range(n_samples):
assign = {v: rng.randint(box[v][0], box[v][1]) for v in variables}
if not all(holds(a, assign) for a in base):
continue
if any(not holds(c, assign) for c in safety):
hits += 1
p_lo, p_hi = clopper_pearson(hits, n_samples, alpha)
return ApproxCount(
estimate=round(hits / n_samples * volume),
ci_low=math.floor(p_lo * volume),
ci_high=math.ceil(p_hi * volume),
n_samples=n_samples,
n_hits=hits,
alpha=alpha,
box_volume=volume,
seed=seed,
)
# --------------------------------------------------------------------------- #
# decide-only: the yes/no question, without paying for the count
#
# Most callers -- CI gates, agents -- ask "is this guard sound?", not "how
# unsound?". Counting the whole violating region to answer that is enormous
# waste: on a measured unsound guard the full count took 286 ms while the first
# witness surfaced in 0.04 ms.
#
# The trap is that `find_witness` returns None for two different reasons: no
# witness exists, or the search hit the cap. Collapsing those would report SOUND
# for a search that never ran -- exactly the class of confident-wrong answer this
# package exists to avoid. So the cap is checked *before* searching, and a capped
# query abstains rather than answering.
# --------------------------------------------------------------------------- #
@dataclass
class Decision:
"""A yes/no soundness verdict, or an explicit abstention."""
is_sound: bool | None
witness: dict[str, int] | None
enumerated_points: int
exact_cap: int
reason: str = ""
@property
def decided(self) -> bool:
return self.is_sound is not None
def __bool__(self) -> bool: # pragma: no cover - guard against silent coercion
raise TypeError(
"Decision has three states (sound, unsound, undecided); test "
".is_sound explicitly rather than relying on truthiness."
)
def to_dict(self) -> dict[str, Any]:
return asdict(self) | {"decided": self.decided}
def _repr_html_(self) -> str:
"""Notebook rendering. Undecided gets its own colour and its own
sentence -- the same reason ``__bool__`` raises."""
from .notebook import decision_html
return decision_html(self)
def decide_soundness(
domain: Sequence[Atom],
guard: Sequence[Atom],
safety: Sequence[Atom],
box: Mapping[str, tuple[int, int]],
*,
exact_cap: int = DEFAULT_EXACT_CAP,
allow_degenerate: bool = False,
) -> Decision:
"""Decide whether ``guard`` implies ``safety`` over ``box``, without counting.
Stops at the first escaping state. Returns ``is_sound=None`` -- never
``True`` -- when the enumeration would exceed ``exact_cap``, because a search
that did not run has not established anything.
Same verdict as :func:`over_acceptance`, reached without computing the
magnitude. Use ``over_acceptance`` when you need the number itself.
"""
if not safety:
raise ValueError("safety must have at least one conjunct")
base: list[Atom] = list(domain) + list(guard)
validate_box(
box,
base + [negate(c) for c in safety],
allow_degenerate=allow_degenerate,
)
enumerated, _ = enumeration_cost(box)
if enumerated > exact_cap:
return Decision(
is_sound=None,
witness=None,
enumerated_points=enumerated,
exact_cap=exact_cap,
reason=(
f"deciding this box would enumerate {enumerated:,} points, above the "
f"{exact_cap:,} limit. No search was run, so nothing is established."
),
)
for conjunct in safety:
witness = find_witness(base + [negate(conjunct)], box, exact_cap=exact_cap)
if witness is not None:
return Decision(
is_sound=False,
witness=witness,
enumerated_points=enumerated,
exact_cap=exact_cap,
reason="a state satisfying the guard violates the safety property",
)
return Decision(
is_sound=True,
witness=None,
enumerated_points=enumerated,
exact_cap=exact_cap,
reason="every enumerated state satisfying the guard also satisfies safety",
)
"""HTML rendering for notebooks, with the abstention intact.
A :class:`~exploit_counter.api.Decision` has three states, and the third one is
the whole point: ``is_sound is None`` means the question was not decided. In a
notebook that state is the easiest to lose -- an undecided result rendered in the
same grey as a passing one, or worse, rendered as a tick.
So each state gets its own colour and its own sentence, and the undecided one
says out loud that nothing was established. There is no boolean anywhere in the
output, because ``Decision.__bool__`` raises for exactly the same reason.
"""
from __future__ import annotations
from html import escape
from typing import Any
__all__ = ["decision_html", "over_acceptance_html", "COLOURS"]
COLOURS = {
"sound": ("#0f7b3f", "#eaf6ee"),
"unsound": ("#a41b1b", "#fbeaea"),
"undecided": ("#8a5a00", "#fff6e5"),
}
def _frame(state: str, headline: str, body: str) -> str:
fg, bg = COLOURS[state]
return (
f'<div style="border-left:4px solid {fg};background:{bg};padding:10px 14px;'
f'font-family:system-ui,sans-serif;font-size:13px;color:#111;max-width:46em">'
f'<div style="font-weight:700;color:{fg};font-family:ui-monospace,monospace">'
f"{escape(headline)}</div>{body}</div>"
)
def decision_html(decision: Any) -> str:
if decision.is_sound is None:
return _frame(
"undecided",
"UNDECIDED",
"<div>No verdict was reached: the enumeration needed "
f"{decision.enumerated_points:,} points against a cap of {decision.exact_cap:,}. "
"<b>This is not a clean result.</b> Nothing was established about the guard "
"either way.</div>"
+ (f"<div><small>{escape(decision.reason)}</small></div>" if decision.reason else ""),
)
if decision.is_sound:
return _frame(
"sound",
"SOUND over the declared box",
f"<div>No state in the declared box escapes the guard "
f"({decision.enumerated_points:,} points enumerated). This says nothing about "
"states outside the box.</div>",
)
witness = ", ".join(f"{k} = {v}" for k, v in sorted((decision.witness or {}).items()))
return _frame(
"unsound",
"UNSOUND",
f"<div>The guard admits a state the safety property forbids. "
f"Counterexample: <code>{escape(witness)}</code></div>",
)
def over_acceptance_html(result: Any) -> str:
"""Render an :class:`~exploit_counter.api.OverAcceptance`.
``is_sound_guard`` is ``None`` when only a sampled bracket exists, because
sampling can bound a count but never establish a zero. That case renders as
undecided, never as sound.
"""
sound = result.is_sound_guard
if sound is None:
return _frame(
"undecided",
"NOT DECIDED",
"<div>Only a sampled estimate is available. Sampling can bound a count; it "
"cannot establish that the count is zero. <b>Do not read this as sound.</b></div>",
)
if sound:
return _frame(
"sound",
"0 escaping states",
f"<div>Exact count over a domain of {result.domain_volume:,} states, "
f"{result.n_obligations} obligation(s), method <code>{escape(result.method)}</code>.</div>",
)
return _frame(
"unsound",
f"{result.exact:,} escaping states",
f"<div>Out of a domain of {result.domain_volume:,}. This is a triggerability figure "
"under uniform sampling, not a severity score.</div>",
)
"""``exploit-counter`` command-line entry point.
Two questions, two subcommands, because they are not the same question:
exploit-counter check --spec s.json --box p=0:255,r=0:255 # is it sound?
exploit-counter count --spec s.json --box p=0:255,r=0:255 # how unsound?
`check` stops at the first escaping state; `count` enumerates the whole violating
region. On an unsound guard that difference is large, and on a sound one there is
no difference at all -- the full enumeration is required either way.
Exit codes:
0 sound over the declared box (no state escapes)
1 a state escapes, or only a sampled bracket was obtainable
2 usage error
3 UNDECIDED -- the box is too large to enumerate, so no verdict was reached.
Not a pass. `count` never returns this: it falls back to sampling and says
so, which is a different kind of answer rather than an absent one.
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Any
from certkit import atom_from_json
from .api import decide_soundness, over_acceptance
from .exact import DEFAULT_EXACT_CAP
def example_path(name: str) -> Path:
"""Path to a bundled example file, wherever the package is installed."""
return Path(__file__).resolve().parent / "examples" / name
def _demo() -> int:
"""Count over-acceptance for a sound guard and a weakened one.
Needs no files on disk, so the documented quickstart works immediately after
`pip install exploit-counter`.
"""
from certkit import atom_from_json
box = {"payload": (0, 255), "record_len": (0, 255)}
print("exploit-counter demo -- CVE-2014-0160 (Heartbleed) shape")
print(" safety: 3 + payload <= record_len, payload in [0, 255]")
print()
for label, filename in (
("sound (19 + p <= r)", "sound.spec.json"),
("weakened (1 + p <= r)", "weakened.spec.json"),
):
spec = json.loads(example_path(filename).read_text(encoding="utf-8"))
result = over_acceptance(
[atom_from_json(a) for a in spec["domain"]],
[atom_from_json(a) for a in spec["guard"]],
[atom_from_json(a) for a in spec["safety"]],
box,
)
print(f" {label}")
print(f" over-acceptance: {result.exact} of {result.domain_volume} states")
draws = result.expected_draws_to_hit()
if draws is None:
print(" no state escapes; the guard is sound over this box")
else:
print(f" expected uniform draws to hit: {draws:,.0f}")
print()
print(" The weakened guard's gap is exactly counted, not estimated.")
return 0
def _parse_box(text: str) -> dict[str, tuple[int, int]]:
"""Parse ``p=0:65535,r=0:4294967295`` into a box dict."""
box: dict[str, tuple[int, int]] = {}
for part in text.split(","):
part = part.strip()
if not part:
continue
if "=" not in part or ":" not in part:
raise ValueError(f"bad box segment {part!r}; expected name=lo:hi")
name, rng = part.split("=", 1)
lo, hi = rng.split(":", 1)
box[name.strip()] = (int(lo), int(hi))
if not box:
raise ValueError("box is empty")
return box
def main(argv: Any = None) -> int:
parser = argparse.ArgumentParser(
prog="exploit-counter",
description="Count exactly how many states a guard wrongly admits.",
)
sub = parser.add_subparsers(dest="command", required=True)
c = sub.add_parser(
"check",
help="decide whether the guard is sound, without counting how unsound it is",
)
c.add_argument("--spec", required=True, type=Path)
c.add_argument("--box", required=True, help="e.g. p=0:255,r=0:255")
c.add_argument("--exact-cap", type=int, default=DEFAULT_EXACT_CAP)
c.add_argument("--json", action="store_true")
p = sub.add_parser("count", help="count over-acceptance of a spec's guard")
p.add_argument("--spec", required=True, type=Path)
p.add_argument("--box", required=True, help="e.g. p=0:65535,r=0:65535")
p.add_argument("--exact-cap", type=int, default=DEFAULT_EXACT_CAP)
p.add_argument("--samples", type=int, default=200_000)
p.add_argument("--alpha", type=float, default=0.05)
p.add_argument("--seed", type=int, default=0)
p.add_argument("--json", action="store_true")
sub.add_parser(
"demo",
help="run the bundled example (needs no files; works straight from pip install)",
)
args = parser.parse_args(argv)
if args.command == "demo":
return _demo()
try:
spec = json.loads(args.spec.read_text(encoding="utf-8"))
box = _parse_box(args.box)
except (OSError, json.JSONDecodeError, ValueError) as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
try:
domain = [atom_from_json(a) for a in spec.get("domain", [])]
guard = [atom_from_json(a) for a in spec.get("guard", [])]
safety = [atom_from_json(a) for a in spec.get("safety", [])]
except (KeyError, TypeError, ValueError) as exc:
print(f"error: malformed spec: {exc}", file=sys.stderr)
return 2
if not safety:
print("error: spec has no safety conjuncts", file=sys.stderr)
return 2
if args.command == "check":
decision = decide_soundness(domain, guard, safety, box, exact_cap=args.exact_cap)
name = spec.get("name", "<unnamed>")
if args.json:
print(json.dumps(decision.to_dict() | {"name": name}, indent=2))
elif decision.is_sound is None:
# The third state, and the reason this subcommand cannot simply
# return a boolean: nothing was established either way.
print(f"{name}: UNDECIDED -- {decision.reason}")
print(
f" {decision.enumerated_points:,} points would have to be enumerated "
f"against a cap of {decision.exact_cap:,}."
)
print(" This is not a pass. Narrow the box, or raise --exact-cap deliberately.")
elif decision.is_sound:
print(f"{name}: SOUND over the declared box")
print(f" {decision.enumerated_points:,} points enumerated; none escapes the guard.")
print(" This says nothing about states outside the box.")
else:
witness = ", ".join(f"{k} = {v}" for k, v in sorted((decision.witness or {}).items()))
print(f"{name}: UNSOUND")
print(f" the guard admits a forbidden state: {witness}")
print(" use `exploit-counter count` for the size of the violating region")
if decision.is_sound is None:
return 3
return 0 if decision.is_sound else 1
result = over_acceptance(
domain,
guard,
safety,
box,
exact_cap=args.exact_cap,
n_samples=args.samples,
alpha=args.alpha,
seed=args.seed,
)
if args.json:
print(json.dumps(result.to_dict(), indent=2))
else:
name = spec.get("name", "<unnamed>")
if result.is_exact:
n = result.exact
print(f"{name}: over-acceptance = {n} state(s) [exact, {result.method}]")
print(f" domain volume: {result.domain_volume}")
if n == 0:
print(" the guard admits no state the safety property forbids")
else:
draws = result.expected_draws_to_hit()
print(f" per-draw hit probability: {result.hit_probability():.6g}")
if draws is not None:
print(f" expected uniform draws to hit: {draws:,.0f}")
else:
a = result.approx or {}
print(f"{name}: over-acceptance bracketed [sampled, {result.method}]")
print(
f" {int((1 - args.alpha) * 100)}% interval: [{a.get('ci_low')}, {a.get('ci_high')}]"
)
print(
f" point estimate: {a.get('estimate')} ({a.get('n_hits')}/{a.get('n_samples')} hits)"
)
print(" note: sampling can bound a count; it cannot establish a zero")
return 0 if result.is_sound_guard else 1
if __name__ == "__main__":
raise SystemExit(main())
CORPUS = [{'certificate': {'obligations': [{'multipliers': {'3': [1, 1], '5': [1, 1]}}, {'multipliers': {'4': [1, 1], '5': [1, 1]}}], 'schema': 'certkit/farkas/v1', 'spec_fingerprint': '0d3ddf1ee131906ec857b96c51211665dcbb078de0bfe6225e6cd36cd9955ed7'}, 'common_name': 'curl NTLM type-3 overflow', 'cve': 'CVE-2019-3822', 'cwe': 'CWE-787', 'has_certificate': True, 'id': 'curl_ntlm_type3', 'n_domain_atoms': 3, 'n_guard_atoms': 2, 'n_safety_conjuncts': 2, 'project': 'curl', 'relation_in_words': 'the NTLM response length plus header must fit the output buffer', 'schema': 'cve-proof-corpus/v1', 'spec': {'domain': [{'coeff': {'cu_size': [-1, 1]}, 'const': [0, 1], 'strict': False}, {'coeff': {'cu_ntresplen': [-1, 1]}, 'const': [0, 1], 'strict': False}, {'coeff': {'cu_bufsize': [-1, 1]}, 'const': [0, 1], 'strict': False}], 'fingerprint': '0d3ddf1ee131906ec857b96c51211665dcbb078de0bfe6225e6cd36cd9955ed7', 'guard': [{'coeff': {'cu_ntresplen': [1, 1], 'cu_size': [1, 1]}, 'const': [-4294967295, 1], 'strict': False}, {'coeff': {'cu_bufsize': [-1, 1], 'cu_ntresplen': [1, 1], 'cu_size': [1, 1]}, 'const': [0, 1], 'strict': False}], 'name': 'curl_ntlm_type3', 'safety': [{'coeff': {'cu_ntresplen': [1, 1], 'cu_size': [1, 1]}, 'const': [-4294967295, 1], 'strict': False}, {'coeff': {'cu_bufsize': [-1, 1], 'cu_ntresplen': [1, 1], 'cu_size': [1, 1]}, 'const': [0, 1], 'strict': False}], 'schema': 'certkit/spec/v1'}, 'variables': ['cu_size', 'cu_ntresplen', 'cu_bufsize'], 'weakness': 'stack out-of-bounds write'}, {'certificate': {'obligations': [{'multipliers': {'1': [1, 1], '3': [1, 1]}}, {'multipliers': {'2': [1, 1], '3': [1, 1]}}], 'schema': 'certkit/farkas/v1', 'spec_fingerprint': '1506ef6cd6ee1a5377e4523e745599bdd848226e37b667e85c4a331f5443e94b'}, 'common_name': 'libpng time formatting overflow', 'cve': 'CVE-2015-7981', 'cwe': 'CWE-125', 'has_certificate': True, 'id': 'libpng_rfc1123', 'n_domain_atoms': 1, 'n_guard_atoms': 2, 'n_safety_conjuncts': 2, 'project': 'libpng', 'relation_in_words': 'the formatted output must fit the fixed destination buffer', 'schema': 'cve-proof-corpus/v1', 'spec': {'domain': [{'coeff': {'pn_month': [-1, 1]}, 'const': [0, 1], 'strict': False}], 'fingerprint': '1506ef6cd6ee1a5377e4523e745599bdd848226e37b667e85c4a331f5443e94b', 'guard': [{'coeff': {'pn_month': [-1, 1]}, 'const': [1, 1], 'strict': False}, {'coeff': {'pn_month': [1, 1]}, 'const': [-12, 1], 'strict': False}], 'name': 'libpng_rfc1123', 'safety': [{'coeff': {'pn_month': [-1, 1]}, 'const': [1, 1], 'strict': False}, {'coeff': {'pn_month': [1, 1]}, 'const': [-12, 1], 'strict': False}], 'schema': 'certkit/spec/v1'}, 'variables': ['pn_month'], 'weakness': 'out-of-bounds read'}, {'certificate': {'obligations': [{'multipliers': {'4': [1, 1], '5': [1, 1]}}], 'schema': 'certkit/farkas/v1', 'spec_fingerprint': '5be7d1c5c62ce8bc3f63a16fe4e587404feec56027a0f0d66b6561f637223998'}, 'common_name': 'libxml2 length overflow', 'cve': 'CVE-2015-8317', 'cwe': 'CWE-125', 'has_certificate': True, 'id': 'libxml2_len_overflow', 'n_domain_atoms': 4, 'n_guard_atoms': 1, 'n_safety_conjuncts': 1, 'project': 'libxml2', 'relation_in_words': 'a declared length must not exceed the remaining input', 'schema': 'cve-proof-corpus/v1', 'spec': {'domain': [{'coeff': {'xl_len': [-1, 1]}, 'const': [0, 1], 'strict': False}, {'coeff': {'xl_len': [1, 1]}, 'const': [-2147483647, 1], 'strict': False}, {'coeff': {'xl_incr': [-1, 1]}, 'const': [1, 1], 'strict': False}, {'coeff': {'xl_incr': [1, 1]}, 'const': [-4, 1], 'strict': False}], 'fingerprint': '5be7d1c5c62ce8bc3f63a16fe4e587404feec56027a0f0d66b6561f637223998', 'guard': [{'coeff': {'xl_incr': [1, 1], 'xl_len': [1, 1]}, 'const': [-2147483647, 1], 'strict': False}], 'name': 'libxml2_len_overflow', 'safety': [{'coeff': {'xl_incr': [1, 1], 'xl_len': [1, 1]}, 'const': [-2147483647, 1], 'strict': False}], 'schema': 'certkit/spec/v1'}, 'variables': ['xl_len', 'xl_incr'], 'weakness': 'out-of-bounds read'}, {'certificate': {'obligations': [{'multipliers': {'3': [1, 1], '4': [1, 1]}}], 'schema': 'certkit/farkas/v1', 'spec_fingerprint': 'eec4d1b54a17fd9baa5103f0f9b98394c7eb164f831d6cb1c413f9b5045bae13'}, 'common_name': 'Heartbleed', 'cve': 'CVE-2014-0160', 'cwe': 'CWE-125', 'has_certificate': True, 'id': 'openssl_heartbeat', 'n_domain_atoms': 3, 'n_guard_atoms': 1, 'n_safety_conjuncts': 1, 'project': 'OpenSSL', 'relation_in_words': 'the heartbeat payload length must fit inside the record it arrived in', 'schema': 'cve-proof-corpus/v1', 'spec': {'domain': [{'coeff': {'hb_payload': [-1, 1]}, 'const': [0, 1], 'strict': False}, {'coeff': {'hb_record_len': [-1, 1]}, 'const': [0, 1], 'strict': False}, {'coeff': {'hb_payload': [1, 1]}, 'const': [-65535, 1], 'strict': False}], 'fingerprint': 'eec4d1b54a17fd9baa5103f0f9b98394c7eb164f831d6cb1c413f9b5045bae13', 'guard': [{'coeff': {'hb_payload': [1, 1], 'hb_record_len': [-1, 1]}, 'const': [19, 1], 'strict': False}], 'name': 'openssl_heartbeat', 'safety': [{'coeff': {'hb_payload': [1, 1], 'hb_record_len': [-1, 1]}, 'const': [3, 1], 'strict': False}], 'schema': 'certkit/spec/v1'}, 'variables': ['hb_payload', 'hb_record_len'], 'weakness': 'out-of-bounds read'}, {'certificate': {'obligations': [{'multipliers': {'3': [1, 1], '4': [1, 1]}}], 'schema': 'certkit/farkas/v1', 'spec_fingerprint': '4f252ba298a05dab568f41f58cc1b102d34feeb57cbb9142a6329c723f4841e9'}, 'common_name': 'Baron Samedit', 'cve': 'CVE-2021-3156', 'cwe': 'CWE-787', 'has_certificate': True, 'id': 'sudo_set_cmnd', 'n_domain_atoms': 3, 'n_guard_atoms': 1, 'n_safety_conjuncts': 1, 'project': 'sudo', 'relation_in_words': 'the concatenated argument length must fit the allocated buffer', 'schema': 'cve-proof-corpus/v1', 'spec': {'domain': [{'coeff': {'su_idx': [-1, 1]}, 'const': [0, 1], 'strict': False}, {'coeff': {'su_size': [-1, 1]}, 'const': [0, 1], 'strict': False}, {'coeff': {'su_size': [-1, 1]}, 'const': [1, 1], 'strict': False}], 'fingerprint': '4f252ba298a05dab568f41f58cc1b102d34feeb57cbb9142a6329c723f4841e9', 'guard': [{'coeff': {'su_idx': [1, 1], 'su_size': [-1, 1]}, 'const': [0, 1], 'strict': True}], 'name': 'sudo_set_cmnd', 'safety': [{'coeff': {'su_idx': [1, 1], 'su_size': [-1, 1]}, 'const': [0, 1], 'strict': True}], 'schema': 'certkit/spec/v1'}, 'variables': ['su_idx', 'su_size'], 'weakness': 'heap out-of-bounds write'}, {'certificate': {'obligations': [{'multipliers': {'3': [1, 1], '5': [1, 1]}}, {'multipliers': {'4': [1, 1], '5': [1, 1]}}], 'schema': 'certkit/farkas/v1', 'spec_fingerprint': '1bd1394177f1e56fbd1bf7d81e3ae63f8213edeb355fa13919366d25c834301c'}, 'common_name': 'zlib inflate extra-field overflow', 'cve': 'CVE-2022-37434', 'cwe': 'CWE-787', 'has_certificate': True, 'id': 'zlib_inflate_extra', 'n_domain_atoms': 3, 'n_guard_atoms': 2, 'n_safety_conjuncts': 2, 'project': 'zlib', 'relation_in_words': 'the extra-field copy must fit the destination state buffer', 'schema': 'cve-proof-corpus/v1', 'spec': {'domain': [{'coeff': {'zl_off': [-1, 1]}, 'const': [0, 1], 'strict': False}, {'coeff': {'zl_copy': [-1, 1]}, 'const': [0, 1], 'strict': False}, {'coeff': {'zl_extra_max': [-1, 1]}, 'const': [0, 1], 'strict': False}], 'fingerprint': '1bd1394177f1e56fbd1bf7d81e3ae63f8213edeb355fa13919366d25c834301c', 'guard': [{'coeff': {'zl_copy': [1, 1], 'zl_off': [1, 1]}, 'const': [-4294967295, 1], 'strict': False}, {'coeff': {'zl_copy': [1, 1], 'zl_extra_max': [-1, 1], 'zl_off': [1, 1]}, 'const': [0, 1], 'strict': False}], 'name': 'zlib_inflate_extra', 'safety': [{'coeff': {'zl_copy': [1, 1], 'zl_off': [1, 1]}, 'const': [-4294967295, 1], 'strict': False}, {'coeff': {'zl_copy': [1, 1], 'zl_extra_max': [-1, 1], 'zl_off': [1, 1]}, 'const': [0, 1], 'strict': False}], 'schema': 'certkit/spec/v1'}, 'variables': ['zl_off', 'zl_copy', 'zl_extra_max'], 'weakness': 'out-of-bounds write'}]
"""Interactive certkit demo.
Three tabs, each answering a question a sceptic actually asks:
Certify a guard -- does this check imply the safety property, and if not,
which exact input escapes?
Check a certificate -- here is a proof; is it valid? (and watch a forgery fail)
CVE corpus -- six real vulnerability classes with real proofs, all
re-checked live in your browser session
Everything runs server-side in pure Python with exact rational arithmetic. There
is no solver, no network call, and no floating point anywhere on the verdict
path.
"""
from __future__ import annotations
import json
from pathlib import Path
import gradio as gr
from certkit import atom, check_certificate, make_spec
from exploit_counter import over_acceptance
CORPUS_PATH = None # replaced by the static build; see load_corpus()
# --------------------------------------------------------------------------- #
# tab 1 -- certify a guard
# --------------------------------------------------------------------------- #
PRESETS = {
"Heartbleed, correctly fixed (19 + payload <= record_len)": (19, 3, 255),
"Heartbleed, weakened guard (1 + payload <= record_len)": (1, 3, 255),
"Heartbleed, off by one (2 + payload <= record_len)": (2, 3, 255),
"Exactly tight (3 + payload <= record_len)": (3, 3, 255),
}
def certify(guard_overhead: int, safety_overhead: int, upper: int) -> str:
"""Decide whether `guard_overhead + p <= r` implies `safety_overhead + p <= r`."""
try:
guard_overhead = int(guard_overhead)
safety_overhead = int(safety_overhead)
upper = int(upper)
except (TypeError, ValueError):
return "### Input error\n\nAll three fields must be integers."
if upper < 1 or upper > 4095:
return (
"### Out of range\n\n"
"Keep the domain bound between 1 and 4095 so the demo answers instantly. "
"This is a limit of the *demo*, not of the method."
)
domain = [atom({"payload": -1}), atom({"payload": 1}, -upper)]
guard = [atom({"payload": 1, "record_len": -1}, guard_overhead)]
safety = [atom({"payload": 1, "record_len": -1}, safety_overhead)]
box = {"payload": (0, upper), "record_len": (0, upper)}
result = over_acceptance(domain, guard, safety, box)
volume = result.domain_volume
head = (
f"**Guard** `{guard_overhead} + payload <= record_len` \n"
f"**Safety** `{safety_overhead} + payload <= record_len` \n"
f"**Domain** `0 <= payload, record_len <= {upper}` ({volume:,} states)\n\n---\n\n"
)
if result.exact == 0:
return head + (
"## CERTIFIED\n\n"
f"The guard admits **no state** the safety property forbids, over all {volume:,} "
"points of the declared domain.\n\n"
"This is a proof over the domain declared above. It says nothing about states "
"outside it."
)
draws = result.expected_draws_to_hit()
p = result.hit_probability()
# Find one concrete escaping state to show.
from certkit import negate
from exploit_counter import find_witness
witness = find_witness(domain + guard + [negate(safety[0])], box)
wtxt = ""
if witness:
pv, rv = witness["payload"], witness["record_len"]
wtxt = (
f"\n\n**Counterexample:** `payload = {pv}, record_len = {rv}`\n\n"
f"At that input the guard passes (`{guard_overhead} + {pv} <= {rv}`) "
f"but the safety property does not hold "
f"(`{safety_overhead} + {pv} <= {rv}` is false).\n"
)
return head + (
"## PROVEN UNSOUND\n\n"
f"The guard admits **{result.exact:,} states** the safety property forbids, "
f"out of {volume:,}." + wtxt + f"\n**Per-draw hit probability:** {p:.3g} \n"
f"**Expected uniform draws to find one:** {draws:,.0f}\n\n"
"That last number is why testing may not have caught this."
)
def apply_preset(name: str):
g, s, u = PRESETS[name]
return g, s, u
# --------------------------------------------------------------------------- #
# tab 2 -- check a certificate
# --------------------------------------------------------------------------- #
def _demo_pair() -> tuple[dict, dict, dict]:
domain = [atom({"payload": -1}), atom({"payload": 1}, -65535)]
guard = [atom({"payload": 1, "record_len": -1}, 19)]
safety = [atom({"payload": 1, "record_len": -1}, 3)]
spec = make_spec(domain, guard, safety, name="heartbleed")
good = {
"schema": "certkit/farkas/v1",
"spec_fingerprint": spec["fingerprint"],
"obligations": [{"multipliers": {"2": 1, "3": 1}}],
}
forged = {
"schema": "certkit/farkas/v1",
"spec_fingerprint": spec["fingerprint"],
"obligations": [{"multipliers": {"0": 1, "1": 1}}],
}
return spec, good, forged
DEMO_SPEC, DEMO_GOOD, DEMO_FORGED = _demo_pair()
def check_pair(spec_text: str, cert_text: str) -> str:
try:
spec = json.loads(spec_text)
cert = json.loads(cert_text)
except json.JSONDecodeError as exc:
return f"### Malformed JSON\n\n```\n{exc}\n```"
try:
report = check_certificate(spec, cert)
except Exception as exc: # a checker must never crash on hostile input
return f"### Rejected\n\nThe checker refused this input: `{exc}`"
lines = [f"## {report.verdict}", ""]
if report.reason:
lines += [f"**Reason:** {report.reason}", ""]
for ob in report.obligations:
mark = "ok" if ob["ok"] else "**FAIL**"
detail = f" — {ob['reason']}" if ob["reason"] else ""
lines.append(f"- obligation {ob['index']}: {mark}{detail}")
if report.ok:
lines += [
"",
"---",
"",
"The multipliers combine the guard and the negated safety property so that every "
"variable cancels and an impossible constant remains. No counterexample exists.",
]
else:
lines += [
"",
"---",
"",
"The checker rebuilt the obligation **from the spec** and ignored any atoms the "
"certificate carried, so a certificate that proves some easier unrelated system "
"cannot pass here.",
]
return "\n".join(lines)
def load_good():
return json.dumps(DEMO_SPEC, indent=2), json.dumps(DEMO_GOOD, indent=2)
def load_forged():
return json.dumps(DEMO_SPEC, indent=2), json.dumps(DEMO_FORGED, indent=2)
# --------------------------------------------------------------------------- #
# tab 3 -- the CVE corpus
# --------------------------------------------------------------------------- #
def load_corpus() -> list[dict]:
from corpus_data import CORPUS as _BUNDLED
return list(_BUNDLED)
CORPUS = load_corpus()
def verify_corpus() -> str:
if not CORPUS:
return (
"### Corpus not bundled\n\n"
"`cve-proof-corpus.jsonl` is missing from this Space. "
"Copy it in from the dataset repository."
)
rows = ["| Class | CVE | Project | Obligations | Verdict |", "|---|---|---|---|---|"]
ok = 0
for rec in CORPUS:
report = check_certificate(rec["spec"], rec["certificate"])
if report.ok:
ok += 1
rows.append(
f"| `{rec['id']}` | {rec['cve']} | {rec['project']} | "
f"{rec['n_safety_conjuncts']} | {'**VERIFIED**' if report.ok else 'FAILED'} |"
)
verdict = f"\n\n**{ok} of {len(CORPUS)} re-checked successfully, just now, in this session.**"
note = (
"\n\nThese are the real upstream relations — `openssl_heartbeat` is the CVE-2014-0160 "
"length relation itself, not a paraphrase. The certificates were produced elsewhere; "
"what just ran is only the checker."
)
return "\n".join(rows) + verdict + note
def show_record(record_id: str) -> str:
rec = next((r for r in CORPUS if r["id"] == record_id), None)
if rec is None:
return "Select a class."
return (
f"### {rec['common_name']} — {rec['cve']}\n\n"
f"**Project:** {rec['project']} · **Weakness:** {rec['weakness']} ({rec['cwe']})\n\n"
f"**The relation:** {rec['relation_in_words']}\n\n"
f"**Variables:** `{', '.join(rec['variables'])}`\n\n"
"```json\n"
+ json.dumps({"guard": rec["spec"]["guard"], "safety": rec["spec"]["safety"]}, indent=2)
+ "\n```"
)
# --------------------------------------------------------------------------- #
# UI
# --------------------------------------------------------------------------- #
INTRO = """
# certkit — check the proof, not the promise
You should not have to trust the tool that *produced* a proof in order to believe the proof.
Everything below runs in pure Python with exact rational arithmetic. No solver, no floating point,
no network call. The checker is about 100 lines and you are meant to read it.
"""
with gr.Blocks(title="certkit — check the proof, not the promise") as demo:
gr.Markdown(INTRO)
with gr.Tab("Certify a guard"):
gr.Markdown(
"Does a bounds check actually imply the safety property? If not, **exactly how many "
"inputs escape**, and which one?"
)
preset = gr.Dropdown(choices=list(PRESETS), value=list(PRESETS)[1], label="Preset")
with gr.Row():
g_in = gr.Number(value=1, label="Guard overhead", precision=0)
s_in = gr.Number(value=3, label="Safety overhead (required)", precision=0)
u_in = gr.Number(value=255, label="Domain upper bound", precision=0)
btn = gr.Button("Certify", variant="primary")
out = gr.Markdown()
preset.change(apply_preset, preset, [g_in, s_in, u_in])
btn.click(certify, [g_in, s_in, u_in], out)
demo.load(certify, [g_in, s_in, u_in], out)
with gr.Tab("Check a certificate"):
gr.Markdown(
"Paste a spec and a certificate. Then press **Load a forgery** and watch it refuse — "
"the checker rebuilds the obligation from the spec, so a certificate carrying its own "
"easier system cannot pass."
)
with gr.Row():
load_ok = gr.Button("Load the valid pair")
load_bad = gr.Button("Load a forgery")
with gr.Row():
spec_box = gr.Code(value=json.dumps(DEMO_SPEC, indent=2), language="json", label="Spec")
cert_box = gr.Code(
value=json.dumps(DEMO_GOOD, indent=2), language="json", label="Certificate"
)
check_btn = gr.Button("Check", variant="primary")
check_out = gr.Markdown()
load_ok.click(load_good, None, [spec_box, cert_box])
load_bad.click(load_forged, None, [spec_box, cert_box])
check_btn.click(check_pair, [spec_box, cert_box], check_out)
demo.load(check_pair, [spec_box, cert_box], check_out)
with gr.Tab("CVE corpus"):
gr.Markdown(
"Six real vulnerability classes with real proofs. Press the button and every "
"certificate is re-checked live."
)
corpus_btn = gr.Button("Re-check all six now", variant="primary")
corpus_out = gr.Markdown()
corpus_btn.click(verify_corpus, None, corpus_out)
demo.load(verify_corpus, None, corpus_out)
if CORPUS:
gr.Markdown("### Inspect a class")
pick = gr.Dropdown(
choices=[r["id"] for r in CORPUS], value=CORPUS[0]["id"], label="Class"
)
detail = gr.Markdown()
pick.change(show_record, pick, detail)
demo.load(show_record, pick, detail)
gr.Markdown(
"""
---
**Scope, stated plainly.** This decides by exhaustive integer counting over the domain you declare.
That is sound and complete *for that domain* and silent outside it. The demo caps the domain so it
answers instantly — that is a limit of the demo, not the method. Deciding full 32-bit domains needs a
decision procedure that does not enumerate, which is not part of the open packages.
**Pre-release.** The PyPI names are reserved but not yet published, so these are the installs that
actually work today:
```
pip install "certkit@git+https://github.com/nickharris808/certkit@main"
pip install "exploit-counter@git+https://github.com/nickharris808/exploit-counter@main"
```
[certkit](https://github.com/nickharris808/certkit) ·
[exploit-counter](https://github.com/nickharris808/exploit-counter)
"""
)
if __name__ == "__main__":
demo.launch()