GDPR & CCPA Privacy Engineering - Lawful Bases, DSARs, Erasure in Practice, Data Transfers, DPIAs & Consent

"""Right to erasure in an append-only world: crypto-shredding.

Concept first:
  * GDPR Art. 17 says erase personal data "without undue delay" when a ground
    applies. Mutable tables are easy (DELETE). Append-only event logs, Kafka
    topics with long retention, immutable backups and warehouse snapshots are not.
  * Crypto-shredding: encrypt each subject's personal fields with a per-subject
    data key (DEK). Erasure = destroy the DEK. Every copy of the ciphertext, in
    every log, replica and backup, becomes unreadable at once.
  * It only works if plaintext never leaked elsewhere (search index, warehouse,
    logs, caches). Those still need their own delete/rebuild path.
Cipher: dependency-free HMAC-SHA256 keystream + tag for the demo; production
uses AES-GCM with DEKs wrapped by a KMS key (see envelope encryption page).
"""
import hmac, hashlib, json, secrets

class KeyStore:                       # in production: KMS-wrapped DEKs in a small, backed-up table
    def __init__(self): self.keys: dict[str, bytes] = {}
    def dek(self, subject: str) -> bytes:
        return self.keys.setdefault(subject, secrets.token_bytes(32))
    def shred(self, subject: str) -> None:
        self.keys.pop(subject, None)

def _ks(key: bytes, nonce: bytes, n: int) -> bytes:
    out = b""; i = 0
    while len(out) < n:
        out += hmac.new(key, nonce + i.to_bytes(4, "big"), hashlib.sha256).digest(); i += 1
    return out[:n]

def seal(key: bytes, pt: str) -> str:
    nonce = secrets.token_bytes(12); b = pt.encode()
    ct = bytes(x ^ y for x, y in zip(b, _ks(key, nonce, len(b))))
    tag = hmac.new(key, nonce + ct, hashlib.sha256).digest()[:16]
    return (nonce + tag + ct).hex()

def open_(key: bytes | None, blob: str) -> str:
    if key is None:
        return "<unreadable: key shredded>"
    raw = bytes.fromhex(blob); nonce, tag, ct = raw[:12], raw[12:28], raw[28:]
    assert hmac.compare_digest(tag, hmac.new(key, nonce + ct, hashlib.sha256).digest()[:16])
    return bytes(x ^ y for x, y in zip(ct, _ks(key, nonce, len(ct)))).decode()

ks = KeyStore()
event_log: list[dict] = []            # append-only; also replicated + backed up
def emit(subject: str, kind: str, pii: dict, non_personal: dict) -> None:
    event_log.append({"subject_ref": subject, "kind": kind,
                      "pii": seal(ks.dek(subject), json.dumps(pii)), **non_personal})

emit("u_101", "signup", {"email": "lena@example.eu", "name": "Lena K"}, {"country": "DE", "plan": "pro"})
emit("u_102", "signup", {"email": "omar@example.fr", "name": "Omar B"}, {"country": "FR", "plan": "free"})
emit("u_101", "address_changed", {"street": "Hauptstr 5", "city": "Berlin"}, {"country": "DE"})
backup = json.loads(json.dumps(event_log))       # nightly immutable backup copy

def project(log: list[dict]) -> list[str]:
    return [f"{e['subject_ref']}:{e['kind']}:{open_(ks.keys.get(e['subject_ref']), e['pii'])}" for e in log]

print("before erasure:")
for row in project(event_log):
    print("  ", row)
ks.shred("u_101")                                 # erasure request for u_101
print("after shredding u_101's key (log untouched, backup untouched):")
for row in project(event_log):
    print("  ", row)
print("backup copy, same result:", project(backup)[0])
# non-personal analytics survive erasure because they were never personal
print("aggregate still works:", {c: sum(1 for e in event_log if e.get("country") == c and e["kind"] == "signup") for c in ("DE", "FR")})
print("still to purge separately: search index docs, warehouse rows, caches, vendor copies -> see DSAR orchestrator")

before erasure:
   u_101:signup:{"email": "lena@example.eu", "name": "Lena K"}
   u_102:signup:{"email": "omar@example.fr", "name": "Omar B"}
   u_101:address_changed:{"street": "Hauptstr 5", "city": "Berlin"}
after shredding u_101's key (log untouched, backup untouched):
   u_101:signup:<unreadable: key shredded>
   u_102:signup:{"email": "omar@example.fr", "name": "Omar B"}
   u_101:address_changed:<unreadable: key shredded>
backup copy, same result: u_101:signup:<unreadable: key shredded>
aggregate still works: {'DE': 1, 'FR': 1}
still to purge separately: search index docs, warehouse rows, caches, vendor copies -> see DSAR orchestrator

// DSAR / erasure orchestrator: one request, many systems, two legal clocks.
// Concept: a deletion or access request is a distributed workflow, not a SQL
// statement. Every system holding personal data registers a handler; the
// orchestrator fans out, tracks deadlines, respects legal holds, and records
// evidence. Deadlines: GDPR Art. 12(3) = one month (extendable by two more
// months for complex/numerous requests); CCPA = 45 days (one 45-day extension).

type Law = "GDPR" | "CCPA";
type Outcome = "deleted" | "shredded" | "suppressed-until-expiry" | "held" | "forwarded" | "pending";
interface Handler { system: string; run: (subject: string) => Outcome; note: string }
interface DsarRequest { id: string; subject: string; law: Law; received: Date; heldSystems: string[] }  // legal hold scope, set by counsel

function addDays(d: Date, n: number): Date { const x = new Date(d); x.setUTCDate(x.getUTCDate() + n); return x; }
function addMonths(d: Date, n: number): Date { const x = new Date(d); x.setUTCMonth(x.getUTCMonth() + n); return x; }
const iso = (d: Date) => d.toISOString().slice(0, 10);

function deadline(r: DsarRequest): { due: Date; maxExtended: Date } {
  return r.law === "GDPR"
    ? { due: addMonths(r.received, 1), maxExtended: addMonths(r.received, 3) }
    : { due: addDays(r.received, 45), maxExtended: addDays(r.received, 90) };
}

const handlers: Handler[] = [
  { system: "primary Postgres", run: () => "deleted", note: "hard delete rows, cascade" },
  { system: "event log (Kafka)", run: () => "shredded", note: "destroy per-subject DEK" },
  { system: "search index", run: () => "deleted", note: "delete by subject_ref, then verify count = 0" },
  { system: "warehouse", run: () => "deleted", note: "delete + reprocess snapshots older than the delete" },
  { system: "backups (35-day)", run: () => "suppressed-until-expiry", note: "beyond use; on restore, replay suppression list" },
  { system: "email vendor (processor)", run: () => "forwarded", note: "Art. 28 / service-provider contract obliges them to delete" },
];

function handleRequest(r: DsarRequest) {
  const { due, maxExtended } = deadline(r);
  console.log(`== ${r.id} ${r.law} received ${iso(r.received)} due ${iso(due)} (max with extension ${iso(maxExtended)})`);
  const evidence = handlers.map((h) => {
    const outcome: Outcome = r.heldSystems.includes(h.system) ? "held" : h.run(r.subject);
    return { system: h.system, outcome, note: outcome === "held" ? "legal hold: retain, restrict processing, tell requester why" : h.note };
  });
  for (const e of evidence) console.log(`   ${e.system.padEnd(24)} ${e.outcome.padEnd(24)} ${e.note}`);
  const held = evidence.filter((e) => e.outcome === "held").length;
  const waiting = evidence.filter((e) => e.outcome === "forwarded").length;
  console.log(`   status: ${held ? "PARTIAL (" + held + " held)" : "DONE"}; ${waiting} processor confirmation(s) awaited; receipt stores subject_ref + outcomes, not the data`);
}

handleRequest({ id: "DSAR-81", subject: "u_101", law: "GDPR", received: new Date("2026-10-08T00:00:00Z"), heldSystems: [] });
handleRequest({ id: "DSAR-82", subject: "u_555", law: "CCPA", received: new Date("2026-10-08T00:00:00Z"), heldSystems: ["primary Postgres", "event log (Kafka)", "backups (35-day)"] });

== DSAR-81 GDPR received 2026-10-08 due 2026-11-08 (max with extension 2027-01-08)
   primary Postgres         deleted                  hard delete rows, cascade
   event log (Kafka)        shredded                 destroy per-subject DEK
   search index             deleted                  delete by subject_ref, then verify count = 0
   warehouse                deleted                  delete + reprocess snapshots older than the delete
   backups (35-day)         suppressed-until-expiry  beyond use; on restore, replay suppression list
   email vendor (processor) forwarded                Art. 28 / service-provider contract obliges them to delete
   status: DONE; 1 processor confirmation(s) awaited; receipt stores subject_ref + outcomes, not the data
== DSAR-82 CCPA received 2026-10-08 due 2026-11-22 (max with extension 2027-01-06)
   primary Postgres         held                     legal hold: retain, restrict processing, tell requester why
   event log (Kafka)        held                     legal hold: retain, restrict processing, tell requester why
   search index             deleted                  delete by subject_ref, then verify count = 0
   warehouse                deleted                  delete + reprocess snapshots older than the delete
   backups (35-day)         held                     legal hold: retain, restrict processing, tell requester why
   email vendor (processor) forwarded                Art. 28 / service-provider contract obliges them to delete
   status: PARTIAL (3 held); 1 processor confirmation(s) awaited; receipt stores subject_ref + outcomes, not the data