#!/usr/bin/env python3 """List every sensitive table whose owner is gone, missing or nobody, from your own exports, in DuckDB. python3 sensitive_table_owners.py --demo # clean generated records python3 sensitive_table_owners.py --demo --plant-error # generated records with one table per finding python3 sensitive_table_owners.py --demo --plant-input # generated records with planted input faults python3 sensitive_table_owners.py --demo --plant-header # generated records with one column renamed in members.csv python3 sensitive_table_owners.py # your exports: set DATA_DIR and the settings below first python3 sensitive_table_owners.py --data # your exports in (instead of editing DATA_DIR) python3 sensitive_table_owners.py --strict # exit 1 if any table has a finding add --out to write every file under that folder (default: ./owner-check-out) It reads five CSV files (see COLUMN_MAP): the owner of every table, the list of sensitive tables, the people and groups that can own a table, the members of each group or role, and your HR file. For each sensitive table it prints: NO_OWNER no owner recorded, or an owner your principals.csv does not list, or a listed person who has no HR record or was not employed on AS_OF OWNER_LEFT the owner is a person whose HR end date is on or before AS_OF (days since they left shown) GROUP_NOBODY_CURRENT the owner is a role or group (Snowflake role, Databricks or Redshift group) with no member who still works here, or with no members at all (roles and groups inside it are followed, except Snowflake's system roles in SKIP_ROLES, whose users are administrators) SERVICE_ACCOUNT_NO_PERSON the owner is a service account or service principal with no named person, or whose named person is not employed on AS_OF Each departure is read separately, so a person who left and was hired back is current. A person counts as current on AS_OF when one HR row has a start date on or before AS_OF and no end date or an end date after AS_OF. Input checks run first. If one fails, no findings are computed and the run says INCOMPLETE. The script reads files; it never connects to a platform and never writes to one. The demo records are GENERATED (fixed, no random numbers): not a company's data, and not a run on Snowflake, Databricks or Redshift. Their column names are the platforms' own: Snowflake https://docs.snowflake.com/en/sql-reference/info-schema/tables (TABLE_OWNER, the role that owns the table) https://docs.snowflake.com/en/sql-reference/sql/show-grants (SHOW GRANTS OF ROLE) Databricks https://docs.databricks.com/aws/en/sql/language-manual/information-schema/tables (table_owner) Redshift https://docs.aws.amazon.com/redshift/latest/dg/r_SVV_REDSHIFT_TABLES.html (table_owner) """ import csv import datetime as dt import hashlib import pathlib import re import sys import duckdb # ---- CHANGE THESE FOR YOUR OWN DATA --------------------------------------------------------- AS_OF = "2026-11-05" # the day you took the exports (YYYY-MM-DD); an HR end date on or before it means "left" DATA_DIR = None # folder holding the five files named in FILE_NAMES PLATFORMS = ["snowflake", "databricks", "redshift"] # the platforms you exported; one with no owner rows is NOT CHECKED FILE_NAMES = {"owners": "owners.csv", "sensitive": "sensitive.csv", "principals": "principals.csv", "members": "members.csv", "hr": "hr.csv"} # Snowflake's system roles. Snowflake advises granting every custom role to SYSADMIN, so SHOW GRANTS OF ROLE lists # SYSADMIN as a role the owner role is granted to. Its users are administrators, not the people behind the table, so # membership is never followed from another role into these roles: a role whose own users have all left is still # GROUP_NOBODY_CURRENT. A table owned by a system role itself is followed as usual (SYSADMIN to ACCOUNTADMIN to users). # Remove a name only if that role really is how your owners hold the table. SKIP_ROLES = {"snowflake": ["ACCOUNTADMIN", "ORGADMIN", "SECURITYADMIN", "SYSADMIN", "USERADMIN"]} # For each file: the script's column name (left, do not change) = the column name in your file (right). # Names are matched without regard to case or surrounding spaces. COLUMN_MAP = { "owners": {"platform": "platform", "table_catalog": "table_catalog", "table_schema": "table_schema", "table_name": "table_name", "table_owner": "table_owner"}, "sensitive": {"platform": "platform", "table_catalog": "table_catalog", "table_schema": "table_schema", "table_name": "table_name", "data_area": "data_area"}, "principals": {"platform": "platform", "principal": "principal", "kind": "kind", "person_email": "person_email"}, "members": {"platform": "platform", "group_name": "group_name", "member": "member"}, "hr": {"email": "email", "start_date": "start_date", "end_date": "end_date"}, } # ---------------------------------------------------------------------------------------------- KNOWN_PLATFORMS = ["snowflake", "databricks", "redshift"] KNOWN_KINDS = ["person", "group", "role", "service_account"] GROUP_KINDS = ["group", "role"] CODES = ["NO_OWNER", "OWNER_LEFT", "GROUP_NOBODY_CURRENT", "SERVICE_ACCOUNT_NO_PERSON"] RUN_AT = dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") def dcol(col): """A date read strictly: YYYY-MM-DD and a real calendar day; anything else is NULL and counted by an input check.""" return (f"CASE WHEN regexp_full_match(trim({col}), '[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]') " f"THEN try_strptime(trim({col}), '%Y-%m-%d')::DATE END") def blank(col): return f"nullif(trim({col}), '')" def norm(col): return f"lower(trim({col}))" def quote_list(items): return ", ".join("'" + str(i).replace("'", "''") + "'" for i in items) def cell(v): """A CSV cell a spreadsheet will not run: text starting with = + - @, a tab or a carriage return gets a ' first.""" return "'" + v if isinstance(v, str) and v[:1] in ("=", "+", "-", "@", "\t", "\r") else v # --------------------------------------------------------------------------------------------- # Demo records: GENERATED, in the column shapes the three platforms document. # --------------------------------------------------------------------------------------------- def generate(folder, plant_error, plant_input, plant_header): folder.mkdir(parents=True, exist_ok=True) first = ["ada", "ben", "chloe", "dev", "elena", "farid", "gwen", "hugo", "iris", "jonas", "kira", "luca"] email = {n: f"{n}.reyes@example.com" if i % 2 else f"{n}.novak@example.com" for i, n in enumerate(first)} hr = [[email[n], "2021-03-01", ""] for n in first] def hr_row(n, start, end, replace=True): if replace: hr[:] = [r for r in hr if r[0] != email[n]] hr.append([email[n], start, end]) hr_row("hugo", "2020-03-02", "2026-08-05") # left hr_row("jonas", "2022-01-10", "2026-09-18") # left hr_row("luca", "2023-02-06", "2026-07-10") # left hr_row("iris", "2019-05-01", "2024-12-31") # left, then hired back: current hr_row("iris", "2025-06-02", "", replace=False) hr_row("kira", "2026-10-01", "") # joined recently sf = lambda n: n.upper() # Snowflake logins are upper case login = {"snowflake": sf, "databricks": lambda n: email[n], "redshift": lambda n: n} principals = [[p, login[p](n), "person", email[n]] for p in KNOWN_PLATFORMS for n in first] principals += [["snowflake", r, "role", ""] for r in ["FINANCE_OWNER", "CUSTOMER_OWNER", "HR_OWNER", "PEOPLE_TEAM", "OLD_PROJECT_ROLE", "SYSADMIN", "ACCOUNTADMIN"]] principals += [["databricks", g, "group", ""] for g in ["finance-stewards", "data-stewards"]] principals += [["databricks", "sp-etl-nightly", "service_account", email["ada"]]] members = [["snowflake", "FINANCE_OWNER", u] for u in ("ADA", "BEN", "HUGO")] members += [["snowflake", "CUSTOMER_OWNER", u] for u in ("CHLOE", "DEV", "JONAS")] members += [["snowflake", "HR_OWNER", "PEOPLE_TEAM"], ["snowflake", "PEOPLE_TEAM", "FARID"], ["snowflake", "PEOPLE_TEAM", "JONAS"], ["snowflake", "OLD_PROJECT_ROLE", "LUCA"], ["snowflake", "SYSADMIN", "ACCOUNTADMIN"], ["snowflake", "ACCOUNTADMIN", "ADA"]] # an admin through a chain members += [["databricks", "finance-stewards", email["ben"]], ["databricks", "finance-stewards", email["elena"]], ["databricks", "data-stewards", email["iris"]], ["databricks", "data-stewards", email["farid"]]] sens = [ # platform, catalog, schema, table, data area, owner ["snowflake", "PROD_DB", "FINANCE", "GL_BALANCES", "finance", "FINANCE_OWNER"], ["snowflake", "PROD_DB", "FINANCE", "AP_INVOICES", "finance", "SYSADMIN"], # owned by a system role ["snowflake", "PROD_DB", "CUSTOMER", "ACCOUNTS", "customer", "CUSTOMER_OWNER"], ["snowflake", "PROD_DB", "CUSTOMER", "CONTACTS", "customer", "CUSTOMER_OWNER"], ["snowflake", "PROD_DB", "HR", "EMPLOYEES", "hr", "HR_OWNER"], ["databricks", "main", "finance", "gl_balances", "finance", email["ada"]], ["databricks", "main", "finance", "revenue_recognition", "finance", "finance-stewards"], ["databricks", "main", "marketing", "leads", "customer", email["dev"]], ["databricks", "main", "marketing", "consents", "customer", "sp-etl-nightly"], ["databricks", "main", "hr", "performance_reviews", "hr", "data-stewards"], ["redshift", "dev", "finance", "ledger_entries", "finance", "ben"], ["redshift", "dev", "customer", "orders_pii", "customer", "iris"], ["redshift", "dev", "hr", "salaries", "hr", "farid"], ["redshift", "dev", "hr", "benefits", "hr", "elena"]] other = [ # tables nobody listed as sensitive: owned by people who left, and out of scope ["snowflake", "PROD_DB", "PUBLIC", "CALENDAR", "OLD_PROJECT_ROLE"], ["databricks", "main", "sandbox", "scratch", email["luca"]], ["redshift", "dev", "public", "etl_scratch", "luca"]] if plant_error: def table(name): hit = [r for r in sens if r[3] == name] if len(hit) != 1: sys.exit(f"Could not plant a fault: table {name} matched {len(hit)} rows.") return hit[0] if any(r[1] in ("contractor-tmp", "LEGACY_OWNER") for r in principals) or any(r[5] == "" for r in sens): sys.exit("A planted value already occurs in the clean records: stopping.") table("gl_balances")[5] = "" # databricks: no owner recorded table("leads")[5] = "contractor-tmp" # databricks: owner not listed anywhere table("salaries")[5] = "hugo" # redshift: owner left 2026-08-05 table("ACCOUNTS")[5] = "LEGACY_OWNER" # snowflake: a role holding only a leaver principals.append(["snowflake", "LEGACY_OWNER", "role", ""]) members.append(["snowflake", "LEGACY_OWNER", "JONAS"]) members.append(["snowflake", "LEGACY_OWNER", "SYSADMIN"]) # ...and granted to SYSADMIN, as Snowflake # advises: not followed before = len(members) members[:] = [m for m in members if m[1] != "data-stewards"] # databricks: a group with nobody in it if before - len(members) != 2: sys.exit("Could not plant a fault: data-stewards did not have two members.") for p in principals: if p[1] == "sp-etl-nightly": p[3] = "" # databricks: service principal, no named person owners = [r[:4] + [r[5]] for r in sens] + [r[:4] + [r[4]] for r in other] sensitive = [r[:5] for r in sens] if plant_input: hr[[r[0] for r in hr].index(email["kira"])][1] = "01/10/26" # not a date hr.append([email["hugo"], "2026-09-01", "2026-08-01"]) # ends before it starts sensitive.append(["snowflake", "PROD_DB", "FINANCE", "NOT_EXPORTED", "finance"]) # not in the owner export sensitive.append(list(sensitive[0])) # the same sensitive table twice sensitive.append(["redshift", "dev", "finance", "blank_area", ""]) # no data area... owners.append(["redshift", "dev", "finance", "blank_area", "ben"]) # ...on a table the export holds members.append(["snowflake", "GHOST_ROLE", "ADA"]) # a group principals.csv does not list members.append(["snowflake", "CUSTOMER_OWNER", "ZELDA"]) # a member principals.csv does not list principals.append(["snowflake", "finance_owner", "role", ""]) # FINANCE_OWNER again, other case principals.append(["databricks", "tiger-team", "team", ""]) # a kind the script does not know owners.append(["oracle", "x", "y", "z", "w"]) # a platform the script does not know owners.append(list(owners[0])) # the same owner row twice header_fault = plant_header # members.csv: member -> login tables = {"owners": owners, "sensitive": sensitive, "principals": principals, "members": members, "hr": hr} for name, rows in tables.items(): cols = list(COLUMN_MAP[name]) with open(folder / FILE_NAMES[name], "w", newline="") as f: w = csv.writer(f) w.writerow(cols) w.writerows(rows) if header_fault: p = folder / FILE_NAMES["members"] p.write_text(p.read_text().replace("group_name,member", "group_name,login", 1)) # --------------------------------------------------------------------------------------------- # ENGINE # --------------------------------------------------------------------------------------------- def main(): args = sys.argv[1:] global DATA_DIR out = pathlib.Path(args[args.index("--out") + 1]) if "--out" in args else pathlib.Path("owner-check-out") if "--data" in args: DATA_DIR = args[args.index("--data") + 1] plant_error, plant_input = "--plant-error" in args, "--plant-input" in args plant_header = "--plant-header" in args suffix = ("-planted-error" if plant_error else "-planted-input" if plant_input else "-planted-header" if plant_header else "") if "--demo" in args: gen = out / f"generated{suffix}" generate(gen, plant_error, plant_input, plant_header) DATA_DIR = str(gen) print(f"Generated records written to {gen}/ (not a company's data).") if DATA_DIR is None: print("Set DATA_DIR at the top of the script, pass --data (or run with --demo).") return 2 folder = pathlib.Path(DATA_DIR) out.mkdir(parents=True, exist_ok=True) problems_found = [] def check(name, bad, detail=""): print(f" [{'FAIL' if bad else 'ok'}] {name}" + (f": {detail}" if bad and detail else "")) if bad: problems_found.append(name) con = duckdb.connect() print(f"\nRun at {RUN_AT}. Exports taken {AS_OF}.") print("\nInput checks") bad_date = not (isinstance(AS_OF, str) and re.fullmatch(r"\d{4}-\d{2}-\d{2}", AS_OF)) if not bad_date: try: dt.date.fromisoformat(AS_OF) except ValueError: bad_date = True check("AS_OF is a real date written YYYY-MM-DD", bad_date, f"AS_OF is {AS_OF!r}") bad_plat = not isinstance(PLATFORMS, list) or not PLATFORMS or [p for p in PLATFORMS if p not in KNOWN_PLATFORMS] check("PLATFORMS lists the platforms you exported, from snowflake, databricks, redshift", bad_plat, f"PLATFORMS is {PLATFORMS!r}") if bad_date or bad_plat: return 2 # --- load every file; a missing file or column is a failure, the table is then empty bad_files = [] for name, cmap in COLUMN_MAP.items(): path = folder / FILE_NAMES[name] have = [] if path.is_file(): have = [d[0] for d in con.execute("SELECT * FROM read_csv(?, header=true, all_varchar=true) LIMIT 0", [str(path)]).description] lowered = {h.strip().lower(): h for h in have} missing = [theirs for theirs in cmap.values() if theirs.strip().lower() not in lowered] if not path.is_file(): bad_files.append(f"{FILE_NAMES[name]} is missing") elif missing: bad_files.append(f"{FILE_NAMES[name]} has no column {', '.join(missing)}") if path.is_file() and not missing: sel = ", ".join(f'"{lowered[theirs.strip().lower()]}" AS "{mine}"' for mine, theirs in cmap.items()) con.execute(f"CREATE TABLE t_{name} AS SELECT {sel} FROM read_csv(?, header=true, all_varchar=true)", [str(path)]) else: con.execute(f"CREATE TABLE t_{name} ({', '.join(chr(34) + c + chr(34) + ' VARCHAR' for c in cmap)})") check("every file is there with every column the script reads", bad_files, "; ".join(bad_files)) counts = {n: con.execute(f"SELECT count(*) FROM t_{n}").fetchone()[0] for n in COLUMN_MAP} for n, c in counts.items(): print(f" read {c} rows from {FILE_NAMES[n]}") q = lambda sql: con.execute(sql).fetchone()[0] # --- unreadable and unknown values are counted, never skipped n = q(f"""SELECT count(*) FROM t_hr WHERE ({dcol('start_date')}) IS NULL OR ({blank('end_date')} IS NOT NULL AND ({dcol('end_date')}) IS NULL)""") check("every HR start and end date is a real date written YYYY-MM-DD (a start date is required)", n, f"{n} HR row(s) with a missing start date or a date not written YYYY-MM-DD") n = q(f"SELECT count(*) FROM t_hr WHERE ({dcol('end_date')}) < ({dcol('start_date')})") check("no HR row ends before it starts", n, f"{n} HR row(s) with an end date before the start date") odd = [] for tbl in ("owners", "sensitive", "principals", "members"): k = q(f"SELECT count(*) FROM t_{tbl} WHERE {norm('platform')} NOT IN ({quote_list(PLATFORMS)})") if k: odd.append(f"{FILE_NAMES[tbl]}: {k}") check("every platform value is one of PLATFORMS", odd, "unlisted platforms: " + ", ".join(odd)) n = q(f"SELECT count(*) FROM t_principals WHERE {norm('kind')} NOT IN ({quote_list(KNOWN_KINDS)})") check("every principal kind is person, group, role or service_account", n, f"{n} principal row(s) with another kind") key_cols = {"owners": ["platform", "table_catalog", "table_schema", "table_name"], "sensitive": ["platform", "table_catalog", "table_schema", "table_name", "data_area"], "principals": ["platform", "principal", "kind"], "members": ["platform", "group_name", "member"]} n = sum(q(f"SELECT count(*) FROM t_{t} WHERE " + " OR ".join(f"{blank(c)} IS NULL" for c in cols)) for t, cols in key_cols.items()) check("no key column is blank (platform, catalog, schema, table, data area, principal, kind, group, member)", n, f"{n} row(s) with a blank key column") tk = "lower(trim(platform)), lower(trim(table_catalog)), lower(trim(table_schema)), lower(trim(table_name))" n = q(f"SELECT count(*) - count(DISTINCT ({tk})) FROM t_owners") check("every table is in the owner export once", n, f"{n} owner row(s) repeat a table") n = q(f"SELECT count(*) - count(DISTINCT ({tk})) FROM t_sensitive") check("every sensitive table is listed once", n, f"{n} sensitive row(s) repeat a table") n = q(f"""SELECT count(*) FROM t_sensitive s WHERE NOT EXISTS (SELECT 1 FROM t_owners o WHERE {norm('o.platform')} = {norm('s.platform')} AND {norm('o.table_catalog')} = {norm('s.table_catalog')} AND {norm('o.table_schema')} = {norm('s.table_schema')} AND {norm('o.table_name')} = {norm('s.table_name')})""") check("every sensitive table is in the owner export", n, f"{n} sensitive table(s) have no row in {FILE_NAMES['owners']}: a stale list, or the export cannot see them") n = q(f"SELECT count(*) - count(DISTINCT ({norm('platform')}, {norm('principal')})) FROM t_principals") check("each principal is listed once per platform (names compared without regard to case)", n, f"{n} principal row(s) repeat a name") n = q(f"""SELECT count(*) FROM t_members m WHERE NOT EXISTS (SELECT 1 FROM t_principals p WHERE {norm('p.platform')} = {norm('m.platform')} AND {norm('p.principal')} = {norm('m.group_name')} AND {norm('p.kind')} IN ({quote_list(GROUP_KINDS)}))""") check("every membership group is a group or role listed in principals", n, f"{n} membership row(s) name a group that principals.csv does not list as a group or role") n = q(f"""SELECT count(*) FROM t_members m WHERE NOT EXISTS (SELECT 1 FROM t_principals p WHERE {norm('p.platform')} = {norm('m.platform')} AND {norm('p.principal')} = {norm('m.member')})""") check("every member is listed in principals", n, f"{n} membership row(s) name a member that principals.csv does not list") if problems_found: print(f"\nINCOMPLETE: {len(problems_found)} input check(s) failed, so no findings are computed. Fix the input.") return 2 # --- who is a person who still works here asof = f"DATE '{AS_OF}'" con.execute(f"""CREATE TEMP TABLE person AS SELECT {norm('email')} AS email, coalesce(bool_or(({dcol('start_date')}) <= {asof} AND (({dcol('end_date')}) IS NULL OR ({dcol('end_date')}) > {asof})), false) AS is_current, max(({dcol('end_date')})) FILTER (WHERE ({dcol('end_date')}) <= {asof}) AS last_left FROM t_hr GROUP BY 1""") con.execute(f"""CREATE TEMP TABLE pr AS SELECT {norm('platform')} AS platform, {norm('principal')} AS principal, trim(principal) AS shown, {norm('kind')} AS kind, lower({blank('person_email')}) AS email FROM t_principals""") skip = [(p.lower(), r.lower()) for p, roles in SKIP_ROLES.items() for r in roles] or [("", "")] skip_sql = ", ".join(f"('{p}', '{r}')" for p, r in skip) con.execute(f"""CREATE TEMP TABLE mem AS SELECT {norm('platform')} AS platform, {norm('group_name')} AS grp, {norm('member')} AS member FROM t_members WHERE ({norm('platform')}, {norm('member')}) NOT IN (SELECT * FROM (VALUES {skip_sql})) OR ({norm('platform')}, {norm('group_name')}) IN (SELECT * FROM (VALUES {skip_sql}))""") # people inside each group or role, following groups inside groups (at most 10 levels, so a loop ends) con.execute(f"""CREATE TEMP TABLE leaf AS WITH RECURSIVE x(platform, root, node, depth) AS ( SELECT platform, grp, member, 1 FROM mem UNION SELECT x.platform, x.root, m.member, x.depth + 1 FROM x JOIN pr p ON p.platform = x.platform AND p.principal = x.node AND p.kind IN ({quote_list(GROUP_KINDS)}) JOIN mem m ON m.platform = x.platform AND m.grp = x.node WHERE x.depth < 10) SELECT DISTINCT x.platform, x.root, p.principal AS person_principal, p.email FROM x JOIN pr p ON p.platform = x.platform AND p.principal = x.node AND p.kind = 'person'""") con.execute(f"""CREATE TEMP TABLE gstat AS SELECT g.platform, g.principal, count(l.person_principal) AS n_people, count(l.person_principal) FILTER (WHERE coalesce(pe.is_current, false)) AS n_current, max(pe.last_left) FILTER (WHERE NOT coalesce(pe.is_current, false)) AS last_left FROM pr g LEFT JOIN leaf l ON l.platform = g.platform AND l.root = g.principal LEFT JOIN person pe ON pe.email = l.email WHERE g.kind IN ({quote_list(GROUP_KINDS)}) GROUP BY 1, 2""") con.execute(f"""CREATE TEMP TABLE base AS SELECT row_number() OVER (ORDER BY {norm('s.platform')}, trim(s.data_area), trim(s.table_catalog), trim(s.table_schema), trim(s.table_name)) AS no, {norm('s.platform')} AS platform, trim(s.table_catalog) AS cat, trim(s.table_schema) AS sch, trim(s.table_name) AS tbl, trim(s.data_area) AS area, {blank('o.table_owner')} AS owner, lower({blank('o.table_owner')}) AS owner_l FROM t_sensitive s JOIN t_owners o ON {norm('o.platform')} = {norm('s.platform')} AND {norm('o.table_catalog')} = {norm('s.table_catalog')} AND {norm('o.table_schema')} = {norm('s.table_schema')} AND {norm('o.table_name')} = {norm('s.table_name')}""") n_base, n_sens = q("SELECT count(*) FROM base"), counts["sensitive"] if n_base != n_sens: # a guard on the script itself: every sensitive row joined to exactly one owner row sys.exit(f"Internal error: {n_sens} sensitive tables read but {n_base} matched an owner row.") con.execute("""CREATE TEMP TABLE b2 AS SELECT b.*, p.kind, p.email, p.shown, pe.email IS NOT NULL AS p_hr, coalesce(pe.is_current, false) AS p_cur, pe.last_left AS p_left, gs.n_people, gs.n_current, gs.last_left AS g_left, np.email IS NOT NULL AS n_hr, coalesce(np.is_current, false) AS n_cur, np.last_left AS n_left FROM base b LEFT JOIN pr p ON p.platform = b.platform AND p.principal = b.owner_l LEFT JOIN person pe ON pe.email = p.email AND p.kind = 'person' LEFT JOIN gstat gs ON gs.platform = b.platform AND gs.principal = b.owner_l LEFT JOIN person np ON np.email = p.email AND p.kind = 'service_account'""") con.execute(f"""CREATE TEMP TABLE found AS SELECT no, 'NO_OWNER' AS code, 'no owner recorded' AS detail FROM b2 WHERE owner IS NULL UNION ALL SELECT no, 'NO_OWNER', 'owner ' || owner || ' is not listed in principals.csv' FROM b2 WHERE owner IS NOT NULL AND kind IS NULL UNION ALL SELECT no, 'NO_OWNER', 'person ' || owner || ' has no HR email in principals.csv' FROM b2 WHERE kind = 'person' AND email IS NULL UNION ALL SELECT no, 'NO_OWNER', 'person ' || owner || ' has no HR record' FROM b2 WHERE kind = 'person' AND email IS NOT NULL AND NOT p_hr UNION ALL SELECT no, 'NO_OWNER', 'person ' || owner || ' was not employed on {AS_OF}' FROM b2 WHERE kind = 'person' AND p_hr AND NOT p_cur AND p_left IS NULL UNION ALL SELECT no, 'OWNER_LEFT', 'person ' || owner || ' left ' || strftime(p_left, '%Y-%m-%d') || ', ' || date_diff('day', p_left, {asof}) || ' days before {AS_OF}' FROM b2 WHERE kind = 'person' AND p_hr AND NOT p_cur AND p_left IS NOT NULL UNION ALL SELECT no, 'GROUP_NOBODY_CURRENT', kind || ' ' || owner || ' has no members' FROM b2 WHERE kind IN ({quote_list(GROUP_KINDS)}) AND n_people = 0 UNION ALL SELECT no, 'GROUP_NOBODY_CURRENT', kind || ' ' || owner || ': none of its ' || n_people || ' member(s) works here' || coalesce(' (latest left ' || strftime(g_left, '%Y-%m-%d') || ')', '') FROM b2 WHERE kind IN ({quote_list(GROUP_KINDS)}) AND n_people > 0 AND n_current = 0 UNION ALL SELECT no, 'SERVICE_ACCOUNT_NO_PERSON', 'service account ' || owner || ' has no named person' FROM b2 WHERE kind = 'service_account' AND email IS NULL UNION ALL SELECT no, 'SERVICE_ACCOUNT_NO_PERSON', 'service account ' || owner || ': named person has no HR record' FROM b2 WHERE kind = 'service_account' AND email IS NOT NULL AND NOT n_hr UNION ALL SELECT no, 'SERVICE_ACCOUNT_NO_PERSON', 'service account ' || owner || ': named person ' || coalesce('left ' || strftime(n_left, '%Y-%m-%d'), 'was not employed on {AS_OF}') FROM b2 WHERE kind = 'service_account' AND n_hr AND NOT n_cur""") # --- the population, with a fingerprint pop = con.execute("""SELECT platform || '|' || cat || '|' || sch || '|' || tbl || '|' || area || '|' || coalesce(owner, '') FROM base ORDER BY 1""").fetchall() sha = hashlib.sha256("\n".join(r[0] for r in pop).encode()).hexdigest() print(f"\nPopulation: {n_base} sensitive tables, owners as of {AS_OF}, run {RUN_AT}, SHA-256 {sha[:16]}...") print("Platforms") have = dict(con.execute(f"SELECT {norm('platform')}, count(*) FROM t_owners GROUP BY 1").fetchall()) for p in PLATFORMS: if have.get(p, 0) == 0: print(f" {p:<11} NOT CHECKED: no owner rows exported, so nothing is said about its tables") else: print(f" {p:<11} checked: {have[p]} table owners read") print("\nFindings (one line per gap; a table can have more than one)") rows = con.execute("""SELECT b.platform, b.area, b.cat || '.' || b.sch || '.' || b.tbl, f.code, f.detail FROM found f JOIN base b USING (no) ORDER BY b.no, f.code, f.detail""").fetchall() wide = max([len(r[2]) for r in rows] + [10]) for plat, area, tname, code, detail in rows: print(f" {plat:<11} {area:<9} {tname:<{wide}} {code:<26} {detail}") if not rows: print(" none") by = dict(con.execute("SELECT code, count(*) FROM found GROUP BY 1").fetchall()) print("\nCounts by finding") for c in CODES: print(f" {c:<26} {by.get(c, 0)}") print("\nBy platform and data area") print(f" {'platform':<11} {'data area':<9} {'tables':>6} {'with a finding':>15}") for plat, area, tot, bad in con.execute("""SELECT platform, area, count(*), count(*) FILTER (WHERE no IN (SELECT no FROM found)) FROM base GROUP BY 1, 2 ORDER BY 1, 2""").fetchall(): print(f" {plat:<11} {area:<9} {tot:>6} {bad:>15}") flagged = q("SELECT count(DISTINCT no) FROM found") print(f"\n{n_base} sensitive tables read: {n_base - flagged} with an owner who still works here, a group or role with " f"someone who does, or a service account with a named person who does; {flagged} with at least one finding.") # --- the evidence file: one row per sensitive table evidence = out / f"owner_evidence{suffix}.csv" cur = con.execute("""SELECT b.platform, b.area AS data_area, b.cat || '.' || b.sch || '.' || b.tbl AS "table", coalesce(b.owner, '') AS owner, coalesce(b.kind, '') AS owner_kind, CASE WHEN b.kind IN ('group', 'role') THEN b.n_current || ' of ' || b.n_people || ' members work here' WHEN b.kind = 'person' THEN CASE WHEN b.p_cur THEN 'works here' ELSE 'does not work here' END WHEN b.kind = 'service_account' THEN CASE WHEN b.n_cur THEN 'named person works here' ELSE 'no named person who works here' END ELSE '' END AS current_people, coalesce((SELECT string_agg(DISTINCT f.code, ' ' ORDER BY f.code) FROM found f WHERE f.no = b.no), 'OK') AS verdict, coalesce((SELECT string_agg(f.detail, ' | ' ORDER BY f.code, f.detail) FROM found f WHERE f.no = b.no), '') AS detail FROM b2 b ORDER BY b.no""") cols, data = [d[0] for d in cur.description], cur.fetchall() with open(evidence, "w", newline="") as f: w = csv.writer(f) w.writerow(cols) w.writerows([[cell(v) for v in r] for r in data]) print(f"Evidence for every sensitive table written to {evidence} ({len(data)} rows).") return 1 if ("--strict" in args and rows) else 0 if __name__ == "__main__": sys.exit(main())