Threat-Model a Warehouse Including BI Exports, Shared Credentials, Staging Data, Backups, and Over-Broad Analyst Roles

asset/path                         threat/failure                         evidence/controlhuman identity                      shared account / weak attribution             disable shared identities; audit unique subjectservice identity                    over-broad BI/admin privileges                separate pipeline_writer from export/admin rolessecure semantic/customer view       policy bypass through raw table               enumerate effective raw/staging grantsraw/staging customer data           direct identifier exposure                   restricted sensitivity + narrow service grantsmasked BI export                    copied data escapes live policy                export role + copy registry + retention ownerimmutable backup                    deleted subject can reappear on restore       quarantine/expiry + erasure overlay on restoreprivacy operator                    can erase and grant own privileges            separation of duties: erase yes, ADMIN nosecurity administrator              can manage policy and read all PII by default avoid implicit data access; separate admin/data rolesdev environment                     accidental production access                 environment-bound identities/resources

initial shared_credentials = [shared_analyst]initial bypass_paths = [marketing_analyst -> prod.raw_customer]repaired shared_credentials = []repaired bypass_paths = []repaired separation_of_duties = []finance mart SELECT = ALLOWfinance PII SELECT = DENYETL raw INSERT = ALLOWETL security ADMIN = DENY

# Bash / Git Bashexport ATLASMART_TOKEN_KEY="synthetic-lab-key-2026"python ch22_lab.py# PowerShell$env:ATLASMART_TOKEN_KEY = "synthetic-lab-key-2026"python .\ch22_lab.py

from __future__ import annotationsimport hashlib, hmac, json, os, sqlite3, sysfrom pathlib import PathOUT=Path('/mnt/data/atlasmart_ch22_lab'); OUT.mkdir(parents=True,exist_ok=True)DB=OUT/'atlasmart_ch22.sqlite'if DB.exists(): DB.unlink()KEY=os.environ.get('ATLASMART_TOKEN_KEY')if not KEY:    raise SystemExit('Set ATLASMART_TOKEN_KEY to a synthetic lab-only value before running.')SALES=[('O1001',1,'2026-09-18','P100','D-CUST-001','web',2,10000,6000,'USD','paid'),('O1001',2,'2026-09-18','P200','D-CUST-001','web',1,2500,1500,'USD','paid'),('O1002',1,'2026-09-18','P300','D-CUST-002','mobile',1,19000,12000,'USD','paid'),('O1003',1,'2026-09-19','P400','D-CUST-001','web',1,10000,6500,'USD','paid'),('O1005',1,'2026-09-20','P100','D-CUST-004','mobile',1,5000,3000,'USD','paid'),('O1005',2,'2026-09-20','P400','D-CUST-004','mobile',1,10000,6500,'USD','paid'),('O1007',1,'2026-09-21','P300','D-CUST-003','sales',1,7500,4500,'USD','paid'),('O1008',1,'2026-09-21','P200','D-CUST-002','mobile',2,5000,2500,'USD','paid'),('O1009',1,'2026-09-22','P100','D-CUST-005','web',1,5000,2500,'USD','paid'),('O1010',1,'2026-09-22','P100','D-CUST-003','sales',1,8000,4500,'USD','paid')]CUSTOMERS=[('D-CUST-001','C001','Ada Retail','ada@example.test','Mid-Market','G-NORTH','active',1),('D-CUST-002','C002','Ben Home','ben@example.test','Consumer','G-EAST','active',1),('D-CUST-003','C003','Cyra Labs','cyra@example.test','Enterprise','G-EAST','active',1),('D-CUST-004','C004','Dara Studio','dara@example.test','Consumer','G-SOUTH','inactive',0),('D-CUST-005','C005','__INFERRED__',None,'__UNKNOWN__','__UNKNOWN__','active',0)]EXPECTED={'lines':10,'orders':8,'units':12,'revenue_usd':820,'cost_usd':495,'profit_usd':325}IDENTITIES=[('alice.marketing','human','prod',0,1),('frank.finance','human','prod',0,1),('olivia.ops','human','prod',0,1),('svc_etl_prod','service','prod',0,1),('svc_bi_export_prod','service','prod',0,1),('priya.privacy','human','prod',0,1),('sara.security','human','prod',0,1),('dev.dana','human','dev',0,1),('shared_analyst','human','prod',1,1)]ROLE_BINDINGS=[('alice.marketing','marketing_analyst','prod'),('frank.finance','finance_analyst','prod'),('olivia.ops','operations_analyst','prod'),('svc_etl_prod','pipeline_writer','prod'),('svc_bi_export_prod','bi_exporter','prod'),('priya.privacy','privacy_operator','prod'),('sara.security','security_admin','prod'),('dev.dana','developer','dev'),('shared_analyst','marketing_analyst','prod')]RESOURCES=[('prod.fact_sales','prod','internal'),('prod.dim_customer','prod','restricted'),('prod.raw_customer','prod','restricted'),('prod.secure_customer_view','prod','restricted'),('prod.finance_mart','prod','internal'),('prod.bi_export_masked','prod','restricted'),('prod.security_catalog','prod','confidential'),('dev.dim_customer','dev','synthetic')]# Initial grants deliberately contain one bypass: marketing_analyst can read prod.raw_customer.GRANTS=[('marketing_analyst','prod.secure_customer_view','SELECT','ALLOW'),('marketing_analyst','prod.raw_customer','SELECT','ALLOW'),('finance_analyst','prod.finance_mart','SELECT','ALLOW'),('operations_analyst','prod.fact_sales','SELECT','ALLOW'),('pipeline_writer','prod.raw_customer','INSERT','ALLOW'),('pipeline_writer','prod.dim_customer','UPSERT','ALLOW'),('bi_exporter','prod.secure_customer_view','SELECT','ALLOW'),('bi_exporter','prod.bi_export_masked','INSERT','ALLOW'),('privacy_operator','prod.dim_customer','SELECT_PII','ALLOW'),('privacy_operator','prod.dim_customer','ERASE_PII','ALLOW'),('privacy_operator','prod.raw_customer','ERASE_PII','ALLOW'),('security_admin','prod.security_catalog','ADMIN','ALLOW'),('developer','dev.dim_customer','SELECT','ALLOW')]PII=[('prod.dim_customer','customer_name','direct_identifier','restricted'),('prod.dim_customer','email','contact_identifier','restricted'),('prod.dim_customer','durable_customer_id','pseudonymous_identifier','restricted'),('prod.dim_customer','segment','business_attribute','internal'),('prod.raw_customer','customer_name','direct_identifier','restricted'),('prod.raw_customer','email','contact_identifier','restricted')]PURPOSE_POLICIES=[('marketing_campaign_analysis','marketing_analyst','token,segment,geography_code,masked_email','no direct name/email; consent policy evaluated separately'),('finance_reporting','finance_analyst','aggregate revenue/cost/profit','no customer direct identifiers required'),('privacy_request','privacy_operator','direct identifiers only for verified privacy workflow','case-bound access; audited'),('pipeline_processing','pipeline_writer','raw/integration fields required by governed load','non-human service identity; no ad-hoc BI export')]COPIES=[('COPY-LIVE-DIM','prod.dim_customer','live_table','active',1,None),('COPY-RAW','prod.raw_customer','raw_generation','active',1,None),('COPY-EXPORT','prod.bi_export_masked','export','active',1,None),('COPY-BACKUP-20260920','backup://prod/2026-09-20','immutable_backup','active',1,'2026-10-21')]def sha(v:str)->str: return hashlib.sha256(v.encode()).hexdigest()def token(durable_id:str)->str:    return 'tok_'+hmac.new(KEY.encode(),durable_id.encode(),hashlib.sha256).hexdigest()[:20]def masked_email(email):    if not email: return None    local,domain=email.split('@',1)    return (local[:1]+'***@'+domain) if local else '***@'+domaindef setup():    c=sqlite3.connect(DB)    c.row_factory=sqlite3.Row    c.executescript('''    PRAGMA foreign_keys=ON;    CREATE TABLE fact_sales(order_id TEXT,line_no INT,order_date TEXT,product_id TEXT,durable_customer_id TEXT,channel TEXT,quantity INT,amount_cents INT,cost_cents INT,currency_code TEXT,status TEXT,PRIMARY KEY(order_id,line_no));    CREATE TABLE dim_customer(durable_customer_id TEXT PRIMARY KEY,source_customer_id TEXT,customer_name TEXT,email TEXT,segment TEXT,geography_code TEXT,lifecycle_status TEXT,marketing_consent INT);    CREATE TABLE raw_customer(durable_customer_id TEXT PRIMARY KEY,source_customer_id TEXT,customer_name TEXT,email TEXT,raw_generation INT NOT NULL,erased INT NOT NULL DEFAULT 0);    CREATE TABLE token_vault(durable_customer_id TEXT PRIMARY KEY,customer_token TEXT UNIQUE NOT NULL);    CREATE TABLE identity_registry(identity TEXT PRIMARY KEY,identity_type TEXT,environment TEXT,is_shared INT,enabled INT);    CREATE TABLE role_binding(identity TEXT,role TEXT,environment TEXT,PRIMARY KEY(identity,role,environment));    CREATE TABLE resource_catalog(resource TEXT PRIMARY KEY,environment TEXT,sensitivity TEXT);    CREATE TABLE permission(role TEXT,resource TEXT,action TEXT,effect TEXT,PRIMARY KEY(role,resource,action));    CREATE TABLE pii_catalog(resource TEXT,column_name TEXT,classification TEXT,sensitivity TEXT,PRIMARY KEY(resource,column_name));    CREATE TABLE purpose_policy(purpose TEXT,role TEXT,allowed_fields TEXT,restriction TEXT,PRIMARY KEY(purpose,role));    CREATE TABLE access_audit(event_id INTEGER PRIMARY KEY AUTOINCREMENT,event_time TEXT,identity TEXT,resource TEXT,action TEXT,decision TEXT,reason TEXT);    CREATE TABLE copy_registry(copy_id TEXT PRIMARY KEY,location TEXT,copy_type TEXT,status TEXT,contains_subject INT,retention_until TEXT);    CREATE TABLE deletion_manifest(case_id TEXT PRIMARY KEY,durable_customer_id TEXT,requested_at TEXT,executed_at TEXT,scope TEXT,before_hash TEXT,after_hash TEXT,backup_disposition TEXT,status TEXT);    CREATE TABLE erasure_tombstone(durable_customer_id TEXT PRIMARY KEY,case_id TEXT,applies_from TEXT);    CREATE TABLE operator_note(note_id TEXT PRIMARY KEY,case_id TEXT,note TEXT);    CREATE VIEW finance_mart AS SELECT order_date,SUM(amount_cents)/100.0 revenue_usd,SUM(cost_cents)/100.0 cost_usd,SUM(amount_cents-cost_cents)/100.0 profit_usd FROM fact_sales WHERE status='paid' AND currency_code='USD' GROUP BY order_date;    ''')    c.executemany('INSERT INTO fact_sales VALUES (?,?,?,?,?,?,?,?,?,?,?)',SALES)    c.executemany('INSERT INTO dim_customer VALUES (?,?,?,?,?,?,?,?)',CUSTOMERS)    c.executemany('INSERT INTO raw_customer VALUES (?,?,?,?,1,0)',[(x[0],x[1],x[2],x[3]) for x in CUSTOMERS])    c.executemany('INSERT INTO token_vault VALUES (?,?)',[(x[0],token(x[0])) for x in CUSTOMERS])    c.executemany('INSERT INTO identity_registry VALUES (?,?,?,?,?)',IDENTITIES)    c.executemany('INSERT INTO role_binding VALUES (?,?,?)',ROLE_BINDINGS)    c.executemany('INSERT INTO resource_catalog VALUES (?,?,?)',RESOURCES)    c.executemany('INSERT INTO permission VALUES (?,?,?,?)',GRANTS)    c.executemany('INSERT INTO pii_catalog VALUES (?,?,?,?)',PII)    c.executemany('INSERT INTO purpose_policy VALUES (?,?,?,?)',PURPOSE_POLICIES)    c.executemany('INSERT INTO copy_registry VALUES (?,?,?,?,?,?)',COPIES)    c.commit(); return cdef controls(c):    r=c.execute("SELECT count(*),count(distinct order_id),sum(quantity),sum(amount_cents),sum(cost_cents),sum(amount_cents-cost_cents) FROM fact_sales WHERE status='paid' AND currency_code='USD'").fetchone()    return {'lines':r[0],'orders':r[1],'units':r[2],'revenue_usd':r[3]//100,'cost_usd':r[4]//100,'profit_usd':r[5]//100}def roles(c,identity,env):    return {r[0] for r in c.execute('SELECT role FROM role_binding WHERE identity=? AND environment=?',(identity,env))}def authorize(c,identity,resource,action,now='2026-09-21T13:00:00Z'):    i=c.execute('SELECT environment,is_shared,enabled FROM identity_registry WHERE identity=?',(identity,)).fetchone()    rr=c.execute('SELECT environment FROM resource_catalog WHERE resource=?',(resource,)).fetchone()    if not i or not i['enabled']:        decision,reason='DENY','unknown or disabled identity'    elif not rr:        decision,reason='DENY','unknown resource'    elif i['environment']!=rr['environment']:        decision,reason='DENY','environment isolation'    else:        rs=roles(c,identity,rr['environment'])        hits=c.execute("SELECT role,effect FROM permission WHERE resource=? AND action=? AND role IN (%s)" % (','.join('?'*len(rs)) if rs else "''"),(resource,action,*sorted(rs)) if rs else (resource,action)).fetchall()        decision='ALLOW' if any(h['effect']=='ALLOW' for h in hits) else 'DENY'        reason='role grant' if decision=='ALLOW' else 'no matching least-privilege grant'    c.execute('INSERT INTO access_audit(event_time,identity,resource,action,decision,reason) VALUES (?,?,?,?,?,?)',(now,identity,resource,action,decision,reason)); c.commit()    return decision,reasondef secure_customer_view(c,identity,purpose='marketing_campaign_analysis'):    dec,_=authorize(c,identity,'prod.secure_customer_view','SELECT')    if dec!='ALLOW': return []    rs=roles(c,identity,'prod')    out=[]    for r in c.execute('SELECT * FROM dim_customer ORDER BY durable_customer_id'):        # Marketing sees only consented customers; BI exporter receives the same policy-filtered rows.        if ('marketing_analyst' in rs or 'bi_exporter' in rs) and not r['marketing_consent']:            continue        out.append({'customer_token':token(r['durable_customer_id']),'segment':r['segment'],'geography_code':r['geography_code'],'masked_email':masked_email(r['email'])})    return outdef credential_findings(c):    rows=c.execute('SELECT identity,identity_type,environment FROM identity_registry WHERE is_shared=1 AND enabled=1').fetchall()    return [dict(r) for r in rows]def separation_findings(c):    forbidden=[('pipeline_writer','security_admin'),('privacy_operator','security_admin')]    findings=[]    for ident in [r[0] for r in c.execute('SELECT identity FROM identity_registry WHERE enabled=1')]:        rs=roles(c,ident,'prod')        for a,b in forbidden:            if a in rs and b in rs: findings.append({'identity':ident,'roles':[a,b]})    return findingsdef bypass_findings(c):    findings=[]    # Any analytical/export role able to read raw direct identifiers is a bypass.    for role in ('marketing_analyst','finance_analyst','operations_analyst','bi_exporter'):        row=c.execute("SELECT effect FROM permission WHERE role=? AND resource='prod.raw_customer' AND action='SELECT'",(role,)).fetchone()        if row and row[0]=='ALLOW': findings.append({'role':role,'resource':'prod.raw_customer','reason':'bypasses secure view and exposes direct identifiers'})    return findingsdef repair_identity_and_bypass(c):    c.execute("UPDATE identity_registry SET enabled=0 WHERE identity='shared_analyst'")    c.execute("DELETE FROM permission WHERE role='marketing_analyst' AND resource='prod.raw_customer' AND action='SELECT'")    c.commit()def direct_pii_snapshot(c,did):    r=c.execute('SELECT durable_customer_id,source_customer_id,customer_name,email,segment,geography_code,marketing_consent FROM dim_customer WHERE durable_customer_id=?',(did,)).fetchone()    return dict(r) if r else Nonedef erase_subject(c,did='D-CUST-004',case_id='ERASE-20260921-004'):    prior=c.execute('SELECT status FROM deletion_manifest WHERE case_id=?',(case_id,)).fetchone()    if prior: return 'replay_ignored'    before=direct_pii_snapshot(c,did); before_hash=sha(json.dumps(before,sort_keys=True))    c.execute('BEGIN')    c.execute("UPDATE dim_customer SET source_customer_id=NULL,customer_name='[ERASED]',email=NULL,marketing_consent=0 WHERE durable_customer_id=?",(did,))    # Governed exception to raw immutability: active raw direct identifiers are rewritten and the generation advances.    c.execute("UPDATE raw_customer SET source_customer_id=NULL,customer_name='[ERASED]',email=NULL,raw_generation=raw_generation+1,erased=1 WHERE durable_customer_id=?",(did,))    c.execute('DELETE FROM token_vault WHERE durable_customer_id=?',(did,))    c.execute('INSERT INTO erasure_tombstone VALUES (?,?,?)',(did,case_id,'2026-09-21T13:30:00Z'))    # Active copies can be sanitized; immutable backup is quarantined/pending expiry, not falsely reported as deleted in place.    c.execute("UPDATE copy_registry SET contains_subject=0,status='sanitized' WHERE copy_type IN ('live_table','raw_generation','export')")    c.execute("UPDATE copy_registry SET status='quarantined_pending_expiry' WHERE copy_type='immutable_backup' AND contains_subject=1")    after=direct_pii_snapshot(c,did); after_hash=sha(json.dumps(after,sort_keys=True))    scope='dim_customer, active raw generation, token vault, active exports; backup tracked separately'    backup='immutable backup quarantined; expiry/rewrite depends on approved retention/legal policy; restore hook must reapply tombstone'    c.execute('INSERT INTO deletion_manifest VALUES (?,?,?,?,?,?,?,?,?)',(case_id,did,'2026-09-21T13:20:00Z','2026-09-21T13:30:00Z',scope,before_hash,after_hash,backup,'executed_with_backup_followup'))    c.execute('INSERT INTO operator_note VALUES (?,?,?)',('NOTE-ERASE-004',case_id,'Technical lab workflow only; controller/legal policy decides grounds, exceptions, retention, backups, downstream processors, and proof requirements.'))    c.commit(); return {'before_hash':before_hash,'after_hash':after_hash,'status':'executed_with_backup_followup'}def restore_overlay_test(c,did='D-CUST-004'):    # Simulate a restore row that would reintroduce PII, then apply the erasure tombstone before release.    restored={'durable_customer_id':did,'source_customer_id':'C004','customer_name':'Dara Studio','email':'dara@example.test'}    tomb=c.execute('SELECT case_id FROM erasure_tombstone WHERE durable_customer_id=?',(did,)).fetchone()    if tomb:        restored.update({'source_customer_id':None,'customer_name':'[ERASED]','email':None})    return {'tombstone_found':bool(tomb),'released_name':restored['customer_name'],'released_email':restored['email'],'status':'PASS' if restored['customer_name']=='[ERASED]' and restored['email'] is None else 'FAIL'}def main():    c=setup(); ctl_before=controls(c); assert ctl_before==EXPECTED,ctl_before    initial_creds=credential_findings(c); assert [x['identity'] for x in initial_creds]==['shared_analyst']    initial_bypass=bypass_findings(c); assert initial_bypass and initial_bypass[0]['role']=='marketing_analyst'    assert separation_findings(c)==[]    # Masked view test: direct names are never returned; consented rows only.    mkt=secure_customer_view(c,'alice.marketing'); assert len(mkt)==3 and all('customer_name' not in r for r in mkt) and all(r['masked_email'] for r in mkt)    # Initial bypass is real even though the secure view masks.    d1,_=authorize(c,'alice.marketing','prod.raw_customer','SELECT'); assert d1=='ALLOW'    # Environment isolation: a dev identity cannot query prod restricted data.    d2,reason2=authorize(c,'dev.dana','prod.dim_customer','SELECT_PII'); assert d2=='DENY' and reason2=='environment isolation'    # Service identity is narrow: can write raw but cannot administer security or export BI.    assert authorize(c,'svc_etl_prod','prod.raw_customer','INSERT')[0]=='ALLOW'    assert authorize(c,'svc_etl_prod','prod.security_catalog','ADMIN')[0]=='DENY'    assert authorize(c,'svc_etl_prod','prod.bi_export_masked','INSERT')[0]=='DENY'    repair_identity_and_bypass(c)    assert credential_findings(c)==[] and bypass_findings(c)==[]    assert authorize(c,'alice.marketing','prod.raw_customer','SELECT')[0]=='DENY'    # Finance gets governed aggregate, not PII.    assert authorize(c,'frank.finance','prod.finance_mart','SELECT')[0]=='ALLOW'    assert authorize(c,'frank.finance','prod.dim_customer','SELECT_PII')[0]=='DENY'    # Privacy operator can perform case-bound PII workflow but cannot administer grants.    assert authorize(c,'priya.privacy','prod.dim_customer','ERASE_PII')[0]=='ALLOW'    assert authorize(c,'priya.privacy','prod.security_catalog','ADMIN')[0]=='DENY'    er=erase_subject(c); assert er!='replay_ignored'    assert erase_subject(c)=='replay_ignored'    after=direct_pii_snapshot(c,'D-CUST-004'); assert after['customer_name']=='[ERASED]' and after['email'] is None and after['source_customer_id'] is None    assert c.execute("SELECT count(*) FROM token_vault WHERE durable_customer_id='D-CUST-004'").fetchone()[0]==0    raw=c.execute("SELECT customer_name,email,raw_generation,erased FROM raw_customer WHERE durable_customer_id='D-CUST-004'").fetchone(); assert tuple(raw)==('[ERASED]',None,2,1)    ctl_after=controls(c); assert ctl_after==EXPECTED    restore=restore_overlay_test(c); assert restore['status']=='PASS'    copies=[dict(r) for r in c.execute('SELECT * FROM copy_registry ORDER BY copy_id')]    assert [r['status'] for r in copies if r['copy_type']=='immutable_backup']==['quarantined_pending_expiry']    manifest=dict(c.execute('SELECT * FROM deletion_manifest').fetchone())    access=[dict(r) for r in c.execute('SELECT * FROM access_audit ORDER BY event_id')]    report={      'runtime':{'python':sys.version.split()[0],'sqlite':sqlite3.sqlite_version},      'controls_before':ctl_before,'controls_after_erasure':ctl_after,      'initial_findings':{'shared_credentials':initial_creds,'bypass_paths':initial_bypass,'separation_of_duties':[]},      'repaired_findings':{'shared_credentials':credential_findings(c),'bypass_paths':bypass_findings(c),'separation_of_duties':separation_findings(c)},      'masked_marketing_rows':mkt,      'access_matrix_tests':access,      'pii_catalog':[dict(r) for r in c.execute('SELECT * FROM pii_catalog ORDER BY resource,column_name')],      'purpose_policies':[dict(r) for r in c.execute('SELECT * FROM purpose_policy ORDER BY purpose')],      'copy_registry':copies,'deletion_manifest':manifest,'restore_overlay_test':restore,      'erasure_replay':'replay_ignored',      'privacy_note':'Technical teaching workflow only. Jurisdiction, controller policy, legal basis, exceptions, retention, backups, processors, and proof requirements must be decided outside this lab.',      'tokenization_note':'Tokens are pseudonymous identifiers, not proof of anonymization; auxiliary mapping/key access changes re-identification risk.'    }    (OUT/'report.json').write_text(json.dumps(report,indent=2,sort_keys=True),encoding='utf-8')    (OUT/'access_matrix.json').write_text(json.dumps(access,indent=2),encoding='utf-8')    (OUT/'deletion_manifest.json').write_text(json.dumps(manifest,indent=2,sort_keys=True),encoding='utf-8')    print(json.dumps(report,indent=2,sort_keys=True)); c.close()if __name__=='__main__': main()

ERASE-20260921-004before direct identifiers: source_customer_id=C004, customer_name=Dara Studio, email=dara@example.testafter direct identifiers:  source_customer_id=NULL, customer_name=[ERASED], email=NULLtoken_vault mapping:       removedactive raw generation:     rewritten under authorized privacy exception; generation 1 -> 2active export copy:         sanitizedimmutable backup:           quarantined_pending_expiry (not falsely reported deleted in place)restore-overlay test:       PASS; erasure tombstone reapplied before restored data is releasedrerun same case:            replay_ignoredwarehouse controls:         unchanged at 10 lines / 8 orders / 12 units / 820 / 495 / 325 USD