Production API W1+W2: Ed25519 CA + durable store

services/api/NOTES.md

@@ -0,0 +1,94 @@
+# AxiomCode production build — W1 + W2
+
+Local working copy built from `api-prototype/` (state/secret files —
+keys.txt, meter.jsonl, revoked.txt — deliberately excluded). Nothing
+pushed anywhere; this tree is for review/checkpoint.
+
+## W1 — Ed25519 certification authority (`ca.py`, new)
+
+- `CA` class: Ed25519 keypair generation; `CA.load_or_create(path)`
+  persists the private key as PEM (PKCS8, no passphrase) with mode
+  0600 and reloads it afterwards. Key file path from
+  `AXIOMCODE_CA_KEY_FILE` (default: `ca_key.pem` beside the module).
+- Certificate format v2: the prototype fields — serial, issuer
+  "AxiomCode CA", valid_from / valid_until (90-day window),
+  code_sha256 (the artifact hash), verified (the verdict), toolchain,
+  attempts, key_id (the caller) — plus `version: 2`, `alg: "Ed25519"`,
+  `kid` (CA key id = first 16 hex of SHA-256 over the raw public key),
+  and `signature` (base64). Signature covers the canonical JSON of the
+  cert body — `json.dumps(..., sort_keys=True)`, the same
+  canonicalization certs.py used.
+- `verify_certificate(cert, public_key=...)` checks the signature with
+  a PUBLIC key only (accepts a key object, base64 raw key, or PEM —
+  i.e. exactly what the published payload carries) plus the validity
+  window; returns the prototype's `{valid, reason, payload}` shape.
+  Revocation is layered on top by the servers via the store, the same
+  split as X.509 vs CRL.
+- `public_keys_payload()` returns the document for
+  /.well-known/axiomcode-keys.json: issuer, keys[] with kid / alg /
+  use / public_key (base64) / public_key_pem, published_at.
+  verify_api serves it at `GET /.well-known/axiomcode-keys.json`.
+- verify_api.py and mcp_server.py issue + check via Ed25519 by
+  default. certs.py (HMAC) is unchanged and importable for legacy
+  fixtures; v2 verification rejects HMAC certs (no Ed25519 alg).
+
+## W2 — Durable store + hash-chained audit log (`store.py`, new)
+
+- sqlite3 (stdlib), one DB (`AXIOMCODE_DB`, default `axiomcode.db`
+  beside the module), WAL mode, lock-guarded single connection.
+- Tables: certificates (serial PK, cert_json, status, issued_at),
+  revocations (serial PK, reason, ts), meter_events (full billing
+  entry kept as JSON + indexed columns), audit_log (seq, ts,
+  event_type, payload_json, prev_hash, hash) with
+  hash = SHA-256(prev_hash + canonical payload_json); genesis
+  prev_hash is 64 zeros.
+- record_cert / revoke append their issuance / revocation audit
+  events by construction; the servers append "verification" events on
+  every /check and check_certificate call.
+- `verify_chain()` recomputes linkage + hashes and reports the first
+  broken seq.
+- billing.py: pricing math untouched (2.25× loaded-cost doctrine);
+  meter() / usage_for() now route through the store. A plain path
+  argument still uses the legacy JSONL code path for old fixtures.
+- revoked.txt and meter.jsonl are gone from both servers.
+- Temp Lean projects: created under `AXIOMCODE_WORK_ROOT` (default
+  system temp) and removed in a `finally` block in both servers.
+
+## Tests — 14 passed, 0 failed (`~/ax-venv/bin/python -m pytest tests/ -q`)
+
+- test_ca.py (6): key persistence + 0600 perms; issue→verify
+  round-trip (v2 fields, 90-day window); tampered verdict and mangled
+  signature fail; wrong key fails (bad_signature); published payload
+  verifies a real cert via both base64 and PEM; legacy HMAC cert
+  rejected.
+- test_store.py (5): record/get round-trip; revocation persists
+  across store close/reopen; audit chain verifies (issuance +
+  verification + revocation); one flipped byte in a stored payload is
+  detected with broken_seq = 2; meter events recorded and aggregated
+  (billing routes to the same numbers).
+- test_wiring.py (3): REST end-to-end with a stubbed prover —
+  /verify issues an Ed25519 cert, persists it, serves keys payload
+  that verifies it, /check valid → revoked after operator revocation,
+  /revoked lists it, temp dir cleaned, audit chain intact; MCP tools
+  (verify / check_certificate / usage_report) run on the same CA +
+  store; legacy HMAC module still works standalone.
+- The Lean prover is stubbed in wiring tests on purpose: they prove
+  CA/store wiring, not Lean. Real-Lean behaviour was proven by the
+  prototype demo (2026-09-26) and the engine suite.
+
+## Left stubbed / out of scope (later workstreams)
+
+- Revocation is operator-side only (store.revoke) — no public revoke
+  endpoint, same as the prototype; an authenticated admin path is
+  still to be designed.
+- API-key enforcement on REST (prototype had none; MCP bearer auth
+  unchanged, including its known mutable-identity issue — W3 scope).
+- Trial limits (10 cases), certificate withholding until payment,
+  Stripe entitlement/webhooks, quotas, rate limits, idempotency —
+  not in W1/W2 scope.
+- restart_rest.sh / restart_mcp.sh were not copied from the prototype
+  (they point at prototype paths/ports).
+- Ruff: clean except deliberate fail-closed broad `except` catches
+  (BLE001 in ca/certs/billing legacy path, S112 in the legacy JSONL
+  reader) and import-order notes where sys.path setup must precede
+  imports — all inherited patterns, left as-is intentionally.

services/api/billing.py

@@ -8,6 +8,11 @@
 
 1 credit = $0.01. Tune via AXIOMCODE_PRICING_JSON or the defaults below.
 Currently: 2.25x multiple over fully-loaded cost (~55.6% gross margin).
+
+Production (W2): meter records go through the durable store
+(store.record_meter — sqlite + hash-chained audit trail), not the
+prototype's meter.jsonl. meter()/usage_for() accept a Store; a plain
+file path still works via the legacy JSONL path for old fixtures.
 """
 
 import json
@@ -72,10 +77,13 @@ def load_keys(keys_file: str) -> dict:
     return keys
 
 
-def meter(
-    meter_log: str, key_id: str, code_sha: str, verified: bool, compute_s: float, pricing: dict | None = None
-) -> dict:
-    """Append one metered call to the billing/audit log. Returns the entry."""
+def _is_store(target) -> bool:
+    return hasattr(target, "record_meter") and hasattr(target, "usage_for")
+
+
+def meter(target, key_id: str, code_sha: str, verified: bool, compute_s: float, pricing: dict | None = None) -> dict:
+    """Record one metered call. `target` is the durable Store
+    (production) or a legacy JSONL path. Returns the entry."""
     bill = charge_for(compute_s, pricing)
     entry = {
         "ts": int(time.time()),
@@ -84,14 +92,22 @@ def meter(
         "verified": verified,
         **bill,
     }
-    with open(meter_log, "a") as f:
+    if _is_store(target):
+        entry_out: dict = target.record_meter(entry)
+        return entry_out
+    with open(target, "a") as f:
         f.write(json.dumps(entry) + "\n")
     return entry
 
 
-def usage_for(meter_log: str, key_id: str) -> dict:
+def usage_for(target, key_id: str) -> dict:
+    """Total metered spend for a key — from the Store (production)
+    or a legacy JSONL path."""
+    if _is_store(target):
+        usage: dict = target.usage_for(key_id)
+        return usage
     total_c, total_usd, calls = 0.0, 0.0, 0
-    p = Path(meter_log)
+    p = Path(target)
     if p.exists():
         for line in p.read_text().splitlines():
             try:

services/api/ca.py

@@ -0,0 +1,201 @@
+"""AxiomCode CA — Ed25519 certification authority (production, W1).
+
+Replaces the prototype's HMAC-SHA256 signing (certs.py, kept for legacy
+fixtures) with asymmetric Ed25519 signatures: certificates can be
+verified by ANYONE holding only the public key, which is published at
+/.well-known/axiomcode-keys.json. The private key never leaves the CA
+key file (PEM, mode 0600).
+
+Certificate format v2: the prototype fields (serial, issuer,
+valid_from / valid_until — a 90-day window, code_sha256 [artifact hash],
+verified [verdict], toolchain, attempts, key_id [caller]) plus:
+  "version": 2, "alg": "Ed25519", "kid": <CA key id>,
+  "signature": base64 Ed25519 signature over the canonical JSON of the
+  certificate body (every field except "signature", json.dumps with
+  sort_keys=True — the same canonicalization as certs.py).
+"""
+
+import base64
+import hashlib
+import json
+import os
+import time
+import uuid
+from pathlib import Path
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+    Ed25519PrivateKey,
+    Ed25519PublicKey,
+)
+
+_HERE = Path(__file__).parent
+ISSUER = "AxiomCode CA"
+ALG = "Ed25519"
+VALIDITY_DAYS = int(os.environ.get("AXIOMCODE_CERT_VALIDITY_DAYS", "90"))
+DEFAULT_KEY_FILE = os.environ.get("AXIOMCODE_CA_KEY_FILE", str(_HERE / "ca_key.pem"))
+
+
+def canonical(body: dict) -> bytes:
+    """Canonical bytes of a certificate body — same approach as certs.py."""
+    return json.dumps(body, sort_keys=True).encode()
+
+
+def kid_for(public_key: Ed25519PublicKey) -> str:
+    """Stable key id: first 16 hex of SHA-256 over the raw public key."""
+    raw = public_key.public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+    return hashlib.sha256(raw).hexdigest()[:16]
+
+
+def _pem_private(key: Ed25519PrivateKey) -> bytes:
+    return key.private_bytes(
+        serialization.Encoding.PEM,
+        serialization.PrivateFormat.PKCS8,
+        serialization.NoEncryption(),
+    )
+
+
+def _pem_public(key: Ed25519PublicKey) -> str:
+    return key.public_bytes(
+        serialization.Encoding.PEM,
+        serialization.PublicFormat.SubjectPublicKeyInfo,
+    ).decode()
+
+
+def coerce_public_key(key) -> Ed25519PublicKey:
+    """Accept an Ed25519PublicKey, a base64 raw-key string, raw bytes,
+    or a PEM string — so a key taken straight out of the published
+    /.well-known/axiomcode-keys.json payload can verify certificates."""
+    if isinstance(key, Ed25519PublicKey):
+        return key
+    if isinstance(key, str) and "BEGIN PUBLIC KEY" in key:
+        loaded = serialization.load_pem_public_key(key.encode())
+        if not isinstance(loaded, Ed25519PublicKey):
+            raise ValueError("PEM does not hold an Ed25519 public key")
+        return loaded
+    raw = base64.b64decode(key) if isinstance(key, str) else bytes(key)
+    return Ed25519PublicKey.from_public_bytes(raw)
+
+
+class CA:
+    """An Ed25519 certification authority bound to one keypair."""
+
+    def __init__(self, private_key: Ed25519PrivateKey):
+        self._private = private_key
+        self.public_key = private_key.public_key()
+        self.kid = kid_for(self.public_key)
+
+    # ─── key management ──────────────────────────────────────────────
+    @classmethod
+    def generate(cls) -> "CA":
+        return cls(Ed25519PrivateKey.generate())
+
+    @classmethod
+    def load_or_create(cls, key_file: str | Path | None = None) -> "CA":
+        """Load the CA key from `key_file`, generating and persisting a
+        new one (PEM, mode 0600) if it does not exist yet."""
+        path = Path(key_file or DEFAULT_KEY_FILE)
+        if path.exists():
+            key = serialization.load_pem_private_key(path.read_bytes(), password=None)
+            if not isinstance(key, Ed25519PrivateKey):
+                raise ValueError(f"{path} does not hold an Ed25519 key")
+            return cls(key)
+        ca = cls.generate()
+        path.parent.mkdir(parents=True, exist_ok=True)
+        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        try:
+            os.write(fd, _pem_private(ca._private))
+        finally:
+            os.close(fd)
+        os.chmod(path, 0o600)
+        return ca
+
+    # ─── issuance ────────────────────────────────────────────────────
+    def sign_body(self, body: dict) -> str:
+        sig = self._private.sign(canonical(body))
+        return base64.b64encode(sig).decode()
+
+    def issue(self, subject: dict) -> dict:
+        """Issue a v2 certificate. `subject` holds the verdict fields
+        (code_sha256, verified, attempts, toolchain, key_id, ...)."""
+        now = int(time.time())
+        cert = {
+            "version": 2,
+            "alg": ALG,
+            "kid": self.kid,
+            "serial": uuid.uuid4().hex[:16],
+            "issuer": ISSUER,
+            "valid_from": now,
+            "valid_until": now + VALIDITY_DAYS * 86400,
+            **subject,
+        }
+        cert["signature"] = self.sign_body(cert)
+        return cert
+
+    # ─── publication ─────────────────────────────────────────────────
+    def public_keys_payload(self) -> dict:
+        """The document served at /.well-known/axiomcode-keys.json."""
+        raw = self.public_key.public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+        return {
+            "issuer": ISSUER,
+            "keys": [
+                {
+                    "kid": self.kid,
+                    "alg": ALG,
+                    "use": "cert-signing",
+                    "public_key": base64.b64encode(raw).decode(),
+                    "public_key_pem": _pem_public(self.public_key),
+                }
+            ],
+            "published_at": int(time.time()),
+        }
+
+
+def verify_certificate(cert: dict, public_key=None, ca: CA | None = None) -> dict:
+    """Validate a v2 certificate's signature and validity window using
+    ONLY a public key (the CA's own by default, or one supplied — e.g.
+    taken from the published keys payload). Revocation is layered on
+    top by the caller via store.is_revoked — a public key alone cannot
+    know the revocation list, same split as X.509 vs CRL.
+
+    Returns the certs.check shape: {valid, reason?, payload?}."""
+    if not isinstance(cert, dict) or cert.get("alg") != ALG:
+        return {"valid": False, "reason": "bad_signature"}
+    sig_b64 = cert.get("signature")
+    if not sig_b64:
+        return {"valid": False, "reason": "bad_signature"}
+    pub = coerce_public_key(public_key if public_key is not None else (ca or default_ca()).public_key)
+    body = {k: v for k, v in cert.items() if k != "signature"}
+    try:
+        pub.verify(base64.b64decode(sig_b64), canonical(body))
+    except Exception:  # InvalidSignature, malformed base64, wrong key type
+        return {"valid": False, "reason": "bad_signature"}
+    now = int(time.time())
+    if now < cert.get("valid_from", 0):
+        return {"valid": False, "reason": "not_yet_valid"}
+    if now > cert.get("valid_until", 0):
+        return {"valid": False, "reason": "expired", "payload": cert, "renew": True}
+    return {"valid": True, "payload": cert}
+
+
+_DEFAULT_CA: CA | None = None
+
+
+def default_ca() -> CA:
+    """Process-wide CA, loaded/created from AXIOMCODE_CA_KEY_FILE."""
+    global _DEFAULT_CA
+    if _DEFAULT_CA is None:
+        _DEFAULT_CA = CA.load_or_create()
+    return _DEFAULT_CA
+
+
+def encode(cert: dict) -> str:
+    return json.dumps(cert, sort_keys=True)
+
+
+def decode(raw: str) -> dict | None:
+    try:
+        c = json.loads(raw)
+        return c if isinstance(c, dict) else None
+    except Exception:
+        return None