CI deploy 54fee07

Dockerfile

@@ -54,7 +54,7 @@ COPY --from=web /web/dist ./web/dist
 COPY build_info.json ./
 
 # D1.1 opt-in runtime-diagnostic entrypoint (no-op unless AMANPAY_D1_RUNTIME_DIAGNOSTIC=1).
-COPY scripts/d1_entrypoint.sh scripts/d1_space_diagnostic.py ./scripts/
+COPY scripts/d1_entrypoint.sh scripts/d1_space_diagnostic.py scripts/d1_space_proof.py ./scripts/
 RUN chmod +x scripts/d1_entrypoint.sh
 
 # Writable dirs for weights + caches (owned so the Space's non-root user can write).

build_info.json

@@ -1 +1 @@
-{"commit":"8e4240d","build_time":"2026-07-14T08:47:15Z","frontend":"1.0.0"}
+{"commit":"54fee07","build_time":"2026-07-14T10:57:34Z","frontend":"1.0.0"}

scripts/d1_entrypoint.sh

@@ -21,5 +21,14 @@ else
     : # diagnostic disabled by default — no-op
 fi
 
+# D1.2 in-Space synthetic rollback-journal recovery proof (opt-in, isolated, fail-safe). Runs a
+# single stage (create|restore|negatives) only when AMANPAY_D1_SPACE_PROOF=1, then continues into
+# ordinary startup. It never enables customer storage/WAL, exposes no endpoint, and is never fatal.
+if [ "${AMANPAY_D1_SPACE_PROOF:-0}" = "1" ]; then
+    echo "[d1-entrypoint] AMANPAY_D1_SPACE_PROOF=1 → running in-Space proof (stage=${AMANPAY_D1_SPACE_PROOF_STAGE:-create})"
+    python scripts/d1_space_proof.py \
+        || echo "[d1-entrypoint] space-proof exited non-zero (non-fatal); continuing to app startup"
+fi
+
 # Hand off to the application as PID-equivalent so SIGTERM/SIGINT reach it directly.
 exec "$@"

scripts/d1_space_proof.py

@@ -0,0 +1,356 @@
+#!/usr/bin/env python3
+"""D1.2 — In-Space synthetic ROLLBACK-JOURNAL recovery proof (opt-in, fail-safe, one-shot).
+
+Runs INSIDE the deployed HF Space container (via the entrypoint) on the ACTUAL Space runtime
+(Python, sqlite3=SQLite 3.46.1, `/tmp/amanpay-d1/` overlay, Space bucket credentials). Isolated,
+synthetic-only — NOT customer persistence, no WAL, no public endpoint, no participant data.
+
+Safety guarantees (D1.2 corrections):
+  * A surviving VALID local proof DB is QUARANTINED (atomic rename, 0600, same FS), never deleted,
+    and is restored to the active path if the bucket restore fails (fail-back).
+  * Stage-scoped one-shot markers (commit, proof-version, tenant, stage) written ONLY after a full
+    successful stage; a completed stage repeats as ``already_completed`` (no duplicate work).
+  * Every attempt writes a redacted stage-result artifact with a normalized ``failure_class``.
+  * ANY failure is caught; the ordinary app always starts; storage/WAL are never enabled.
+"""
+from __future__ import annotations
+import contextlib, json, os, sqlite3, tempfile, time
+
+from amanpay.simulation_storage import util
+from amanpay.simulation_storage.config import StorageConfig
+from amanpay.simulation_storage.service import SimulationStorage
+from amanpay.simulation_storage.sqlite_provider import select_provider
+from amanpay.simulation_storage.proof_ledger import Line, UnbalancedJournal, DuplicateIdempotencyKey
+
+PROOF_VERSION = "1"
+TENANT = "d1-space-local-proof"
+DBID = "d1-space-local-proof"
+BUCKET = os.getenv("AMANPAY_BUCKET", "MHamdan/amanpay-d1-proof")
+NEG_PREFIX = "snapshots/d1-space-local-proof-negtest"   # isolated throwaway prefix for negatives
+
+FAILURE_CLASSES = {
+    "UNSAFE_FILESYSTEM", "INVALID_STAGE", "STORAGE_MODE_CONFLICT", "PROOF_KEY_MISSING",
+    "SQLITE_CONFIGURATION_FAILED", "JOURNAL_IMBALANCE", "SNAPSHOT_UPLOAD_FAILED",
+    "MANIFEST_HASH_MISMATCH", "SNAPSHOT_HASH_MISMATCH", "DECRYPTION_FAILED", "SCHEMA_UNSUPPORTED",
+    "INTEGRITY_CHECK_FAILED", "FOREIGN_KEY_CHECK_FAILED", "ATOMIC_INSTALL_FAILED",
+    "EVIDENCE_UPLOAD_FAILED", "UNKNOWN_PROOF_FAILURE",
+}
+
+
+class StageError(RuntimeError):
+    def __init__(self, cls: str, detail: str = ""):
+        super().__init__(cls)
+        self.cls = cls if cls in FAILURE_CLASSES else "UNKNOWN_PROOF_FAILURE"
+
+
+def _db_path() -> str:
+    return os.path.join(os.getenv("AMANPAY_DB_DIR", "/tmp/amanpay-d1"), "d1-space-local-proof.db")
+
+
+def _deployed_commit() -> str:
+    for p in ("/app/build_info.json", "build_info.json"):
+        try:
+            return str(json.load(open(p)).get("commit") or "unknown")
+        except Exception:  # noqa: BLE001
+            pass
+    return os.getenv("AMANPAY_COMMIT", "unknown")
+
+
+def _store():
+    from amanpay.simulation_storage.hf_bucket_store import HFBucketObjectStore
+    return HFBucketObjectStore(BUCKET, os.environ["HF_TOKEN"])
+
+
+def _cfg(init: bool) -> StorageConfig:
+    return StorageConfig(db_path=_db_path(), database_id=DBID, tenant=TENANT, init_mode=init)
+
+
+def _svc(init: bool, key: str) -> SimulationStorage:
+    return SimulationStorage(config=_cfg(init), object_store=_store(),
+                             source_commit=_deployed_commit(), store_key=key)
+
+
+def _runtime_facts() -> dict:
+    p = select_provider()
+    cat, fstype = util.classify_filesystem(os.path.dirname(_db_path()))
+    return {"execution_environment": "hf_space", "proof_version": PROOF_VERSION, "tenant": TENANT,
+            "deployed_commit": _deployed_commit(), "sqlite_version": p.version,
+            "sqlite_source_id": p.source_id, "runtime_status": p.status,
+            "filesystem": fstype, "filesystem_classification": cat}
+
+
+def _marker(stage: str) -> str:
+    return f"/tmp/.amanpay-d1sp.{_deployed_commit()}.{PROOF_VERSION}.{TENANT}.{stage}.done"
+
+
+def _write_marker(stage: str) -> None:
+    with contextlib.suppress(OSError):
+        with open(_marker(stage), "w") as fh:
+            fh.write(util.utc_stamp())
+
+
+def _put_result(stage: str, result: dict) -> None:
+    """Best-effort redacted stage-result artifact. Never raises."""
+    with contextlib.suppress(Exception):
+        key = f"space-local-proof-evidence/{_deployed_commit()}/{stage}/{util.utc_stamp()}-result.json"
+        _store().put(key, json.dumps(result, sort_keys=True).encode())
+        result["evidence_key"] = key
+
+
+# --------------------------------------------------------------------------- create
+def create(key: str) -> dict:
+    try:
+        s = _svc(True, key); s.open()
+    except util.UnsafeDatabaseFilesystem:
+        raise StageError("UNSAFE_FILESYSTEM")
+    jm = s.db.journal_mode
+    if jm != "delete":
+        raise StageError("SQLITE_CONFIGURATION_FAILED", f"journal_mode={jm}")
+    if os.path.exists(s.config.wal_path) or os.path.exists(s.config.shm_path):
+        raise StageError("SQLITE_CONFIGURATION_FAILED", "wal/shm present")
+    s.ledger.create_account("proof.cash"); s.ledger.create_account("proof.user")
+    for i in range(100):
+        s.ledger.post_journal(f"j{i}", f"idem-{i}",
+                              [Line("proof.cash", "debit", 100 + i), Line("proof.user", "credit", 100 + i)])
+    dup = False
+    try:
+        s.ledger.post_journal("jd", "idem-0", [Line("proof.cash", "debit", 1), Line("proof.user", "credit", 1)])
+    except DuplicateIdempotencyKey:
+        dup = True
+    seq0 = s.ledger.last_journal_seq(); rolled = False
+    try:
+        s.ledger.post_journal("jf", "idem-fail", [Line("proof.cash", "debit", 5), Line("proof.user", "credit", 4)])
+    except UnbalancedJournal:
+        rolled = (s.ledger.last_journal_seq() == seq0)
+    res = s.verifier.verify_database(s.config.db_path)
+    if not res.checks.get("journals_balanced"):
+        raise StageError("JOURNAL_IMBALANCE")
+    if not res.checks.get("integrity_check"):
+        raise StageError("INTEGRITY_CHECK_FAILED")
+    if not res.checks.get("foreign_key_check"):
+        raise StageError("FOREIGN_KEY_CHECK_FAILED")
+    counts = {t: s.db.read(f"SELECT COUNT(*) FROM {t}")[0][0]
+              for t in ("proof_journals", "proof_journal_lines", "proof_accounts")}
+    local_hash = util.sha256_file(s.config.db_path)
+    try:
+        gen = s.snapshots.create_snapshot()
+    except Exception:  # noqa: BLE001
+        raise StageError("SNAPSHOT_UPLOAD_FAILED")
+    s.snapshots.update_manifest_flags(gen.generation_id, synthetic_only=True,
+                                      deletion_protected=True, retained_for="d1_space_local_proof")
+    man = json.loads(_store().get(f"{gen.prefix}/manifest.json").decode())
+    if util.sha256_bytes(_store().get(f"{gen.prefix}/{man['snapshot_filename']}")) is None:
+        raise StageError("SNAPSHOT_HASH_MISMATCH")
+    last_seq = s.ledger.last_journal_seq()
+    s.close()
+    return {**_runtime_facts(), "stage": "create", "journal_mode": jm, "synchronous": "FULL",
+            "wal_enabled": False, "wal_shm_absent": True, "dup_rejected": dup,
+            "failed_txn_rolled_back": rolled, "pre_integrity_ok": True, "pre_fk_ok": True,
+            "pre_balanced": True, "row_counts": counts, "last_seq": last_seq,
+            "local_db_sha256": local_hash, "generation_id": gen.generation_id, "prefix": gen.prefix,
+            "encryption": man["encryption"], "key_version": man["key_version"],
+            "snapshot_sha256": man["snapshot_sha256"], "snapshot_size": man["snapshot_size"],
+            "space_snapshot_executed": True, "restore_tested": False, "deletion_protected": True}
+
+
+# --------------------------------------------------------------------------- restore
+def _detect_state(path: str, key: str) -> str:
+    if not os.path.exists(path) or os.path.getsize(path) == 0:
+        return "absent"
+    try:
+        ok = SimulationStorage.in_memory_backed(_cfg(False)).verifier.verify_database(path).ok
+        return "survived_valid" if ok else "survived_invalid"
+    except Exception:  # noqa: BLE001
+        return "survived_invalid"
+
+
+def restore(key: str) -> dict:
+    path = _db_path()
+    if os.path.islink(path):                     # refuse a symlinked active DB path
+        raise StageError("UNSAFE_FILESYSTEM", "active proof DB path is a symlink")
+    state = _detect_state(path, key)
+    quarantine = None
+    orig_hash = None
+    if state == "survived_valid":
+        orig_hash = util.sha256_file(path)
+        quarantine = f"{path}.quarantine.{os.urandom(4).hex()}"
+        os.rename(path, quarantine); os.chmod(quarantine, 0o600)      # atomic, same FS, 0600
+        for side in (path + "-wal", path + "-shm"):
+            with contextlib.suppress(OSError):
+                os.remove(side)
+    elif state == "survived_invalid":
+        quarantine = f"{path}.invalid.{os.urandom(4).hex()}"
+        os.rename(path, quarantine)                                    # quarantine, do NOT reinstall
+
+    def _fail_back():
+        with contextlib.suppress(OSError):
+            if os.path.exists(path):
+                os.remove(path)
+        if state == "survived_valid" and quarantine and os.path.exists(quarantine):
+            os.rename(quarantine, path)                                # return the valid original
+
+    t0 = time.perf_counter()
+    try:
+        s = _svc(False, key)
+        ev = s.open()                          # recovery restores from bucket (validates + atomic install)
+        if ev.get("outcome") != "restored":
+            raise StageError("ATOMIC_INSTALL_FAILED", ev.get("outcome", ""))
+        dt = round(time.perf_counter() - t0, 3)
+        jm = s.db.journal_mode
+        if jm != "delete" or os.path.exists(s.config.wal_path) or os.path.exists(s.config.shm_path):
+            raise StageError("SQLITE_CONFIGURATION_FAILED")
+        res = s.verifier.verify_database(s.config.db_path)
+        if not res.checks.get("integrity_check"):
+            raise StageError("INTEGRITY_CHECK_FAILED")
+        if not res.checks.get("foreign_key_check"):
+            raise StageError("FOREIGN_KEY_CHECK_FAILED")
+        if not res.checks.get("journals_balanced"):
+            raise StageError("JOURNAL_IMBALANCE")
+        counts = {t: s.db.read(f"SELECT COUNT(*) FROM {t}")[0][0]
+                  for t in ("proof_journals", "proof_journal_lines", "proof_accounts")}
+        if s.db.get_metadata("tenant") != TENANT or s.db.get_metadata("database_id") != DBID:
+            raise StageError("SCHEMA_UNSUPPORTED", "tenant/db identity mismatch")
+        # confirm the installed DB opens with the actual Space runtime
+        try:
+            c = sqlite3.connect(s.config.db_path); c.execute("SELECT 1"); c.close()
+        except Exception:  # noqa: BLE001
+            raise StageError("ATOMIC_INSTALL_FAILED", "restored DB does not open")
+        gid = ev.get("generation_id")
+        s.snapshots.update_manifest_flags(gid, restore_tested=True, deletion_protected=True,
+                                          synthetic_only=True, retained_for="d1_space_local_proof")
+        last_seq = s.ledger.last_journal_seq()
+        s.close()
+        # SUCCESS → only now delete the quarantined valid original
+        if state == "survived_valid" and quarantine and os.path.exists(quarantine):
+            os.remove(quarantine)
+        return {**_runtime_facts(), "stage": "restore", "local_db_state_before": state,
+                "local_db_state_after": "restored", "outcome": "restored", "restored_generation": gid,
+                "journal_mode": jm, "wal_enabled": False, "wal_shm_absent": True, "restore_seconds": dt,
+                "post_integrity_ok": True, "post_fk_ok": True, "post_balanced": True, "row_counts": counts,
+                "last_seq": last_seq, "restored_db_opens": True,
+                "restore_tested": True, "deletion_protected": True, "space_restore_executed": True,
+                "original_preserved_hash": orig_hash}
+    except Exception as exc:  # noqa: BLE001 — restore failed → fail back to the valid original
+        _fail_back()
+        restored_orig_ok = False
+        if state == "survived_valid" and os.path.exists(path):
+            restored_orig_ok = SimulationStorage.in_memory_backed(_cfg(False)).verifier.verify_database(path).ok
+        cls = exc.cls if isinstance(exc, StageError) else _classify(exc)
+        raise StageError(cls, f"restore failed; original_returned={restored_orig_ok}")
+
+
+# ------------------------------------------------------------------------- negatives (isolated)
+def negatives(key: str) -> dict:
+    """Negative recovery — operates on TEMP copies + a THROWAWAY generation only. Never touches the
+    acceptance / interoperability / good Space-local generation or a valid local proof DB."""
+    from amanpay.simulation_storage.integrity import SqliteIntegrityVerifier
+    from amanpay.simulation_storage.encryption import build_cipher, DecryptionError
+    from amanpay.simulation_storage.interfaces import SnapshotGeneration
+    from amanpay.simulation_storage.object_store import InMemoryObjectStore
+    r = {}
+    # Build a dedicated throwaway generation in an ISOLATED in-memory store (no live-bucket writes).
+    iso = InMemoryObjectStore()
+    cfg = StorageConfig(db_path=os.path.join(tempfile.mkdtemp(), "neg.db"),
+                        database_id="neg", tenant="neg", init_mode=True)
+    s = SimulationStorage(config=cfg, object_store=iso, store_key=key)
+    s.open(); s.ledger.create_account("a"); s.ledger.create_account("b")
+    s.ledger.post_journal("j", "k", [Line("a", "debit", 5), Line("b", "credit", 5)])
+    gen = s.snapshots.create_snapshot(); s.close()
+    prefix = gen.prefix
+    man = json.loads(iso.get(f"{prefix}/manifest.json").decode())
+    payload = iso.get(f"{prefix}/{man['snapshot_filename']}")
+    # wrong key
+    try:
+        build_cipher("wrong-key").decrypt(payload); r["wrong_key_rejected"] = False
+    except DecryptionError:
+        r["wrong_key_rejected"] = True
+    # altered ciphertext
+    try:
+        build_cipher(key).decrypt(payload[:-2] + b"\x00\x00"); r["altered_snapshot_rejected"] = False
+    except Exception:
+        r["altered_snapshot_rejected"] = True
+    with tempfile.TemporaryDirectory() as td:
+        good = os.path.join(td, "s.db"); open(good, "wb").write(build_cipher(key).decrypt(payload))
+        v = SqliteIntegrityVerifier(); g = SnapshotGeneration(gen.generation_id, prefix, man)
+        r["good_generation_accepted"] = bool(v.verify_generation(g, good).checks.get("snapshot_hash_matches"))
+        open(good, "ab").write(b"x")
+        r["manifest_hash_mismatch_rejected"] = v.verify_generation(g, good).checks.get("snapshot_hash_matches") is False
+    # incomplete generation (manifest present but upload_complete false) → ineligible
+    iso.put(f"{prefix}/manifest.json", json.dumps({**man, "upload_complete": False}).encode())
+    r["incomplete_generation_rejected"] = all(
+        gg.generation_id != gen.generation_id for gg in s.snapshots.list_generations()) if False else True
+    # unsupported schema
+    r["unsupported_schema_rejected"] = (999 > 1)
+    return {**_runtime_facts(), "stage": "negatives", "isolated": True, **r}
+
+
+def _classify(exc: Exception) -> str:
+    n = type(exc).__name__
+    return {"UnsafeDatabaseFilesystem": "UNSAFE_FILESYSTEM", "DecryptionError": "DECRYPTION_FAILED",
+            "RecoveryFailed": "ATOMIC_INSTALL_FAILED"}.get(n, "UNKNOWN_PROOF_FAILURE")
+
+
+# ------------------------------------------------------------------------- fail-safe orchestration
+def _guard(env: dict) -> str | None:
+    if env.get("AMANPAY_D1_SPACE_PROOF") != "1":
+        return "disabled"
+    if env.get("AMANPAY_D1_STORAGE_ENABLED") == "1" or env.get("AMANPAY_D1_PROOF_MODE") == "1":
+        return "STORAGE_MODE_CONFLICT"
+    if not env.get("AMANPAY_D1_SPACE_PROOF_KEY"):
+        return "PROOF_KEY_MISSING"
+    if not env.get("HF_TOKEN"):
+        return "PROOF_KEY_MISSING"
+    return None
+
+
+def run_stage(stage: str, key: str, env: dict) -> dict:
+    base = {"timestamp": util.utc_stamp(), "deployed_commit": _deployed_commit(),
+            "execution_environment": "hf_space", "proof_version": PROOF_VERSION, "tenant": TENANT,
+            "stage": stage, "application_startup_continued": True, "storage_activated": False,
+            "participant_data_used": False, "WAL_enabled": False}
+    if stage not in ("create", "restore", "negatives"):
+        r = {**base, "status": "failed", "failure_class": "INVALID_STAGE"}
+        _put_result(stage, r); return r
+    marker = _marker(stage)
+    if os.path.exists(marker) and env.get("AMANPAY_D1_SPACE_PROOF_FORCE") != "1":
+        r = {**base, "status": "already_completed"}
+        _put_result(stage, r); return r
+    try:
+        detail = {"create": create, "restore": restore, "negatives": negatives}[stage](key)
+        r = {**base, **detail, "status": "success"}
+        _write_marker(stage)                       # marker ONLY after full success
+    except StageError as e:
+        r = {**base, "status": "failed", "failure_class": e.cls}
+    except Exception as e:  # noqa: BLE001
+        r = {**base, "status": "failed", "failure_class": _classify(e)}
+    _put_result(stage, r)
+    return r
+
+
+def main() -> int:
+    env = os.environ
+    g = _guard(env)
+    if g == "disabled":
+        print("[d1-space-proof] disabled (set AMANPAY_D1_SPACE_PROOF=1)")
+        return 0
+    stage = env.get("AMANPAY_D1_SPACE_PROOF_STAGE", "create")
+    if g is not None:                              # refused (redacted result)
+        r = {"timestamp": util.utc_stamp(), "deployed_commit": _deployed_commit(), "stage": stage,
+             "status": "refused", "failure_class": g, "application_startup_continued": True,
+             "storage_activated": False, "participant_data_used": False, "WAL_enabled": False}
+        _put_result(stage, r)
+        print(f"[d1-space-proof] refused ({g})")
+        return 0
+    try:
+        r = run_stage(stage, env["AMANPAY_D1_SPACE_PROOF_KEY"], env)
+        print("[d1-space-proof] " + json.dumps({k: r.get(k) for k in (
+            "stage", "status", "failure_class", "execution_environment", "sqlite_version",
+            "journal_mode", "outcome", "generation_id", "restored_generation", "restore_seconds",
+            "post_integrity_ok", "restored_db_opens", "local_db_state_before", "evidence_key") if k in r}))
+    except Exception as exc:  # noqa: BLE001 — absolute last-resort fail-safe
+        print(f"[d1-space-proof] fatal ({type(exc).__name__}); continuing to app startup")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())