#!/usr/bin/env python3 """SOC 2 evidence populations from Claude Enterprise Compliance API Activity Feed records, in DuckDB. python3 soc2_claude_evidence.py # clean generated records python3 soc2_claude_evidence.py --plant # generated records with nine planted faults python3 soc2_claude_evidence.py --plant --members # the same, with the generated members export python3 soc2_claude_evidence.py --strict # exit 1 when anything needs follow-up Builds three lists a SOC 2 auditor may ask for, each printed with its period, filters, run time, row count and a SHA-256 of its rows: CC6.2 every leaver in the period, the date of the first removal record credited to that departure, the days between, and any use of the account after that removal; with MEMBERS_FILE, only leavers who held a seat or appear in the feed CC8.1 every administrative setting change, who made it, and whether they are on the admin roster CC7.2 a count of Compliance API requests, by key Before it builds them it runs five input checks (no duplicate ids, no empty timestamps, no day without records, no HR row without an email or last day, no roster row without an email), and marks the lists INCOMPLETE if one fails. The generated records are NOT a real organization's data. Field names (id, created_at, actor.type, actor.email_address, actor.user_id, actor.api_key_id, type) come from Anthropic's Activity object: https://platform.claude.com/docs/en/manage-claude/compliance-activity-feed The activity-type strings and the removal fields come from the API reference: https://platform.claude.com/docs/en/api/compliance/activities/list There, org_user_deleted names the removed user in deleted_user_email and deleted_user_id, and scim_user_deleted carries only user_id. The script reads the email from TARGET_FIELD, and maps the ids to an email through actor.user_id seen elsewhere in the feed. The reference does not say that these ids are the same as actor.user_id; we assume so. Check one of your own removal records. """ import csv import datetime as dt import hashlib import json import pathlib import random import sys import duckdb # ---- CHANGE THESE FOR YOUR OWN DATA --------------------------------------------------------- PERIOD_START = "2026-07-01" # first day of the audit period (UTC, inclusive) PERIOD_END = "2026-09-30" # last day of the audit period (UTC, inclusive) ACTIVITIES_FILE = None # your Activity Feed pull, one JSON object per line; None = generate LEAVERS_FILE = None # your HR leaver list (CSV with a header row); None = generate ADMINS_FILE = None # your list of people allowed to change Claude settings (CSV) MEMBERS_FILE = None # optional: your export of Claude Enterprise members (CSV); None = no seat list LEAVER_EMAIL_COL = "email" LEAVER_DATE_COL = "last_day" # the leaver's last day of work, as YYYY-MM-DD ADMIN_EMAIL_COL = "email" MEMBERS_EMAIL_COL = "email" TARGET_FIELD = "deleted_user_email" # the field on a removal record that holds the removed user's email GRACE_DAYS = 1 # access must end within this many days of the last day of work REMOVAL_WINDOW_DAYS = 30 # a removal before every last day counts if it is at most this many days early # ---------------------------------------------------------------------------------------------- REMOVAL_TYPES = ["org_user_deleted", "scim_user_deleted", "org_user_left"] ADMIN_CHANGE_TYPES = [ "claude_organization_settings_updated", "org_sso_toggled", "org_sso_provisioning_mode_changed", "org_ip_restriction_created", "org_ip_restriction_updated", "org_ip_restriction_deleted", "org_data_residency_updated", "org_compliance_api_settings_updated", "claude_user_role_updated", ] READ_TYPES = ["compliance_api_accessed"] OUT_DIR = pathlib.Path("evidence-out") GEN_DIR = pathlib.Path("generated") def generate(plant): """Seeded stand-in for an organization's feed, written as JSON lines in the Activity object's shape.""" rnd = random.Random(221) GEN_DIR.mkdir(exist_ok=True) users = [f"u{n:02d}@example.test" for n in range(1, 41)] admins, leavers = users[:4], users[29:35] last_day = {u: dt.date(2026, 7, 10) + dt.timedelta(days=11 * i) for i, u in enumerate(leavers)} start, end = dt.date.fromisoformat(PERIOD_START), dt.date.fromisoformat(PERIOD_END) recs = [] def add(day, typ, actor, **extra): t = dt.datetime.combine(day, dt.time(rnd.randrange(7, 19), rnd.randrange(60), rnd.randrange(60))) recs.append({"id": "", "created_at": t.strftime("%Y-%m-%dT%H:%M:%SZ"), "organization_id": "org_generated", "actor": actor, "type": typ, **extra}) def directory(): return {"type": "scim_directory_sync_actor", "directory_id": "directory_generated", "idp_connection_type": "AzureSCIMV2"} def user(email): return {"type": "user_actor", "email_address": email, "user_id": "user_" + email.split("@")[0], "ip_address": "192.0.2.1", "user_agent": "generated"} leaver_kept = leavers[3] # planted: keeps using Claude, never removed leaver_late = leavers[4] # planted: removed 9 days late rehired = leavers[1] # planted: left 07-21 and was removed; rehired, left again 09-25, never removed came_back = leavers[0] # planted: removed in time, then the account acts again 10 days later day = start while day <= end: staff = [u for u in users if u not in last_day or day <= last_day[u]] if plant and u_kept_active(leaver_kept, day, last_day): staff.append(leaver_kept) add(day, "claude_chat_created", user(leaver_kept)) if plant and dt.date(2026, 8, 5) <= day <= dt.date(2026, 9, 20) and day.day % 5 == 0: add(day, "claude_chat_created", user(rehired)) # the second stint add(day, "sso_login_succeeded", user(rnd.choice(staff))) for _ in range(rnd.randrange(3, 8)): add(day, rnd.choice(["claude_chat_created", "claude_chat_viewed"]), user(rnd.choice(staff))) day += dt.timedelta(days=1) for u in leavers: if plant and u == leaver_kept: continue lag = 9 if (plant and u == leaver_late) else rnd.choice([0, 1]) add(last_day[u] + dt.timedelta(days=lag), "org_user_deleted", directory(), **{TARGET_FIELD: u, "deleted_user_id": "user_" + u.split("@")[0]}) for _ in range(12): add(start + dt.timedelta(days=rnd.randrange(92)), rnd.choice(ADMIN_CHANGE_TYPES[:6]), user(rnd.choice(admins))) if plant: add(dt.date(2026, 8, 25), "org_ip_restriction_updated", user(users[11])) # not an admin for key in ["apikey_audit01", "apikey_siem02", "apikey_legal03"]: for _ in range(rnd.randrange(8, 20)): add(start + dt.timedelta(days=rnd.randrange(92)), "compliance_api_accessed", {"type": "api_actor", "api_key_id": key, "ip_address": "198.51.100.7", "user_agent": "generated"}) extra_hr = [] if plant: add(last_day[came_back] + dt.timedelta(days=10), "claude_chat_created", user(came_back)) extra_hr = [(rehired, dt.date(2026, 9, 25)), # the second departure: never removed ("u41@example.test", dt.date(2026, 9, 15)), # planted: an HR leaver who never had Claude ("u42@example.test", dt.date(2026, 9, 20))] # planted: a seat holder, inactive, never removed recs = [r for r in recs if not r["created_at"].startswith("2026-08-19")] # planted: a day with no records recs.sort(key=lambda r: (r["created_at"], r["type"], json.dumps(r["actor"], sort_keys=True))) for n, r in enumerate(recs, 1): r["id"] = f"activity_{n:06d}" if plant: recs.append(dict(recs[40])) # planted: a duplicated record feed = GEN_DIR / ("feed_planted.jsonl" if plant else "feed_clean.jsonl") feed.write_text("".join(json.dumps(r) + "\n" for r in recs)) with open(GEN_DIR / "leavers.csv", "w", newline="") as f: w = csv.writer(f) w.writerow(["email", "last_day"]) w.writerows([[u, last_day[u].isoformat()] for u in leavers] + [[u, d.isoformat()] for u, d in extra_hr]) with open(GEN_DIR / "admins.csv", "w", newline="") as f: w = csv.writer(f) w.writerow(["email"]) w.writerows([[a] for a in admins]) with open(GEN_DIR / "members.csv", "w", newline="") as f: # seat holders at the start of the period w = csv.writer(f) w.writerow(["email"]) w.writerows([[u] for u in users] + ([["u42@example.test"]] if plant else [])) return str(feed), str(GEN_DIR / "leavers.csv"), str(GEN_DIR / "admins.csv"), str(GEN_DIR / "members.csv") def u_kept_active(u, day, last_day): return day > last_day[u] and (day - last_day[u]).days <= 10 def fingerprint(rows): body = "\n".join("|".join("" if v is None else str(v) for v in r) for r in rows) return hashlib.sha256(body.encode()).hexdigest() def population(con, name, control, filters, sql, header, complete): rows = con.execute(sql).fetchall() OUT_DIR.mkdir(exist_ok=True) with open(OUT_DIR / f"{name}.csv", "w", newline="") as f: w = csv.writer(f) w.writerow(header) w.writerows(rows) print(f"\n== {control}: {name}" + ("" if complete else " ** INCOMPLETE: see input checks **")) print(f"period : {PERIOD_START} to {PERIOD_END} (UTC, inclusive)") print(f"filters : {filters}") print(f"run at : {dt.datetime.now(dt.timezone.utc):%Y-%m-%d %H:%M:%S} UTC") print(f"rows : {len(rows)}") print(f"sha256 : {fingerprint(rows)}") return rows def main(): plant, strict = "--plant" in sys.argv, "--strict" in sys.argv acts, lvr, adm, mem = ACTIVITIES_FILE, LEAVERS_FILE, ADMINS_FILE, MEMBERS_FILE if acts is None: acts, lvr, adm, generated_members = generate(plant) mem = generated_members if "--members" in sys.argv else None print(f"input : GENERATED records (not a real organization), seed 221, plant={plant}") else: print("input : " + acts) print("members : " + (mem if mem else "none (MEMBERS_FILE not set)")) con = duckdb.connect() con.execute(f""" CREATE TABLE raw AS SELECT * FROM read_json('{acts}', format='newline_delimited', columns={{ 'id':'VARCHAR', 'created_at':'VARCHAR', 'type':'VARCHAR', 'actor':'STRUCT(type VARCHAR, email_address VARCHAR, user_id VARCHAR, api_key_id VARCHAR)', '{TARGET_FIELD}':'VARCHAR', 'deleted_user_id':'VARCHAR', 'user_id':'VARCHAR'}})""") con.execute(f""" CREATE TABLE activities AS SELECT id, type, try_cast(replace(created_at, 'Z', '') AS TIMESTAMP) AS ts, actor.type AS actor_type, lower(trim(actor.email_address)) AS actor_email, actor.user_id AS actor_user_id, actor.api_key_id AS api_key_id, lower(trim("{TARGET_FIELD}")) AS target_email, coalesce(deleted_user_id, user_id) AS target_user_id FROM raw""") con.execute(f"""CREATE TABLE leavers AS SELECT lower(trim("{LEAVER_EMAIL_COL}")) AS email, try_cast("{LEAVER_DATE_COL}" AS DATE) AS last_day FROM read_csv('{lvr}', header=true, all_varchar=true)""") con.execute(f"""CREATE TABLE members AS SELECT lower(trim("{MEMBERS_EMAIL_COL}")) AS email FROM read_csv('{mem}', header=true, all_varchar=true)""" if mem else "CREATE TABLE members (email VARCHAR)") con.execute(f"""CREATE TABLE admins AS SELECT lower(trim("{ADMIN_EMAIL_COL}")) AS email FROM read_csv('{adm}', header=true, all_varchar=true)""") p0, p1 = PERIOD_START, PERIOD_END rt = ", ".join(f"'{t}'" for t in REMOVAL_TYPES) # who each removal record names: the email field, else the id mapped through actor.user_id, else # (org_user_left) the person who removed themselves con.execute(f""" CREATE TABLE removals AS WITH ids AS (SELECT actor_user_id, arg_max(actor_email, ts) AS email FROM activities WHERE actor_type = 'user_actor' AND actor_user_id IS NOT NULL AND actor_email IS NOT NULL GROUP BY 1) SELECT a.id, a.ts, a.type, coalesce(a.target_email, i.email, CASE WHEN a.type = 'org_user_left' THEN a.actor_email END) AS email FROM activities a LEFT JOIN ids i ON i.actor_user_id = a.target_user_id WHERE a.type IN ({rt})""") # ---- input checks: each says which planted fault trips it total = con.execute("SELECT count(*) FROM activities").fetchone()[0] dup = con.execute("SELECT count(*) FROM (SELECT id FROM activities GROUP BY id HAVING count(*) > 1)").fetchone()[0] no_time = con.execute("SELECT count(*) FROM activities WHERE ts IS NULL OR id IS NULL").fetchone()[0] outside = con.execute(f"SELECT count(*) FROM activities WHERE ts::DATE NOT BETWEEN '{p0}' AND '{p1}'").fetchone()[0] gaps = [str(r[0]) for r in con.execute(f""" SELECT d::DATE FROM generate_series(DATE '{p0}', DATE '{p1}', INTERVAL 1 DAY) t(d) WHERE d::DATE NOT IN (SELECT ts::DATE FROM activities WHERE ts IS NOT NULL) ORDER BY 1""").fetchall()] hr_bad = con.execute("SELECT count(*) FROM leavers WHERE email IS NULL OR email = '' OR last_day IS NULL").fetchone()[0] adm_bad = con.execute("SELECT count(*) FROM admins WHERE email IS NULL OR email = ''").fetchone()[0] checks = [ ("duplicate activity ids", dup, "a duplicated record"), ("records with no id or no readable created_at", no_time, "a missing timestamp"), ("days in the period with no records", len(gaps), "a day with no records"), ("HR leaver rows with no email or no last day", hr_bad, "a NULL in the HR file"), ("admin roster rows with no email", adm_bad, "a NULL in the roster"), ] print(f"\n== Input checks (records read: {total}; outside the period, left out: {outside})") ok = True for label, n, fault in checks: flag = "ok " if n == 0 else "FAIL" ok &= n == 0 print(f"[{flag}] {label}: {n}" + ("" if n == 0 else f" (trips on: {fault})")) if gaps: print(" empty days: " + ", ".join(gaps)) seen = {r[0] for r in con.execute("SELECT DISTINCT type FROM activities").fetchall()} unseen = [t for t in ADMIN_CHANGE_TYPES + REMOVAL_TYPES + READ_TYPES if t not in seen] print("[info] configured activity types never seen in this feed: " + (", ".join(unseen) or "none")) listed = set(ADMIN_CHANGE_TYPES + REMOVAL_TYPES + READ_TYPES) other = [f"{t} {n}" for t, n in con.execute( "SELECT type, count(*) FROM activities WHERE type IS NOT NULL GROUP BY 1 ORDER BY 1").fetchall() if t not in listed] print("[info] activity types in the feed that no list uses: " + (", ".join(other) or "none")) unnamed = con.execute("SELECT count(*) FROM removals WHERE email IS NULL").fetchone()[0] print(f"[info] removal records that name no email the script can match: {unnamed}") # ---- CC6.2: leavers. Account evidence = seen in the feed by the end of the period, or in MEMBERS_FILE. con.execute(f""" CREATE TABLE had_account AS SELECT actor_email AS email FROM activities WHERE actor_type = 'user_actor' AND actor_email IS NOT NULL AND ts::DATE <= DATE '{p1}' UNION SELECT email FROM removals WHERE email IS NOT NULL UNION SELECT email FROM members WHERE email IS NOT NULL AND email <> ''""") in_period = f"l.last_day BETWEEN DATE '{p0}' AND DATE '{p1}' AND l.email IS NOT NULL" if mem: left_out = con.execute(f"""SELECT l.email, l.last_day FROM leavers l WHERE {in_period} AND l.email NOT IN (SELECT email FROM had_account) ORDER BY l.last_day, l.email""").fetchall() OUT_DIR.mkdir(exist_ok=True) with open(OUT_DIR / "cc6_2_not_members_not_seen.csv", "w", newline="") as f: csv.writer(f).writerows([["email", "last_day"], *left_out]) print(f"\n[info] HR leavers in the period not in MEMBERS_FILE and not seen in the feed, left out of CC6.2: " f"{len(left_out)}" + (" (listed in evidence-out/cc6_2_not_members_not_seen.csv)" if left_out else "")) population_filter = "AND l.email IN (SELECT email FROM had_account)" else: print("\n[info] no MEMBERS_FILE: a leaver with no Claude activity seen stays in CC6.2 as " "CHECK: NO CLAUDE ACTIVITY SEEN; clear those rows against your seat list") population_filter = "" # Each removal record is credited to ONE departure of that person: the latest HR last day on or # before it, or, if it came before every last day, the earliest last day it precedes by at most # REMOVAL_WINDOW_DAYS. So a removal from an earlier stint never counts for a later departure. sql = f""" WITH credited AS ( SELECT r.email, r.ts, coalesce( (SELECT max(l.last_day) FROM leavers l WHERE l.email = r.email AND l.last_day <= r.ts::DATE), (SELECT min(l.last_day) FROM leavers l WHERE l.email = r.email AND l.last_day - INTERVAL {REMOVAL_WINDOW_DAYS} DAY <= r.ts::DATE)) AS last_day FROM removals r WHERE r.email IS NOT NULL), rem AS (SELECT email, last_day, min(ts) AS removed_at FROM credited WHERE last_day IS NOT NULL GROUP BY 1, 2), nxt AS (SELECT email, last_day, lead(last_day) OVER (PARTITION BY email ORDER BY last_day) AS next_last_day FROM leavers WHERE email IS NOT NULL), -- after the grace period and before the removal; and after the removal. When the same person -- has a LATER departure in HR, use after the removal belongs to that later stint, not this row. after AS (SELECT l.email, l.last_day, count(a.id) FILTER (WHERE a.ts::DATE > l.last_day + INTERVAL {GRACE_DAYS} DAY AND (r.removed_at IS NULL OR a.ts < r.removed_at)) AS n_grace, count(a.id) FILTER (WHERE r.removed_at IS NOT NULL AND a.ts > r.removed_at AND n.next_last_day IS NULL) AS n_removed FROM leavers l JOIN nxt n ON n.email = l.email AND n.last_day = l.last_day LEFT JOIN rem r ON r.email = l.email AND r.last_day = l.last_day LEFT JOIN activities a ON a.actor_email = l.email AND a.actor_type = 'user_actor' GROUP BY 1, 2) SELECT l.email, l.last_day, r.removed_at::DATE AS first_removal_record, date_diff('day', l.last_day, r.removed_at::DATE) AS days_to_removal, af.n_grace AS actions_after_grace, af.n_removed AS actions_after_removal, CASE WHEN r.removed_at IS NULL AND l.email IN (SELECT email FROM had_account) THEN 'NO REMOVAL RECORD' WHEN r.removed_at IS NULL THEN 'CHECK: NO CLAUDE ACTIVITY SEEN' WHEN af.n_removed > 0 THEN 'ACTIVE AFTER REMOVAL' WHEN date_diff('day', l.last_day, r.removed_at::DATE) > {GRACE_DAYS} THEN 'REMOVED LATE' ELSE 'removed in time' END AS status FROM leavers l LEFT JOIN rem r ON r.email = l.email AND r.last_day = l.last_day LEFT JOIN after af ON af.email = l.email AND af.last_day = l.last_day WHERE {in_period} {population_filter} ORDER BY l.last_day, l.email""" r62 = population(con, "cc6_2_leavers", "CC6.2", ("HR leavers with last day in period, in MEMBERS_FILE or seen in the feed" if mem else "HR leavers with last day in period") + f"; removal types {REMOVAL_TYPES}, each credited to one departure; grace {GRACE_DAYS} day(s)", sql, ["email", "last_day", "first_removal_record", "days_to_removal", "actions_after_grace", "actions_after_removal", "status"], ok) for r in r62: print(" " + " | ".join("-" if v is None else str(v) for v in r)) # ---- CC8.1: administrative changes ct = ", ".join(f"'{t}'" for t in ADMIN_CHANGE_TYPES) sql = f""" SELECT a.ts, a.type, a.actor_type, coalesce(a.actor_email, '-') AS actor, CASE WHEN a.actor_type <> 'user_actor' THEN 'not a person' WHEN a.actor_email IN (SELECT email FROM admins WHERE email IS NOT NULL) THEN 'on roster' ELSE 'NOT ON ROSTER' END AS roster FROM activities a WHERE a.type IN ({ct}) AND a.ts::DATE BETWEEN DATE '{p0}' AND DATE '{p1}' ORDER BY a.ts, a.id""" r81 = population(con, "cc8_1_admin_changes", "CC8.1", f"types {len(ADMIN_CHANGE_TYPES)} listed in ADMIN_CHANGE_TYPES", sql, ["created_at", "type", "actor_type", "actor", "roster"], ok) for r in r81: if r[4] == "NOT ON ROSTER": print(" " + " | ".join(str(v) for v in r)) # ---- CC7.2: Compliance API requests, by key sql = f""" SELECT coalesce(api_key_id, '-') AS api_key_id, count(*) AS requests, min(ts::DATE) AS first_day, max(ts::DATE) AS last_day FROM activities WHERE type IN ('compliance_api_accessed') AND ts::DATE BETWEEN DATE '{p0}' AND DATE '{p1}' GROUP BY 1 ORDER BY 1""" r72 = population(con, "cc7_2_compliance_api_requests", "CC7.2", "type compliance_api_accessed; grouped by actor.api_key_id", sql, ["api_key_id", "requests", "first_day", "last_day"], ok) for r in r72: print(" " + " | ".join(str(v) for v in r)) follow = sum(1 for r in r62 if r[6] != "removed in time" or r[4] or r[5]) + sum(1 for r in r81 if r[4] == "NOT ON ROSTER") print("\n== Result") print(f"input checks : {'all passed' if ok else 'FAILED, the populations above are not complete'}") print(f"follow-up : {follow} leaver or admin-change row(s) need an explanation") verdict = "NOTHING TO FOLLOW UP" if ok and follow == 0 else "EXCEPTIONS FOUND" print(f"RESULT: {verdict}") return 1 if strict and verdict != "NOTHING TO FOLLOW UP" else 0 if __name__ == "__main__": sys.exit(main())