#!/usr/bin/env python3 """D2 — In-Space synthetic identity/lifecycle + recovery proof (opt-in, fail-safe, one-shot). Runs INSIDE the deployed HF Space (via the entrypoint) on the ACTUAL Space runtime, BEFORE uvicorn starts. It exercises the merged D2 identity code end-to-end on the real filesystem and SQLite runtime, using a real in-process ES256 WebAuthn channel (the "equivalent cryptographically verified WebAuthn test channel" the D2 runbook permits), and uploads REDACTED evidence to the private D2 proof bucket so it can be verified out-of-band. Strictly synthetic — NO participant data, NO real names/emails/phones/biometrics. It never enables WAL, never modifies Payment Core / PDP / ASAP, and is NEVER fatal (a failure is caught and the app still starts). Stages (AMANPAY_D2_SPACE_PROOF_STAGE): * ``full`` — verify runtime facts (§5), bootstrap tenant+operator+customer, run the lifecycle (§6-§9,§12), create an encrypted synthetic snapshot, emit evidence. * ``verify`` — after a factory rebuild: prove restore-on-start recovered the synthetic DB (§10), emit post-recovery evidence, mark the generation restore_tested. * ``negatives`` — isolated negative-recovery checks (§11) on throwaway copies. Evidence NEVER contains invitation codes, session/CSRF tokens, credential public-key bytes, challenge values, private keys, or PII — only states, counts, hashes and generation ids. """ from __future__ import annotations import contextlib import hashlib import json import os import secrets import sqlite3 import tempfile import time # --- deployed D2 identity code (all under amanpay/, shipped in the image) --- from amanpay.identity.config import D2Config, store_key from amanpay.identity.operator import OperatorService from amanpay.identity.service import IdentityService from amanpay.identity.storage import D2Storage from amanpay.simulation_storage import util PROOF_VERSION = "1" TENANT_SLUG = "d2-space-proof" BUCKET = os.getenv("AMANPAY_D2_BUCKET") or os.getenv("AMANPAY_BUCKET") or "MHamdan/amanpay-d2-proof" RP_ID = os.getenv("AMANPAY_D2_RP_ID", "mhamdan-amanpay.hf.space") ORIGIN = os.getenv("AMANPAY_D2_ORIGIN", "https://mhamdan-amanpay.hf.space") # ======================================================================= authenticator class _SoftAuthenticator: """Minimal in-process ES256 (ECDSA P-256) WebAuthn authenticator — real signatures. Verification is performed by the deployed py_webauthn via the D2 WebAuthn service; nothing is weakened. ``backup`` toggles BE/BS so synced vs single-device counter handling is exercised. """ def __init__(self, backup: bool = True): import cbor2 # noqa: F401 (availability check) self.backup = backup self._creds: dict = {} # -- helpers -- @staticmethod def _b64u(data: bytes) -> str: import base64 return base64.urlsafe_b64encode(data).decode().rstrip("=") def _flags(self, *, at: bool) -> int: f = 0x01 | 0x04 # UP | UV if self.backup: f |= 0x08 | 0x10 # BE | BS if at: f |= 0x40 # AT return f def _auth_data(self, rp_id, *, sign_count, cred_id, priv, include_attested): import cbor2 out = hashlib.sha256(rp_id.encode()).digest() + bytes([self._flags(at=include_attested)]) out += sign_count.to_bytes(4, "big") if include_attested: n = priv.public_key().public_numbers() cose = cbor2.dumps({1: 2, 3: -7, -1: 1, -2: n.x.to_bytes(32, "big"), -3: n.y.to_bytes(32, "big")}) out += b"\x00" * 16 + len(cred_id).to_bytes(2, "big") + cred_id + cose return out def create(self, options_json: str, origin: str | None = None) -> dict: import cbor2 from cryptography.hazmat.primitives.asymmetric import ec opts = json.loads(options_json) opts = opts.get("publicKey", opts) cred_id = secrets.token_bytes(20) priv = ec.generate_private_key(ec.SECP256R1()) self._creds[self._b64u(cred_id)] = {"priv": priv, "sc": 0, "uh": opts["user"]["id"]} cdj = json.dumps({"type": "webauthn.create", "challenge": opts["challenge"], "origin": origin or ORIGIN, "crossOrigin": False}, separators=(",", ":")).encode() ad = self._auth_data(opts["rp"]["id"], sign_count=0, cred_id=cred_id, priv=priv, include_attested=True) att = cbor2.dumps({"fmt": "none", "attStmt": {}, "authData": ad}) return {"id": self._b64u(cred_id), "rawId": self._b64u(cred_id), "type": "public-key", "response": {"clientDataJSON": self._b64u(cdj), "attestationObject": self._b64u(att), "transports": ["internal", "hybrid"]}, "clientExtensionResults": {}} def get(self, options_json: str, credential_id: str | None = None, origin: str | None = None) -> dict: from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import ec opts = json.loads(options_json) opts = opts.get("publicKey", opts) cid = credential_id or next(iter(self._creds)) st = self._creds[cid] if not self.backup: st["sc"] += 1 cdj = json.dumps({"type": "webauthn.get", "challenge": opts["challenge"], "origin": origin or ORIGIN, "crossOrigin": False}, separators=(",", ":")).encode() ad = self._auth_data(opts["rpId"], sign_count=st["sc"], cred_id=None, priv=None, include_attested=False) sig = st["priv"].sign(ad + hashlib.sha256(cdj).digest(), ec.ECDSA(hashes.SHA256())) return {"id": cid, "rawId": cid, "type": "public-key", "response": {"clientDataJSON": self._b64u(cdj), "authenticatorData": self._b64u(ad), "signature": self._b64u(sig), "userHandle": st["uh"]}, "clientExtensionResults": {}} # ============================================================================ infra def _deployed_commit() -> str: for p in ("/app/build_info.json", "build_info.json"): with contextlib.suppress(Exception): return str(json.load(open(p)).get("commit") or "unknown") return os.getenv("AMANPAY_COMMIT", "unknown") def _bucket(): from amanpay.simulation_storage.hf_bucket_store import HFBucketObjectStore return HFBucketObjectStore(BUCKET, os.environ["HF_TOKEN"]) def _put_evidence(stage: str, result: dict) -> str | None: """Upload a REDACTED stage-result artifact to the private bucket. Returns its key or None.""" try: result = _redact(result) result.update(proof_version=PROOF_VERSION, deployed_commit=_deployed_commit(), stage=stage, created_at_iso=util.utc_stamp()) key = f"d2-space-proof-evidence/{_deployed_commit()}/{stage}/{util.utc_stamp()}-result.json" _bucket().put(key, json.dumps(result, sort_keys=True, default=str).encode()) return key except Exception as exc: # noqa: BLE001 — best effort print(f"[d2-space-proof] evidence upload failed (non-fatal): {type(exc).__name__}") return None _FORBIDDEN = ("code", "token", "csrf", "secret", "public_key", "credential_id", "challenge", "pepper", "store_key", "password", "email", "phone") def _redact(obj): """Defense-in-depth: drop any key that could carry a secret / credential / PII.""" if isinstance(obj, dict): return {k: _redact(v) for k, v in obj.items() if not any(f in k.lower() for f in _FORBIDDEN)} if isinstance(obj, list): return [_redact(v) for v in obj] return obj def _runtime_facts(storage: D2Storage) -> dict: rr = storage.db.runtime_report() db_path = storage.config.db_path return { "execution_env": "hf_space" if os.getenv("SPACE_ID") or os.path.isdir("/app") else "other", "space_id": os.getenv("SPACE_ID", ""), "db_path_under_tmp_amanpay_d2": db_path.startswith("/tmp/amanpay-d2"), "fs_category": rr.get("fs_category"), "fstype": rr.get("fstype"), "sqlite_version": rr.get("sqlite_version"), "sqlite_source_id": rr.get("sqlite_source_id"), "journal_mode": rr.get("journal_mode"), "synchronous": rr.get("synchronous"), "wal_status": rr.get("wal_status"), # foreign_keys is enforced ON by the store's connections (D1 pragma discipline); the # per-connection default OFF is why the audit connection sets it explicitly above. "foreign_keys_enforced": bool(storage.db.config.foreign_keys), "busy_timeout_ms": storage.config.busy_timeout_ms, "schema_version": storage.db.schema_version(), "no_wal_file": not os.path.exists(storage.config.wal_path), "no_shm_file": not os.path.exists(storage.config.shm_path), } def _integrity(db_path: str) -> dict: conn = sqlite3.connect(db_path) try: conn.execute("PRAGMA foreign_keys=ON") # per-connection; make the check meaningful ic = conn.execute("PRAGMA integrity_check").fetchone()[0] fk = conn.execute("PRAGMA foreign_key_check").fetchall() return {"integrity_check": str(ic).lower(), "foreign_key_violations": len(fk)} finally: conn.close() def _counts(storage: D2Storage) -> dict: from amanpay.identity.storage.schema import CORE_TABLES out = {} for t in CORE_TABLES: with contextlib.suppress(Exception): out[t] = storage.db.read(f"SELECT COUNT(*) FROM {t}")[0][0] out["last_event_seq"] = storage.db.read( "SELECT COALESCE(MAX(seq),0) FROM lifecycle_events")[0][0] return out # ======================================================================= stage: full def _stage_full() -> dict: cfg = D2Config() if not util.is_confirmed_local(cfg.db_path) and not cfg.dev_allow_unknown_fs: # /tmp on the Space is overlay (local); this only guards a misconfiguration. pass key = store_key() if not key: raise RuntimeError("AMANPAY_D2_STORE_KEY missing — refusing unencrypted synthetic proof") storage = D2Storage(config=cfg, object_store=_bucket(), source_commit=_deployed_commit(), store_key=key) # Idempotent synthetic run: always start from a CLEAN local DB (this is throwaway proof data). # Opening with recover=True would restore prior synthetic accounts from the bucket and then a # fresh bootstrap would collide (handle_taken), so wipe the local DB and open without recovery. for p in (cfg.db_path, cfg.wal_path, cfg.shm_path): with contextlib.suppress(OSError): os.remove(p) ev: dict = {"outcome": "", "init_recovery": storage.open(recover=False)} try: facts = _runtime_facts(storage) facts.update(_integrity(cfg.db_path)) ev["runtime_facts"] = facts if facts["journal_mode"] != "delete": raise RuntimeError(f"journal_mode={facts['journal_mode']} (must be delete)") ident = IdentityService.from_storage(storage) op = OperatorService(ident) repo = ident.repo now = time.time() tenant = repo.get_tenant_by_slug(TENANT_SLUG) or repo.create_tenant( slug=TENANT_SLUG, name_en="AmanPay Synthetic Event", name_ar="فعالية أمان باي التجريبية", now=now) # --- §6 operator bootstrap (in-process via the same service the CLI uses) --- op_auth = _SoftAuthenticator(backup=True) _inv, op_code = ident.invites.create(tenant_id=tenant.id, role="operator", now=now) oopts = ident.begin_enrollment(code=op_code, public_handle="d2proofoperator", display_alias="Synthetic Operator", preferred_language="en", consent_accepted=True, now=now) ores = ident.complete_enrollment(credential=op_auth.create(oopts), now=now + 1) operator, osess = ores.user, ores.issued.session ev["operator"] = {"status": operator.status, "role": operator.role, "passkeys": repo.count_active_credentials(operator.id)} # --- §7 customer enrollment (operator issues the invitation) --- pairs = op.create_invitations(operator, role="customer", count=1, now=now + 2) cust_auth = _SoftAuthenticator(backup=True) # validate the invitation is atomic/one-time by re-reading state copts = ident.begin_enrollment(code=pairs[0][1], public_handle="d2proofuser", display_alias="Synthetic Participant", preferred_language="ar", consent_accepted=True, now=now + 3) cres = ident.complete_enrollment(credential=cust_auth.create(copts), now=now + 4) customer = cres.user ev["customer_enrolled"] = {"status": customer.status, "passkeys": repo.count_active_credentials(customer.id), "no_active_without_passkey": repo.count_active_credentials(customer.id) >= 1} # invitation cannot be reused try: ident.begin_enrollment(code=pairs[0][1], public_handle="reuse", display_alias="x", preferred_language="en", consent_accepted=True, now=now + 5) ev["invitation_reuse_blocked"] = False except Exception: # noqa: BLE001 ev["invitation_reuse_blocked"] = True # --- §8 lifecycle --- ev["lifecycle"] = _run_lifecycle(ident, op, operator, osess, cust_auth, customer, now) # --- §9 encrypted snapshot (synthetic-only, protected) --- gen = storage.snapshots.create_snapshot() storage.snapshots.update_manifest_flags( gen.generation_id, synthetic_only=True, deletion_protected=True, retained_for="d2_synthetic_acceptance") # verify by re-download + hash with tempfile.TemporaryDirectory() as td: dest = os.path.join(td, "s.db") storage.snapshots.download_snapshot( type(gen)(gen.generation_id, gen.prefix, storage.snapshots.manifests.read_manifest(gen.prefix)), dest) redl_sha = util.sha256_file(dest) ev["snapshot"] = {"generation_id": gen.generation_id, "encryption": gen.manifest.get("encryption"), "key_version": gen.manifest.get("key_version"), "schema_version": gen.manifest.get("schema_version"), "sqlite_version": gen.manifest.get("sqlite_version"), "upload_complete": gen.manifest.get("upload_complete"), "synthetic_only": True, "deletion_protected": True, "snapshot_sha_matches_after_redownload": redl_sha == gen.manifest.get("snapshot_sha256"), "row_counts": gen.manifest.get("row_counts")} ev["counts_after_full"] = _counts(storage) ev["outcome"] = "success" return ev finally: storage.close() def _run_lifecycle(ident, op, operator, osess, cust_auth, customer, now) -> dict: r: dict = {} # login (discoverable) after logout lg = _login(ident, cust_auth, now + 10) r["login_discoverable"] = lg.user.id == customer.id # wrong-origin rejection: a genuine assertion whose clientData origin is NOT the RP origin. o = ident.begin_login(now=now + 11) wrong = cust_auth.get(o, origin="https://evil.example.com") r["wrong_origin_rejected"] = _expect_fail( lambda: ident.complete_login(credential=wrong, now=now + 12)) # expired-ceremony rejection (ceremony/challenge TTL is 300s). NB: the result key avoids the # substring "challenge" so the evidence redactor does not elide it. o2 = ident.begin_login(now=now + 13) exp = cust_auth.get(o2) r["stale_ceremony_rejected"] = _expect_fail( lambda: ident.complete_login(credential=exp, now=now + 13 + 100_000)) # replay rejection (same assertion twice) o3 = ident.begin_login(now=now + 14) a = cust_auth.get(o3) ident.complete_login(credential=a, now=now + 15) r["replay_rejected"] = _expect_fail( lambda: ident.complete_login(credential=a, now=now + 16)) # multi-passkey: add a second (needs recent auth → fresh login session) lg2 = _login(ident, cust_auth, now + 20) sess2 = lg2.issued.session auth2 = _SoftAuthenticator(backup=False) addopts = ident.begin_add_passkey(lg2.user, sess2, now=now + 21) added = ident.complete_add_passkey(lg2.user, sess2, credential=auth2.create(addopts), nickname="Synthetic laptop", now=now + 22) r["passkeys_after_add"] = ident.repo.count_active_credentials(customer.id) # revoke the added (second) credential, authenticate with the remaining ident.revoke_passkey(lg2.user, sess2, added.id, now=now + 23) r["passkeys_after_revoke"] = ident.repo.count_active_credentials(customer.id) # now exactly one credential remains → the final one is protected from revocation last = ident.list_passkeys(lg2.user)[0] r["final_passkey_protected"] = _expect_fail( lambda: ident.revoke_passkey(lg2.user, sess2, last.id, now=now + 24)) r["login_after_revoke"] = _login(ident, cust_auth, now + 25).user.id == customer.id # sessions + CSRF # CSRF (anti-forgery) rejection. NB: result keys avoid the substring "csrf" so the evidence # redactor does not elide them. s = ident.repo.get_session_by_token_hash(_hash(lg2.issued.token)) r["antiforgery_missing_rejected"] = not ident.sessions.validate_csrf(s, None) r["antiforgery_mismatch_rejected"] = not ident.sessions.validate_csrf(s, "wrong") r["active_sessions_before_revoke_others"] = len(ident.repo.list_sessions(customer.id)) ident.revoke_other_sessions(lg2.user, sess2, now=now + 26) r["active_sessions_after_revoke_others"] = len(ident.repo.list_sessions(customer.id)) # pause (self) → sessions revoked, login rejected lg3 = _login(ident, cust_auth, now + 30) ident.pause_account(lg3.user, lg3.issued.session, now=now + 31) r["paused_state"] = ident.repo.get_user(customer.id).status r["paused_sessions_revoked"] = len(ident.repo.list_sessions(customer.id, active_only=True)) == 0 r["paused_login_rejected"] = _expect_fail(lambda: _login(ident, cust_auth, now + 32)) # operator reactivation (recent operator step-up from enrollment) op.reactivate_account(operator, osess, customer.id, now=now + 33) r["reactivated_state"] = ident.repo.get_user(customer.id).status r["reactivate_requires_signin"] = len( ident.repo.list_sessions(customer.id, active_only=True)) == 0 # recovery (operator-issued) → new passkey; operator never mints a session _iid, rec_code = op.issue_recovery_invitation(operator, osess, customer.id, now=now + 34) rec_auth = _SoftAuthenticator(backup=True) ropts = ident.begin_recovery(code=rec_code, now=now + 35) rres = ident.complete_recovery(credential=rec_auth.create(ropts), now=now + 36) r["recovery_state"] = rres.user.status r["recovery_old_credential_revoked"] = _expect_fail(lambda: _login(ident, cust_auth, now + 37)) r["recovery_new_credential_ok"] = _login(ident, rec_auth, now + 38).user.id == customer.id r["recovery_replay_blocked"] = _expect_fail( lambda: ident.begin_recovery(code=rec_code, now=now + 39)) # §12 deletion (requires fresh passkey assertion) lgd = _login(ident, rec_auth, now + 40) dopts = ident.begin_delete(lgd.user, lgd.issued.session, now=now + 41) receipt = ident.complete_delete(lgd.user, lgd.issued.session, credential=rec_auth.get(dopts), now=now + 42) u = ident.repo.get_user(customer.id) r["deletion"] = {"deleted": receipt.get("deleted"), "status": u.status, "profile_cleared": u.display_alias == "", "credentials_revoked": ident.repo.count_active_credentials(customer.id) == 0, "tombstone": bool(ident.repo.get_tombstone(customer.id)), "generation_monotonic": u.profile_generation > 1, "retention_disclosure_present": bool(receipt.get("note"))} r["deleted_login_rejected"] = _expect_fail(lambda: _login(ident, rec_auth, now + 43)) return r # --------------------------------------------------------------------- lifecycle helpers def _login(ident, authr, now): o = ident.begin_login(now=now) return ident.complete_login(credential=authr.get(o), now=now + 0.5) def _expect_fail(fn) -> bool: try: fn() return False except Exception: # noqa: BLE001 return True def _hash(token: str) -> str: return hashlib.sha256(token.encode()).hexdigest() # ===================================================================== stage: verify def _stage_verify() -> dict: """After a factory rebuild: prove restore-on-start recovered the synthetic DB.""" cfg = D2Config() # init_mode is False unless AMANPAY_D2_INITIALIZE=1 (should be off now) storage = D2Storage(config=cfg, object_store=_bucket(), source_commit=_deployed_commit(), store_key=store_key()) local_existed = os.path.exists(cfg.db_path) and os.path.getsize(cfg.db_path) > 0 ev = {"local_db_existed_before_open": local_existed, "recovery_event": storage.open()} try: facts = _runtime_facts(storage) facts.update(_integrity(cfg.db_path)) ev["runtime_facts"] = facts ev["counts_after_recovery"] = _counts(storage) rows = storage.db.read( "SELECT outcome, source, generation_id FROM recovery_events ORDER BY id DESC LIMIT 3") ev["recovery_events"] = [{"outcome": r[0], "source": r[1], "generation_id": r[2]} for r in rows] # tenant/operator/customer(+tombstone) survived t = storage.db.read("SELECT COUNT(*) FROM tenants WHERE slug=?", (TENANT_SLUG,))[0][0] ev["synthetic_tenant_restored"] = t == 1 ev["tombstone_restored"] = storage.db.read( "SELECT COUNT(*) FROM deletion_tombstones")[0][0] >= 1 ev["no_wal_no_shm"] = facts["no_wal_file"] and facts["no_shm_file"] # mark newest generation restore_tested (recovery succeeded from a bucket generation) gen_id = next((r[2] for r in rows if r[0] == "restored" and r[2]), "") if gen_id: with contextlib.suppress(Exception): storage.snapshots.update_manifest_flags(gen_id, restore_tested=True) ev["restore_tested_marked"] = gen_id ev["outcome"] = "success" if ev["recovery_event"].get("outcome") in ( "restored", "opened_local") else "unexpected" return ev finally: storage.close() # ================================================================== stage: negatives def _stage_negatives() -> dict: """Isolated negative-recovery checks on throwaway copies (never touch the retained gen).""" from amanpay.identity.storage.integrity import D2IntegrityVerifier from amanpay.identity.storage.recovery import RecoveryFailed v = D2IntegrityVerifier() ev: dict = {} with tempfile.TemporaryDirectory() as td: # unsupported future schema p = os.path.join(td, "future.db") c = sqlite3.connect(p) c.execute("CREATE TABLE storage_metadata(key TEXT PRIMARY KEY, value TEXT)") c.execute("INSERT INTO storage_metadata VALUES('schema_version','999')") c.execute("INSERT INTO storage_metadata VALUES('kind','amanpay-d2-accounts')") c.commit(); c.close() ev["unsupported_future_schema_rejected"] = not v.verify_database(p).ok # corrupt file pc = os.path.join(td, "corrupt.db") open(pc, "wb").write(b"not a sqlite database") ev["corrupt_snapshot_rejected"] = not v.verify_database(pc).ok # wrong-key / missing-manifest / fail-safe are covered by the merged unit tests # (tests/test_d2_persistence.py); record that linkage honestly. ev["wrong_key_and_missing_manifest"] = "covered_by_tests/test_d2_persistence.py" ev["fail_safe_no_empty_db"] = "recovery raises RecoveryFailed without init_mode" ev["outcome"] = "success" return ev # ============================================================================ main def main() -> int: stage = os.getenv("AMANPAY_D2_SPACE_PROOF_STAGE", "full").strip().lower() started = time.time() print(f"[d2-space-proof] stage={stage} commit={_deployed_commit()}") try: if stage == "full": result = _stage_full() elif stage == "verify": result = _stage_verify() elif stage == "negatives": result = _stage_negatives() else: result = {"outcome": "invalid_stage", "stage": stage} result["duration_ms"] = int((time.time() - started) * 1000) key = _put_evidence(stage, result) print(f"[d2-space-proof] stage={stage} outcome={result.get('outcome')} evidence={key}") except Exception as exc: # noqa: BLE001 — never fatal detail = {"outcome": "error", "error_class": type(exc).__name__, "error": str(exc)[:200], "duration_ms": int((time.time() - started) * 1000)} with contextlib.suppress(Exception): _put_evidence(stage, detail) print(f"[d2-space-proof] stage={stage} ERROR {type(exc).__name__}: {str(exc)[:160]}") return 0 # ALWAYS 0 — the app must start if __name__ == "__main__": raise SystemExit(main())