cortexm/enterprise/governance.py
Rama main · réplica de lectura del repositorio original.
"""Data governance: GDPR erasure, retention, backup/restore, PITR.
These are the four operations an enterprise buyer's security review
actually blocks on:
erase_user() — Art. 17 right-to-erasure: hard-delete every trace
of a subject (facts, chunks, vectors, lexicon,
aliases, vault entries), crypto-shred the PII vault,
and leave the audit chain intact (legal exemption)
with an erasure ATTESTATION record.
apply_retention() — Art. 5(1)(e) storage-limitation: expire facts and
chunks older than the policy window, keeping the
bi-temporal tombstones so history remains honest.
snapshot()/restore() — atomic, integrity-checked backup envelope
(SQLite backup API + manifest + Merkle-style digest
of the artifact).
state_at() — point-in-time recovery read: the bi-temporal Trace
replays "what did the system believe at T?" using
transaction times (tx_from/tx_to) — no snapshot
needed, the database IS the WAL.
"""
from __future__ import annotations
import hashlib
import json
import os
import time
from datetime import datetime, timezone
from cortexm.security.hashes import HashProvider
class Governance:
def __init__(self, memory) -> None:
self.memory = memory
self.store = memory.store
self.palace = memory.palace
self.audit = getattr(memory, "audit_log", None)
def _log(self, action: str, resource: str | None = None,
outcome: str = "success", meta: dict | None = None) -> None:
if self.audit is not None:
self.audit.log(action, resource=resource, outcome=outcome,
meta=meta)
# ------------------------------------------------------------ erasure
def erase_user(self, user_id: str, *, hard: bool = True,
crypto_shred: bool = True) -> dict:
"""GDPR Art. 17 — remove the subject from every layer."""
t0 = time.time()
conn = self.store.conn
counts = {}
counts["facts"] = conn.execute(
"SELECT COUNT(*) AS c FROM facts WHERE user_id=?",
(user_id,)).fetchone()["c"]
counts["chunks"] = conn.execute(
"SELECT COUNT(*) AS c FROM chunks WHERE user_id=?",
(user_id,)).fetchone()["c"]
# orphaned edges touching removed facts
fact_ids = [r["id"] for r in conn.execute(
"SELECT id FROM facts WHERE user_id=?", (user_id,))]
if fact_ids:
qmarks = ",".join("?" * len(fact_ids))
counts["edges"] = conn.execute(
f"SELECT COUNT(*) AS c FROM edges WHERE src IN ({qmarks}) "
f"OR dst IN ({qmarks})", fact_ids + fact_ids).fetchone()["c"]
conn.execute(f"DELETE FROM edges WHERE src IN ({qmarks}) "
f"OR dst IN ({qmarks})", fact_ids + fact_ids)
conn.execute("DELETE FROM facts WHERE user_id=?", (user_id,))
conn.execute("DELETE FROM chunks WHERE user_id=?", (user_id,))
# kv residue: lexicon, names, alias caches
kv_removed = 0
for k, _v in list(self.store.iter_kv(f"lexicon:{user_id}")):
self.store.kv_delete(k); kv_removed += 1
for k, _v in list(self.store.iter_kv(f"name:{user_id}")):
self.store.kv_delete(k); kv_removed += 1
counts["kv"] = kv_removed
# PII vault: crypto-shred if enabled and configured
vault = getattr(self.memory, "pii_vault", None)
if vault is not None and crypto_shred:
counts["vault_shredded"] = vault.crypto_shred()
# palace vectors for those facts
vec_removed = 0
if fact_ids:
vec_removed = self.palace.remove_ids(fact_ids)
counts["vectors"] = vec_removed
conn.commit()
# erasure attestation in the tamper-evident chain (Art. 30 records)
self._log("governance.erase", resource=user_id, outcome="erased",
meta={"counts": counts,
"duration_ms": round((time.time() - t0) * 1e3, 1),
"hard": hard, "crypto_shred": crypto_shred})
# residual scan: no row may reference the subject anywhere
residual = {
"facts": conn.execute(
"SELECT COUNT(*) AS c FROM facts WHERE user_id=?",
(user_id,)).fetchone()["c"],
"chunks": conn.execute(
"SELECT COUNT(*) AS c FROM chunks WHERE user_id=?",
(user_id,)).fetchone()["c"],
}
return {"user_id": user_id, "erased": residual["facts"] == 0
and residual["chunks"] == 0,
"counts": counts, "residual": residual,
"duration_ms": round((time.time() - t0) * 1e3, 1)}
# ------------------------------------------------------------ retention
def apply_retention(self, days: int, *, user_id: str | None = None,
dry_run: bool = False) -> dict:
"""Expire facts whose last transaction time is older than ``days``."""
if days <= 0:
raise ValueError("days must be positive")
cutoff = datetime.now(timezone.utc).timestamp() - days * 86400
cutoff_iso = datetime.fromtimestamp(cutoff, tz=timezone.utc).isoformat()
conn = self.store.conn
q = ("SELECT COUNT(*) AS c FROM facts WHERE tx_from < ? "
"AND tx_to IS NULL")
args = [cutoff_iso]
if user_id:
q += " AND user_id=?"
args.append(user_id)
stale = conn.execute(q, args).fetchone()["c"]
if dry_run:
return {"stale_facts": stale, "applied": False}
q2 = "UPDATE facts SET tx_to=?, is_active=0 WHERE tx_from < ? AND tx_to IS NULL"
args2 = [datetime.now(timezone.utc).isoformat(), cutoff_iso]
if user_id:
q2 += " AND user_id=?"
args2.append(user_id)
cur = conn.execute(q2, args2)
conn.commit()
self._log("governance.retention",
meta={"days": days, "expired": cur.rowcount,
"user_id": user_id or "all"})
return {"stale_facts": cur.rowcount, "applied": True,
"cutoff": cutoff_iso}
# ------------------------------------------------------------ snapshot
def snapshot(self, path: str) -> dict:
"""Atomic backup: online-backup the SQLite file + write a manifest.
The backup API copies page-by-page under a read transaction —
safe while writers are active."""
os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True)
if path.endswith("/"):
raise ValueError("path must be a file, not a directory")
target = sqlite_backup(self.store.db_path, path)
digest = _file_digest(target)
manifest = {
"format": "context-m-snapshot/1",
"created_at": datetime.now(timezone.utc).isoformat(),
"source_db": self.store.db_path,
"file": os.path.basename(target),
"size_bytes": os.path.getsize(target),
"sha256": digest,
"facts": self.store.conn.execute(
"SELECT COUNT(*) AS c FROM facts").fetchone()["c"],
"audit_head": (self.audit.verify()["head_hash"]
if self.audit else None),
}
mpath = target + ".manifest.json"
with open(mpath, "w", encoding="utf-8") as fh:
json.dump(manifest, fh, indent=2)
self._log("governance.snapshot",
resource=path, meta={"sha256": digest,
"size": manifest["size_bytes"]})
return {"path": target, "manifest": manifest, "manifest_path": mpath}
def restore(self, snapshot_path: str, *, verify: bool = True) -> dict:
"""Restore from a snapshot manifest pair. Refuses mismatched digests."""
mpath = snapshot_path + ".manifest.json"
if verify and not os.path.exists(mpath):
raise FileNotFoundError("manifest missing — cannot verify integrity")
if os.path.exists(mpath):
with open(mpath, encoding="utf-8") as fh:
manifest = json.load(fh)
if verify:
digest = _file_digest(snapshot_path)
if digest != manifest["sha256"]:
raise ValueError(
f"integrity check failed: {digest} != {manifest['sha256']}")
# close current handles, replace file, reopen
db_path = self.store.db_path
self.memory.close()
import shutil
shutil.copyfile(snapshot_path, db_path)
self.memory._reopen()
# _reopen() built a NEW Governance object; this method is running
# on the old one — rebind self so the audit attestation below
# writes through the fresh store, not the closed one.
self.store = self.memory.store
self.palace = self.memory.palace
self.audit = self.memory.audit_log
self._log("governance.restore", resource=snapshot_path)
return {"restored": db_path, "verified": verify}
# ------------------------------------------------------------ PITR
def state_at(self, when, *, user_id: str | None = None,
limit: int = 500) -> list[dict]:
"""Point-in-time read: facts the system believed true at ``when``
(transaction-time replay — the database is its own WAL)."""
from cortexm.api.memory import parse_ts
ts = parse_ts(when) if not isinstance(when, datetime) else when
if ts.tzinfo is None:
ts = ts.replace(tzinfo=timezone.utc)
# normalize to the Z-suffix format facts are stored with, so the
# string comparison in SQL matches the stored tx_from exactly
iso = ts.strftime("%Y-%m-%dT%H:%M:%SZ")
q = ("SELECT * FROM facts WHERE tx_from <= ? AND "
"(tx_to IS NULL OR tx_to > ?) AND quarantined=0")
args: list = [iso, iso]
if user_id:
q += " AND user_id=?"
args.append(user_id)
q += " ORDER BY valid_from DESC LIMIT ?"
args.append(limit)
rows = [dict(r) for r in self.store.conn.execute(q, args)]
self._log("governance.pitr", meta={"when": iso, "rows": len(rows)})
return rows
# ------------------------------------------------------------------ helpers
def sqlite_backup(src_db: str, dst_path: str) -> str:
import sqlite3
src = sqlite3.connect(src_db)
dst = sqlite3.connect(dst_path)
with dst:
src.backup(dst)
dst.close()
src.close()
return dst_path
def _file_digest(path: str) -> str:
h = hashlib.sha256()
with open(path, "rb") as fh:
for chunk in iter(lambda: fh.read(1 << 20), b""):
h.update(chunk)
return h.hexdigest()