#!/usr/bin/env python3 """SOX access evidence for financial tables in Snowflake, Databricks and Redshift, in DuckDB. python3 sox_access_evidence.py --demo # clean generated records python3 sox_access_evidence.py --demo --plant # generated records with planted faults python3 sox_access_evidence.py # your exports: set the settings below first python3 sox_access_evidence.py --strict # exit 1 unless nothing needs follow-up Reads each platform's own records, exported as CSV with the platform's published column names, plus an approval log, an HR leaver file and a login map. Prints three lists for the financial tables you name, each with its period, filters, run time, row count and a SHA-256 of its rows: 1. every access grant in the period, matched to an approval (NO APPROVAL, APPROVED AFTER GRANT, SELF-APPROVED are flagged) 2. every HR leaver in the period, per platform: the first removal record credited to that departure, the days between, and use of a financial table after the removal 3. anyone who both wrote to a financial table and ran or approved access to one in the period, and every login that could not be matched to a person Before it builds them it runs input checks and marks the lists INCOMPLETE if one fails. A platform you do not export is printed as NOT CHECKED, never as clean. The demo records are GENERATED (seed 228): not a company's data. Column names come from: Snowflake https://docs.snowflake.com/en/sql-reference/account-usage/grants_to_users https://docs.snowflake.com/en/sql-reference/account-usage/grants_to_roles https://docs.snowflake.com/en/sql-reference/account-usage/query_history https://docs.snowflake.com/en/sql-reference/account-usage/access_history Databricks https://docs.databricks.com/aws/en/admin/system-tables/audit-logs https://docs.databricks.com/aws/en/admin/system-tables/lineage Redshift https://docs.aws.amazon.com/redshift/latest/dg/r_STL_UTILITYTEXT.html https://docs.aws.amazon.com/redshift/latest/dg/SYS_USERLOG.html https://docs.aws.amazon.com/redshift/latest/dg/SYS_QUERY_HISTORY.html Databricks' audit log reference (https://docs.databricks.com/aws/en/admin/account-settings/audit-logs) lists securable_full_name and changes as the request parameters of updatePermissions, and the system-table page shows full_name_arg and statusCode in its examples. The layout inside changes (a JSON list of principal, add, remove) is our assumption: open one of your own events and check it. """ import csv import datetime as dt import hashlib import json import pathlib import random import re 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) AS_OF = "2026-10-15" # the day you took the exports (UTC); records must run to here EXPORT_DIR = None # folder holding the export files named in FILES below PLATFORMS_EXPORTED = ["snowflake", "databricks", "redshift"] # a platform not listed is NOT CHECKED FINANCIAL_TABLES = None # no default: {"snowflake": ["db.schema.table"], "databricks": [...], "redshift": ["schema.table"]} KNOWN_GAPS = {} # days you can explain, e.g. {"redshift": ["2026-08-19"]} HR_EMAIL_COL = "email" # HR leaver file: the work email HR_LAST_DAY_COL = "last_day" # HR leaver file: the last day of work, as YYYY-MM-DD APPROVAL_GRANTEE_COL = "grantee" # approval log: who the access was approved for (email) APPROVAL_OBJECT_COL = "object" # approval log: the table, written as in FINANCIAL_TABLES APPROVAL_APPROVER_COL = "approver" # approval log: who approved (email) APPROVAL_DATE_COL = "approved_on" # approval log: the day of approval, as YYYY-MM-DD 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 ACCESS_PRIVILEGES = ["select", "insert", "update", "delete", "modify", "ownership", "all", "all privileges"] # ---------------------------------------------------------------------------------------------- FILES = { "snowflake_grants_to_users": "snowflake_grants_to_users.csv", "snowflake_grants_to_roles": "snowflake_grants_to_roles.csv", "snowflake_query_history": "snowflake_query_history.csv", "snowflake_access_history": "snowflake_access_history.csv", "databricks_audit": "databricks_audit.csv", "databricks_table_lineage": "databricks_table_lineage.csv", "redshift_stl_utilitytext": "redshift_stl_utilitytext.csv", "redshift_sys_userlog": "redshift_sys_userlog.csv", "redshift_sys_query_history": "redshift_sys_query_history.csv", "hr_leavers": "hr_leavers.csv", "approvals": "approvals.csv", "login_map": "login_map.csv", } PLATFORM_FILES = { "snowflake": ["snowflake_grants_to_users", "snowflake_grants_to_roles", "snowflake_query_history", "snowflake_access_history"], "databricks": ["databricks_audit", "databricks_table_lineage"], "redshift": ["redshift_stl_utilitytext", "redshift_sys_userlog", "redshift_sys_query_history"], } COLUMNS = { # the platforms' published column names, lower-cased here; the file's own case does not matter "snowflake_grants_to_users": ["created_on", "deleted_on", "role", "grantee_name", "granted_by"], "snowflake_grants_to_roles": ["created_on", "privilege", "granted_on", "name", "table_catalog", "table_schema", "granted_to", "grantee_name", "granted_by", "deleted_on"], "snowflake_query_history": ["query_id", "query_text", "query_type", "user_name", "role_name", "start_time", "execution_status"], "snowflake_access_history": ["query_id", "query_start_time", "user_name", "objects_modified"], "databricks_audit": ["event_time", "event_date", "user_identity", "service_name", "action_name", "request_params", "response"], "databricks_table_lineage": ["target_table_full_name", "created_by", "event_time", "entity_type"], "redshift_stl_utilitytext": ["userid", "xid", "pid", "starttime", "endtime", "sequence", "text"], "redshift_sys_userlog": ["user_id", "user_name", "action", "record_time"], "redshift_sys_query_history": ["user_id", "username", "query_id", "query_type", "status", "start_time", "query_text"], "login_map": ["platform", "login", "email"], } OUT_DIR = pathlib.Path("evidence-out") GEN_DIR = pathlib.Path("generated") DEMO_SCOPE = {"snowflake": ["finance_db.ledger.gl_entries"], "databricks": ["finance.ledger.gl_entries"], "redshift": ["ledger.gl_entries"]} TS_RE = "\\d{4}-\\d{2}-\\d{2}[ T]\\d{2}:\\d{2}:\\d{2}(\\.\\d+)?( ?(Z|\\+00:00|\\+0000))?" DATE_RE = "\\d{4}-\\d{2}-\\d{2}" # Redshift GRANT / REVOKE with one privilege list, one table and one grantee DCL_RE = '(?i)^\\s*(grant|revoke)\\s+(.+?)\\s+on\\s+(?:table\\s+)?([\\w."$-]+)\\s+(?:to|from)\\s+(?:group\\s+|role\\s+|user\\s+)?([\\w."$@-]+)\\s*;?\\s*$' def ts(col): """A timestamp column read strictly: YYYY-MM-DD HH:MM:SS, taken as UTC (a Z, +00:00 or +0000 suffix is accepted, any other offset is NULL and counted). Export from a session set to UTC.""" return (f"CASE WHEN regexp_full_match(trim({col}), '{TS_RE}') THEN " f"try_strptime(substr(replace(trim({col}), 'T', ' '), 1, 19), '%Y-%m-%d %H:%M:%S') END") def day(col): return (f"CASE WHEN regexp_full_match(trim({col}), '{DATE_RE}') THEN " f"try_strptime(trim({col}), '%Y-%m-%d')::DATE END") 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() # --------------------------------------------------------------------------------------------- # Demo records: generated, in each platform's published column names # --------------------------------------------------------------------------------------------- def generate(plant): rnd = random.Random(228) out = GEN_DIR / ("planted" if plant else "clean") out.mkdir(parents=True, exist_ok=True) D = dt.date start, as_of = D(2026, 7, 1), D(2026, 10, 15) em = lambda k: f"{k}@example.test" rows = {k: [] for k in FILES} qn = [1000] def at(d, hour=None): return dt.datetime.combine(d, dt.time(hour if hour is not None else rnd.randrange(8, 18), rnd.randrange(60), rnd.randrange(60))) sfts = lambda t: t.strftime("%Y-%m-%d %H:%M:%S.000 +0000") plain = lambda t: t.strftime("%Y-%m-%d %H:%M:%S") rs_ids = {"f01": 101, "f02": 102, "f03": 103, "f04": 104, "f05": 105, "f06": 106, "f07": 107, "f08": 108, "a01": 201, "a02": 202, "m01": 301, "m02": 302, "etl_svc": 401, "rs_batch2": 402} def sf_q(t, user, role, qtype, text): qn[0] += 1 rows["snowflake_query_history"].append([f"sfq{qn[0]:06d}", text, qtype, user.upper(), role, sfts(t), "SUCCESS"]) return f"sfq{qn[0]:06d}" def sf_use(t, user, write=False): text = ("INSERT INTO finance_db.ledger.gl_entries SELECT 1" if write else "SELECT * FROM finance_db.ledger.gl_entries LIMIT 10") q = sf_q(t, em(user), "FIN_READER", "INSERT" if write else "SELECT", text) if write: rows["snowflake_access_history"].append( [q, sfts(t), em(user).upper(), json.dumps([{"objectName": "FINANCE_DB.LEDGER.GL_ENTRIES", "objectDomain": "Table", "columns": []}])]) def dbx(t, actor, action, params, status=200, service="unityCatalog"): rows["databricks_audit"].append([plain(t), t.date().isoformat(), json.dumps({"email": actor}), service, action, json.dumps(params), json.dumps({"statusCode": status})]) def dbx_perm(t, actor, principal, op, privs, status=200): dbx(t, actor, "updatePermissions", {"securable_type": "table", "securable_full_name": "finance.ledger.gl_entries", "changes": json.dumps([{"principal": principal, op: privs}])}, status) def dbx_use(t, user, write=False): if write: rows["databricks_table_lineage"].append(["finance.ledger.gl_entries", em(user), plain(t), "NOTEBOOK"]) else: dbx(t, em(user), "getTable", {"full_name_arg": "finance.ledger.gl_entries"}) xid = [5000] def rs_dcl(t, admin, text): xid[0] += 1 pad = text.ljust(((len(text) + 199) // 200) * 200) for seq in range(len(pad) // 200): rows["redshift_stl_utilitytext"].append([rs_ids[admin], xid[0], 9000 + xid[0], plain(t), plain(t), seq, pad[seq * 200:(seq + 1) * 200]]) def rs_q(t, user, qtype, text, status="success"): qn[0] += 1 rows["redshift_sys_query_history"].append([rs_ids[user], qn[0], qtype, status, plain(t), text]) # ---- Snowflake roles for role, privs in [("FIN_READER", ["SELECT"]), ("FIN_WRITER", ["SELECT", "INSERT", "UPDATE"])]: for p in privs: rows["snowflake_grants_to_roles"].append(["2026-01-10 09:00:00.000 +0000", p, "TABLE", "GL_ENTRIES", "FINANCE_DB", "LEDGER", "ROLE", role, "SECURITYADMIN", ""]) rows["snowflake_grants_to_roles"].append(["2026-01-10 09:00:00.000 +0000", "SELECT", "TABLE", "DEPT_DIRECTORY", "HR_DB", "PUBLIC", "ROLE", "HR_READER", "SECURITYADMIN", ""]) # ---- access: (person, platform, role or privilege, grant day, approval day, approver, who granted) plan = [("f01", "snowflake", "FIN_READER", D(2026, 7, 6), D(2026, 7, 9) if plant else D(2026, 7, 3), "m01", "a01"), ("f01", "databricks", "MODIFY", D(2026, 7, 8), D(2026, 7, 7), "m01", "a02"), ("f02", "redshift", "INSERT", D(2026, 7, 15), D(2026, 7, 14), "f02" if plant else "m02", "a01"), ("f03", "snowflake", "FIN_WRITER", D(2026, 7, 22), D(2026, 7, 20), "m02", "a02"), ("f04", "databricks", "SELECT", D(2026, 8, 3), None if plant else D(2026, 8, 1), "m01", "a01"), ("f08", "databricks", "SELECT", D(2026, 7, 28), D(2026, 7, 27), "m02", "a02")] before = [("f05", "snowflake", "FIN_READER", D(2026, 5, 11), "a01"), ("f05", "databricks", "SELECT", D(2026, 5, 12), "a02"), ("f06", "redshift", "SELECT", D(2026, 5, 13), "a01"), ("f07", "snowflake", "FIN_READER", D(2026, 5, 14), "a02"), ("f08", "databricks", "SELECT", D(2026, 5, 15), "a01")] sf_grant = {} approvals = [] tk = 100 allgr = [(p, pl, w, g, None, None, ad) for p, pl, w, g, ad in before] + plan for person, plat, what, d, ap_day, approver, admin in allgr: t = at(d) if plat == "snowflake": row = [sfts(t), "", what, em(person).upper(), "SECURITYADMIN"] rows["snowflake_grants_to_users"].append(row) sf_grant[(person, d)] = row sf_q(t, em(admin), "SECURITYADMIN", "GRANT", f'GRANT ROLE {what} TO USER "{em(person).upper()}"') obj = "finance_db.ledger.gl_entries" elif plat == "databricks": dbx_perm(t, em(admin), em(person), "add", [what]) obj = "finance.ledger.gl_entries" else: tkt = f"/* ticket T-{tk} " + "x" * 215 + " */ " if person == "f02" else "" rs_dcl(t, admin, f"{tkt}GRANT {what} ON TABLE ledger.gl_entries TO {person};") obj = "ledger.gl_entries" tk += 1 if ap_day is not None: approvals.append([f"T-{tk}", em(person).title(), obj.upper() if plat == "snowflake" else obj, em(approver), ap_day.isoformat()]) # ---- removals hr =[[em("f05"), "2026-08-07"], [em("f06"), "2026-08-21"], [em("f07"), "2026-09-04"], [em("f08"), "2026-07-10"], [em("f08"), "2026-09-18"]] if not plant: sf_grant[("f05", D(2026, 5, 11))][1] = sfts(at(D(2026, 8, 8))) dbx_perm(at(D(2026, 8, 8)), em("a01"), em("f05"), "remove", ["SELECT"]) rs_dcl(at(D(2026, 8, 30) if plant else D(2026, 8, 22)), "a02", "REVOKE SELECT ON TABLE ledger.gl_entries FROM f06;") sf_grant[("f07", D(2026, 5, 14))][1] = sfts(at(D(2026, 9, 5))) dbx_perm(at(D(2026, 7, 11)), em("a02"), em("f08"), "remove", ["SELECT"]) if not plant: dbx_perm(at(D(2026, 9, 19)), em("a01"), em("f08"), "remove", ["SELECT"]) rows["redshift_sys_userlog"].append([201, "f06", "alter", plain(at(D(2026, 5, 13)))]) # ---- use of the financial tables def every(a, b, step, fn): d = a while d <= min(b, as_of): fn(at(d), d) d += dt.timedelta(days=step) every(D(2026, 7, 7), as_of, 6, lambda t, d: sf_use(t, "f01")) every(D(2026, 7, 9), as_of, 6, lambda t, d: dbx_use(t, "f01", write=True)) every(D(2026, 7, 16), as_of, 5, lambda t, d: (rs_q(t, "f02", "INSERT", "INSERT INTO ledger.gl_entries SELECT 1"),)) every(D(2026, 7, 23), as_of, 6, lambda t, d: sf_use(t, "f03", write=True)) every(D(2026, 8, 4), as_of, 9, lambda t, d: dbx_use(t, "f04")) every(D(2026, 7, 1), D(2026, 8, 7), 7, lambda t, d: sf_use(t, "f05")) every(D(2026, 7, 1), D(2026, 8, 7), 9, lambda t, d: dbx_use(t, "f05")) every(D(2026, 7, 1), D(2026, 8, 21), 8, lambda t, d: rs_q(t, "f06", "SELECT", "SELECT * FROM ledger.gl_entries LIMIT 10")) every(D(2026, 7, 1), D(2026, 9, 4), 9, lambda t, d: sf_use(t, "f07")) every(D(2026, 7, 1), D(2026, 7, 10), 4, lambda t, d: dbx_use(t, "f08")) every(D(2026, 7, 29), D(2026, 9, 18), 8, lambda t, d: dbx_use(t, "f08")) every(D(2026, 7, 5), as_of, 10, lambda t, d: rs_q(t, "etl_svc", "INSERT", "INSERT INTO ledger.gl_entries SELECT 1")) # ---- background: every day has some records on every platform d = start while d <= as_of: sf_q(at(d), "bg@example.test", "ANALYST", "SELECT", "SELECT 1") dbx(at(d), "bg@example.test", "getTable", {"full_name_arg": "hr.dept.directory"}) rs_q(at(d), "a01", "SELECT", "SELECT 1") rs_q(at(d), "a02", "SELECT", "SELECT 1") d += dt.timedelta(days=1) dbx_perm(at(D(2026, 8, 12)), em("a01"), em("f09"), "add", ["SELECT"], status=403) # failed attempt: not an access change for u in ["m01", "m02"]: rs_q(at(D(2026, 7, 2)), u, "SELECT", "SELECT 1") if plant: sf_use(at(D(2026, 8, 15)), "f05") # leaver f05 still reads after the grace period sf_use(at(D(2026, 8, 20)), "f05") rs_q(at(D(2026, 8, 25)), "f06", "SELECT", "SELECT * FROM ledger.gl_entries LIMIT 10") # f06, before the late removal sf_use(at(D(2026, 9, 14)), "f07") # f07 reads 9 days after the removal dbx_use(at(D(2026, 8, 20)), "a02", write=True) # an admin who grants access also writes rs_q(at(D(2026, 8, 12)), "rs_batch2", "INSERT", "INSERT INTO ledger.gl_entries SELECT 1") # a login nobody mapped rs_dcl(at(D(2026, 8, 26)), "a01", "GRANT SELECT ON TABLE ledger.gl_entries TO f03, f04;") # one grantee list rows["snowflake_grants_to_users"].append(["07/09/2026 10:00", "", "HR_READER", "F03@EXAMPLE.TEST", "SECURITYADMIN"]) rows["snowflake_query_history"].append(list(rows["snowflake_query_history"][40])) # a repeated query id hr.append([em("f09"), "8/5/26"]) approvals.append(["T-999", em("f02"), "ledger.gl_entries", "", "2026-08-01"]) rows["redshift_sys_query_history"] = [r for r in rows["redshift_sys_query_history"] if not r[4].startswith("2026-08-19")] rows["redshift_stl_utilitytext"] = [r for r in rows["redshift_stl_utilitytext"] if not r[3].startswith("2026-08-19")] # ---- write for key, cols in COLUMNS.items(): if key in ("login_map", "redshift_sys_query_history"): continue name = FILES[key] head = [c.upper() for c in cols] if key.startswith("snowflake") else cols with open(out / name, "w", newline="") as f: w = csv.writer(f) w.writerow(head) w.writerows(rows[key]) rs_names = {rs_ids[k]: k for k in rs_ids} for r in rows["redshift_sys_query_history"]: r.insert(1, rs_names[r[0]]) with open(out / FILES["redshift_sys_query_history"], "w", newline="") as f: w = csv.writer(f) w.writerow(COLUMNS["redshift_sys_query_history"]) w.writerows(rows["redshift_sys_query_history"]) with open(out / FILES["hr_leavers"], "w", newline="") as f: w = csv.writer(f) w.writerow(["email", "last_day"]) w.writerows(hr) with open(out / FILES["approvals"], "w", newline="") as f: w = csv.writer(f) w.writerow(["ticket", "grantee", "object", "approver", "approved_on"]) w.writerows(approvals) with open(out / FILES["login_map"], "w", newline="") as f: w = csv.writer(f) w.writerow(["platform", "login", "email"]) w.writerows([["redshift", k, em(k)] for k in rs_ids if k not in ("etl_svc", "rs_batch2")] + [["redshift", "etl_svc", "data-eng@example.test"]]) return str(out) # --------------------------------------------------------------------------------------------- def load(con, key, folder, cols, renames=None): """Read a CSV as text with its header lower-cased; return a problem string or None.""" path = pathlib.Path(folder) / FILES[key] if not path.is_file(): return f"{FILES[key]} not found in {folder}" names = [r[0] for r in con.execute("DESCRIBE SELECT * FROM read_csv(?, header=true, all_varchar=true)", [str(path)]).fetchall()] low = [n.strip().lower() for n in names] renames = renames or {} want = [renames.get(c, c) for c in cols] missing = [c for c in want if c not in low] if missing or len(set(low)) != len(low): return f"{FILES[key]} lacks column(s) {missing}" if missing else f"{FILES[key]} has a column name twice" sel = ", ".join(f'"{names[low.index(c)]}" AS "{cols[i]}"' for i, c in enumerate(want)) con.execute(f"CREATE TABLE raw_{key} AS SELECT {sel} FROM read_csv(?, header=true, all_varchar=true)", [str(path)]) return None def population(con, name, title, filters, sql, header, complete, out_dir): 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== {title}: {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(): demo, plant, strict = "--demo" in sys.argv, "--plant" in sys.argv, "--strict" in sys.argv folder, scope = EXPORT_DIR, FINANCIAL_TABLES if demo: folder = generate(plant) scope = {k: list(v) for k, v in DEMO_SCOPE.items()} if plant: scope["redshift"].append("ledger.gl_entires") # planted: a typo in the scope print(f"input : GENERATED records (not a company's data), seed 228, plant={plant}") else: print("input : " + str(folder)) if scope is None: print("STOP: FINANCIAL_TABLES has no default. Name the financial tables for each platform, then run again.") return 2 if folder is None: print("STOP: EXPORT_DIR is not set. Point it at the folder holding your export files, then run again.") return 2 bad_settings = [n for n, v in [("PERIOD_START", PERIOD_START), ("PERIOD_END", PERIOD_END), ("AS_OF", AS_OF)] if not re.fullmatch(DATE_RE, str(v))] if bad_settings or not PERIOD_START <= PERIOD_END <= AS_OF: print("STOP: PERIOD_START, PERIOD_END and AS_OF must be YYYY-MM-DD with PERIOD_START <= PERIOD_END <= AS_OF.") return 2 p0, p1, asof = PERIOD_START, PERIOD_END, AS_OF exported = [p for p in ["snowflake", "databricks", "redshift"] if p in PLATFORMS_EXPORTED] not_checked = [p for p in ["snowflake", "databricks", "redshift"] if p not in exported] unknown = [p for p in PLATFORMS_EXPORTED if p not in PLATFORM_FILES] if unknown or not exported: print(f"STOP: PLATFORMS_EXPORTED must list snowflake, databricks and/or redshift (got {PLATFORMS_EXPORTED}).") return 2 print("platforms : checked " + ", ".join(exported) + ("; NOT CHECKED: " + ", ".join(not_checked) if not_checked else "")) con = duckdb.connect() # ---- load the files; a missing file or column stops the run: a check that cannot run is a failure problems = [] for p in exported: for key in PLATFORM_FILES[p]: problems.append(load(con, key, folder, COLUMNS[key])) problems.append(load(con, "login_map", folder, COLUMNS["login_map"])) problems.append(load(con, "hr_leavers", folder, ["email", "last_day"], {"email": HR_EMAIL_COL.lower(), "last_day": HR_LAST_DAY_COL.lower()})) problems.append(load(con, "approvals", folder, ["grantee", "object", "approver", "approved_on"], {"grantee": APPROVAL_GRANTEE_COL.lower(), "object": APPROVAL_OBJECT_COL.lower(), "approver": APPROVAL_APPROVER_COL.lower(), "approved_on": APPROVAL_DATE_COL.lower()})) problems = [x for x in problems if x] if problems: print("\n== Input checks") for x in problems: print("[FAIL] " + x) print("RESULT: STOPPED: the files above could not be read, so no list was built") return 2 names = [(p, o.strip().lower()) for p in exported for o in scope.get(p, [])] con.execute("CREATE TABLE scope (platform VARCHAR, obj VARCHAR)") con.executemany("INSERT INTO scope VALUES (?, ?)", names) con.execute("CREATE TABLE privs (p VARCHAR)") norm = lambda x: x.strip().lower().replace("_", " ") # ALL_PRIVILEGES (Databricks) is "all privileges" con.executemany("INSERT INTO privs VALUES (?)", [(norm(x),) for x in ACCESS_PRIVILEGES]) priv_list = "[" + ", ".join("'" + norm(x).replace("'", "''") + "'" for x in ACCESS_PRIVILEGES) + "]" con.execute("""CREATE TABLE ident AS SELECT lower(trim(platform)) AS platform, lower(trim(login)) AS login, lower(trim(email)) AS email FROM raw_login_map WHERE trim(coalesce(login, '')) <> '' AND trim(coalesce(email, '')) <> ''""") # a login that is itself an email is the person; anything else needs a row in login_map con.execute("""CREATE MACRO who(p, l) AS coalesce( (SELECT any_value(email) FROM ident WHERE ident.platform = p AND ident.login = l), CASE WHEN l LIKE '%@%' THEN l END)""") bad_ts = [] # (label, count) of timestamps that are not YYYY-MM-DD HH:MM:SS in UTC def count_bad(label, table, text_col, parsed_col, required=True): cond = f"{parsed_col} IS NULL" if required else f"trim(coalesce({text_col}, '')) <> '' AND {parsed_col} IS NULL" bad_ts.append((label, con.execute(f"SELECT count(*) FROM {table} WHERE {cond}").fetchone()[0])) # ---- Snowflake dup = 0 if "snowflake" in exported: con.execute(f"""CREATE TABLE sf_gu AS SELECT {ts('created_on')} AS ts, trim(created_on) AS ts_text, {ts('deleted_on')} AS end_ts, trim(coalesce(deleted_on, '')) AS end_text, lower(trim(role)) AS role, lower(trim(grantee_name)) AS login, lower(trim(granted_by)) AS granted_by FROM raw_snowflake_grants_to_users""") con.execute(f"""CREATE TABLE sf_gr AS SELECT {ts('created_on')} AS ts, {ts('deleted_on')} AS end_ts, trim(coalesce(deleted_on, '')) AS end_text, upper(trim(privilege)) AS privilege, lower(trim(table_catalog) || '.' || trim(table_schema) || '.' || trim(name)) AS obj, lower(trim(grantee_name)) AS role FROM raw_snowflake_grants_to_roles WHERE upper(trim(granted_on)) = 'TABLE' AND upper(trim(granted_to)) = 'ROLE'""") con.execute(f"""CREATE TABLE sf_qh AS SELECT trim(query_id) AS id, lower(replace(query_text, '"', '')) AS text, upper(trim(query_type)) AS query_type, lower(trim(user_name)) AS login, lower(trim(role_name)) AS role, {ts('start_time')} AS ts, lower(trim(execution_status)) AS status FROM raw_snowflake_query_history""") con.execute(f"""CREATE TABLE sf_ah AS SELECT trim(query_id) AS id, {ts('query_start_time')} AS ts, lower(trim(user_name)) AS login, objects_modified FROM raw_snowflake_access_history""") count_bad("GRANTS_TO_USERS.CREATED_ON", "sf_gu", "ts_text", "ts") count_bad("GRANTS_TO_USERS.DELETED_ON", "sf_gu", "end_text", "end_ts", False) count_bad("GRANTS_TO_ROLES.CREATED_ON", "sf_gr", "ts", "ts") count_bad("GRANTS_TO_ROLES.DELETED_ON", "sf_gr", "end_text", "end_ts", False) count_bad("QUERY_HISTORY.START_TIME", "sf_qh", "ts", "ts") count_bad("ACCESS_HISTORY.QUERY_START_TIME", "sf_ah", "ts", "ts") dup += con.execute("SELECT count(*) FROM (SELECT id FROM sf_qh GROUP BY id HAVING count(*) > 1)").fetchone()[0] con.execute("CREATE TABLE sf_mod (id VARCHAR, ts TIMESTAMP, login VARCHAR, obj VARCHAR)") bad_json, mod_rows = 0, [] for qid, t, login, js in con.execute("SELECT id, ts, login, objects_modified FROM sf_ah").fetchall(): try: for o in json.loads(js or "[]"): mod_rows.append((qid, t, login, str(o["objectName"]).strip().lower())) except (ValueError, KeyError, TypeError): bad_json += 1 con.executemany("INSERT INTO sf_mod VALUES (?, ?, ?, ?)", mod_rows) con.execute(f"""CREATE TABLE sf_acc AS SELECT u.ts, u.end_ts, u.login, u.role, u.granted_by, r.obj, string_agg(DISTINCT lower(replace(r.privilege, '_', ' ')), ',' ORDER BY lower(replace(r.privilege, '_', ' '))) AS privs FROM sf_gu u JOIN sf_gr r ON r.role = u.role AND r.obj IN (SELECT obj FROM scope WHERE platform = 'snowflake') AND lower(replace(r.privilege, '_', ' ')) IN (SELECT p FROM privs) AND r.ts <= u.ts AND (r.end_ts IS NULL OR r.end_ts > u.ts) WHERE u.ts IS NOT NULL GROUP BY ALL""") else: bad_json = 0 # ---- Databricks bad_changes = 0 if "databricks" in exported: con.execute(f"""CREATE TABLE dbx_a AS SELECT {ts('event_time')} AS ts, trim(event_time) AS ts_text, lower(trim(json_extract_string(user_identity, '$.email'))) AS login, lower(trim(service_name)) AS service, trim(action_name) AS action, request_params, response, coalesce(json_extract_string(response, '$.statusCode'), json_extract_string(response, '$.status_code')) AS status FROM raw_databricks_audit""") con.execute(f"""CREATE TABLE dbx_l AS SELECT lower(trim(target_table_full_name)) AS obj, lower(trim(created_by)) AS login, {ts('event_time')} AS ts, trim(event_time) AS ts_text FROM raw_databricks_table_lineage""") count_bad("system.access.audit event_time", "dbx_a", "ts_text", "ts") count_bad("system.access.table_lineage event_time", "dbx_l", "ts_text", "ts") con.execute("CREATE TABLE dbx_chg (ts TIMESTAMP, actor VARCHAR, obj VARCHAR, principal VARCHAR, op VARCHAR, privilege VARCHAR)") chg, failed_changes = [], 0 dbx_scope = {o for p, o in names if p == "databricks"} for t, actor, status, rp in con.execute("""SELECT ts, login, status, request_params FROM dbx_a WHERE service = 'unitycatalog' AND action = 'updatePermissions' AND ts IS NOT NULL""").fetchall(): try: p = json.loads(rp) obj = str(p["securable_full_name"]).strip().lower() if obj not in dbx_scope: # a change on another table says nothing here continue if status is None or not status.strip().isdigit(): # unreadable: the change cannot be judged bad_changes += 1 continue if status.strip() != "200": failed_changes += 1 continue ch = p["changes"] for c in (json.loads(ch) if isinstance(ch, str) else ch): for op in ("add", "remove"): for pr in c.get(op) or []: chg.append((t, actor, obj, str(c["principal"]).strip().lower(), op, str(pr).strip().lower().replace("_", " "))) except (ValueError, KeyError, TypeError, AttributeError): bad_changes += 1 con.executemany("INSERT INTO dbx_chg VALUES (?, ?, ?, ?, ?, ?)", chg) bad_changes += con.execute("""SELECT count(*) FROM dbx_a WHERE service = 'unitycatalog' AND action <> 'updatePermissions' AND lower(json_extract_string(request_params, '$.full_name_arg')) IN (SELECT obj FROM scope WHERE platform = 'databricks') AND (status IS NULL OR NOT regexp_full_match(trim(status), '\\d+'))""").fetchone()[0] else: failed_changes = 0 # ---- Redshift bad_dcl = 0 if "redshift" in exported: con.execute(f"""CREATE TABLE rs_qh AS SELECT trim(user_id) AS uid, lower(trim(username)) AS login, trim(query_id) AS id, upper(trim(query_type)) AS query_type, lower(trim(status)) AS status, {ts('start_time')} AS ts, lower(replace(query_text, '"', '')) AS text FROM raw_redshift_sys_query_history""") con.execute(f"""CREATE TABLE rs_ut AS SELECT trim(userid) AS uid, trim(xid) AS xid, trim(pid) AS pid, trim(starttime) AS st_text, {ts('starttime')} AS ts, try_cast(sequence AS INTEGER) AS seq, text FROM raw_redshift_stl_utilitytext""") con.execute(f"""CREATE TABLE rs_ul AS SELECT lower(trim(user_name)) AS login, lower(trim(action)) AS action, {ts('record_time')} AS ts, trim(record_time) AS ts_text FROM raw_redshift_sys_userlog""") count_bad("SYS_QUERY_HISTORY.start_time", "rs_qh", "ts", "ts") count_bad("STL_UTILITYTEXT.starttime", "rs_ut", "st_text", "ts") count_bad("SYS_USERLOG.record_time", "rs_ul", "ts_text", "ts") con.execute("CREATE TABLE rs_users AS SELECT uid, any_value(login) AS login FROM rs_qh GROUP BY uid") # a statement longer than 200 characters is stored in pieces: put them back in order, as AWS shows con.execute(f"""CREATE TABLE rs_dcl AS WITH st AS (SELECT uid, xid, pid, st_text, min(ts) AS ts, string_agg(CASE WHEN len(rtrim(text)) = 0 THEN text ELSE rtrim(text) END, '' ORDER BY seq) AS stmt FROM rs_ut GROUP BY uid, xid, pid, st_text), cl AS (SELECT *, trim(regexp_replace(stmt, '/\\*.*?\\*/', '', 'gs')) AS s FROM st) SELECT uid, ts, s, lower(regexp_extract(s, '{DCL_RE}', 1)) AS kind, regexp_extract(s, '{DCL_RE}', 2) AS privs, lower(replace(regexp_extract(s, '{DCL_RE}', 3), '"', '')) AS obj, lower(replace(regexp_extract(s, '{DCL_RE}', 4), '"', '')) AS grantee FROM cl WHERE regexp_matches(s, '(?i)^\\s*(grant|revoke)\\b')""") bad_dcl = con.execute("""SELECT count(*) FROM rs_dcl WHERE kind = '' AND EXISTS ( SELECT 1 FROM scope WHERE platform = 'redshift' AND contains(lower(s), obj))""").fetchone()[0] # ---- events: grants, ends of access, use of the financial tables (login level) parts_g, parts_e, parts_u = [], [], [] if "snowflake" in exported: con.execute("""CREATE TABLE sf_g AS WITH m AS ( SELECT a.*, q.login AS grantor, row_number() OVER (PARTITION BY a.ts, a.login, a.role, a.obj ORDER BY abs(epoch(q.ts) - epoch(a.ts))) AS rn FROM sf_acc a LEFT JOIN sf_qh q ON q.query_type = 'GRANT' AND q.role = a.granted_by AND abs(epoch(q.ts) - epoch(a.ts)) <= 60 AND contains(q.text, a.role) AND contains(q.text, a.login)) SELECT * FROM m WHERE rn = 1""") parts_g.append("SELECT 'snowflake' AS platform, ts, login, obj, privs, grantor AS grantor_login FROM sf_g") parts_e.append("SELECT 'snowflake' AS platform, end_ts AS ts, login, obj FROM sf_acc WHERE end_ts IS NOT NULL") parts_u.append("""SELECT 'snowflake' AS platform, q.ts, q.login, s.obj, false AS is_write FROM sf_qh q JOIN scope s ON s.platform = 'snowflake' AND contains(q.text, s.obj) WHERE q.query_type <> 'GRANT' AND q.ts IS NOT NULL AND q.status = 'success' AND NOT EXISTS (SELECT 1 FROM sf_mod m WHERE m.id = q.id AND m.obj = s.obj)""") parts_u.append("SELECT 'snowflake', m.ts, m.login, m.obj, true FROM sf_mod m WHERE m.obj IN (SELECT obj FROM scope WHERE platform = 'snowflake') AND m.ts IS NOT NULL") if "databricks" in exported: parts_g.append(f"""SELECT 'databricks' AS platform, ts, principal AS login, obj, string_agg(DISTINCT privilege, ',' ORDER BY privilege) AS privs, any_value(actor) AS grantor_login FROM dbx_chg WHERE op = 'add' AND privilege IN (SELECT p FROM privs) AND obj IN (SELECT obj FROM scope WHERE platform = 'databricks') GROUP BY ts, principal, obj""") parts_e.append("""SELECT DISTINCT 'databricks' AS platform, ts, principal AS login, obj FROM dbx_chg WHERE op = 'remove' AND privilege IN (SELECT p FROM privs) AND obj IN (SELECT obj FROM scope WHERE platform = 'databricks')""") parts_u.append("""SELECT 'databricks' AS platform, a.ts, a.login, s.obj, false AS is_write FROM dbx_a a JOIN scope s ON s.platform = 'databricks' AND lower(json_extract_string(a.request_params, '$.full_name_arg')) = s.obj WHERE a.service = 'unitycatalog' AND a.action <> 'updatePermissions' AND a.ts IS NOT NULL AND a.status = '200'""") parts_u.append("SELECT 'databricks', ts, login, obj, true FROM dbx_l WHERE obj IN (SELECT obj FROM scope WHERE platform = 'databricks') AND ts IS NOT NULL") if "redshift" in exported: parts_g.append(f"""SELECT 'redshift' AS platform, d.ts, d.grantee AS login, d.obj, lower(trim(d.privs)) AS privs, coalesce(u.login, 'userid ' || d.uid) AS grantor_login FROM rs_dcl d LEFT JOIN rs_users u ON u.uid = d.uid WHERE d.kind = 'grant' AND d.ts IS NOT NULL AND list_has_any(list_transform(string_split(lower(replace(d.privs, '_', ' ')), ','), x -> trim(x)), {priv_list}) AND d.obj IN (SELECT obj FROM scope WHERE platform = 'redshift')""") parts_e.append(f"""SELECT 'redshift' AS platform, d.ts, d.grantee AS login, d.obj FROM rs_dcl d WHERE d.kind = 'revoke' AND d.ts IS NOT NULL AND list_has_any(list_transform(string_split(lower(replace(d.privs, '_', ' ')), ','), x -> trim(x)), {priv_list}) AND d.obj IN (SELECT obj FROM scope WHERE platform = 'redshift')""") parts_e.append("SELECT 'redshift' AS platform, ts, login, NULL AS obj FROM rs_ul WHERE action = 'drop' AND ts IS NOT NULL") # a write is an INSERT, UPDATE, DELETE or COPY whose target is the financial table, not one that only reads it parts_u.append("""SELECT 'redshift' AS platform, q.ts, q.login, s.obj, q.query_type IN ('INSERT', 'UPDATE', 'DELETE', 'COPY') AND (q.target = s.obj OR suffix(q.target, '.' || s.obj)) AS is_write FROM (SELECT *, regexp_extract(trim(regexp_replace(regexp_replace(text, '/\\*.*?\\*/', '', 'gs'), '--[^\\n]*', '', 'g')), '^\\s*(?:insert\\s+into|update|delete\\s+from|delete|copy)\\s+([^\\s(]+)', 1) AS target FROM rs_qh) q JOIN scope s ON s.platform = 'redshift' AND contains(q.text, s.obj) WHERE q.status NOT IN ('failed', 'canceled') AND q.ts IS NOT NULL AND NOT regexp_matches(q.text, '^\\s*(grant|revoke)\\b')""") con.execute("CREATE TABLE grants_r AS SELECT platform, ts, login, who(platform, login) AS email, obj, privs, grantor_login, " "who(platform, grantor_login) AS grantor_email FROM (" + " UNION ALL ".join(parts_g) + ")") con.execute("CREATE TABLE ends_r AS SELECT platform, ts, login, who(platform, login) AS email, obj FROM (" + " UNION ALL ".join(parts_e) + ")") con.execute("CREATE TABLE usage_r AS SELECT platform, ts, login, who(platform, login) AS email, obj, is_write FROM (" + " UNION ALL ".join(parts_u) + ")") # ---- people and logs the lists need con.execute(f"""CREATE TABLE leavers AS SELECT lower(trim(coalesce(email, ''))) AS email, trim(coalesce(last_day, '')) AS last_day_text, {day('last_day')} AS last_day FROM raw_hr_leavers""") con.execute(f"""CREATE TABLE approvals AS SELECT lower(trim(coalesce(grantee, ''))) AS grantee, lower(trim(coalesce(object, ''))) AS obj, lower(trim(coalesce(approver, ''))) AS approver, trim(coalesce(approved_on, '')) AS date_text, {day('approved_on')} AS approved_on FROM raw_approvals""") # ---- input checks: each says which planted fault trips it days = {} for p, sql in [("snowflake", "SELECT DISTINCT ts::DATE FROM sf_qh WHERE ts IS NOT NULL"), ("databricks", "SELECT DISTINCT ts::DATE FROM dbx_a WHERE ts IS NOT NULL"), ("redshift", "SELECT DISTINCT ts::DATE FROM rs_qh WHERE ts IS NOT NULL")]: if p in exported: have = {r[0].isoformat() for r in con.execute(sql).fetchall()} allday = [r[0].isoformat() for r in con.execute( f"SELECT d::DATE FROM generate_series(DATE '{p0}', DATE '{asof}', INTERVAL 1 DAY) t(d) ORDER BY 1").fetchall()] days[p] = [d for d in allday if d not in have and d not in set(KNOWN_GAPS.get(p, []))] unseen = [] for p, obj in names: seen = con.execute("""SELECT (SELECT count(*) FROM grants_r WHERE platform = ? AND obj = ?) + (SELECT count(*) FROM usage_r WHERE platform = ? AND obj = ?) + (SELECT count(*) FROM ends_r WHERE platform = ? AND obj = ?)""", [p, obj, p, obj, p, obj]).fetchone()[0] if seen == 0: unseen.append(f"{p}:{obj}") no_scope = [p for p in exported if not scope.get(p)] hr_bad = con.execute("SELECT count(*) FROM leavers WHERE email = '' OR last_day_text = ''").fetchone()[0] hr_date = con.execute("SELECT count(*) FROM leavers WHERE last_day_text <> '' AND last_day IS NULL").fetchone()[0] ap_bad = con.execute("SELECT count(*) FROM approvals WHERE grantee = '' OR obj = '' OR approver = '' OR date_text = ''").fetchone()[0] ap_date = con.execute("SELECT count(*) FROM approvals WHERE date_text <> '' AND approved_on IS NULL").fetchone()[0] map_bad = con.execute("SELECT count(*) FROM raw_login_map WHERE trim(coalesce(login, '')) = '' OR trim(coalesce(email, '')) = ''").fetchone()[0] checks = [ ("duplicate Snowflake query ids", dup, "a repeated query id"), ("timestamps not written YYYY-MM-DD HH:MM:SS (UTC)",sum(n for _, n in bad_ts), "a timestamp written 07/09/2026 10:00"), ("days with no records, per platform, from PERIOD_START to AS_OF", sum(len(v) for v in days.values()), "a day with no Redshift records"), ("scope names that match no grant, use or removal in the exports", len(unseen) + len(no_scope), "a typo in FINANCIAL_TABLES"), ("Databricks audit events or Snowflake write lists the script could not read", bad_changes + bad_json, "malformed JSON"), ("Redshift GRANT or REVOKE on a financial table the script could not parse", bad_dcl, "a grant to two users in one statement"), ("HR leaver rows with no email or no last day", hr_bad, "a NULL in the HR file"), ("HR leaver rows whose last day is not written as YYYY-MM-DD", hr_date, "a last day written 8/5/26"), ("approval rows with a blank grantee, table, approver or date", ap_bad, "a blank approver"), ("approval rows whose date is not written as YYYY-MM-DD", ap_date, "an approval date written 8/5/26"), ("login map rows with a blank login or email", map_bad, "a blank in the login map"), ] print(f"\n== Input checks (span checked for records: {p0} to {asof})") ok = True for label, n, fault in checks: ok &= n == 0 print(f"[{'ok ' if n == 0 else 'FAIL'}] {label}: {n}" + ("" if n == 0 else f" (trips on: {fault})")) for label, n in bad_ts: if n: print(f" unreadable: {label}: {n}") for p, v in days.items(): if v: print(f" {p} days with no records: " + ", ".join(v[:12]) + (f" ... ({len(v)} days)" if len(v) > 12 else "")) if p == "redshift": print(" the STL views keep seven days (AWS): export daily, or unload the logs to Amazon S3") if unseen or no_scope: print(" scope: " + ", ".join(unseen + [f"{p}: no tables named" for p in no_scope])) for p in not_checked: print(f"[NOT CHECKED] {p}: not in PLATFORMS_EXPORTED, so nothing here says anything about it") print(f"[info] permission changes that failed (status not 200), left out: {failed_changes}") n_unk = con.execute("SELECT count(*) FROM grants_r WHERE ts::DATE BETWEEN DATE '%s' AND DATE '%s' AND grantor_login IS NULL" % (p0, p1)).fetchone()[0] print(f"[info] grants in the period where no person who ran them was found: {n_unk}") # ---- list 1: access approved before it was granted sql = f""" WITH g AS (SELECT row_number() OVER (ORDER BY ts, platform, login, obj, privs) AS gid, * FROM grants_r WHERE ts::DATE BETWEEN DATE '{p0}' AND DATE '{p1}'), cand AS (SELECT g.gid, a.approver, a.approved_on, row_number() OVER (PARTITION BY g.gid ORDER BY CASE WHEN a.approved_on <= g.ts::DATE THEN 0 ELSE 1 END, abs(date_diff('day', a.approved_on, g.ts::DATE)), a.approver = coalesce(g.email, g.login), a.approver) AS rn FROM g JOIN approvals a ON a.grantee = coalesce(g.email, g.login) AND a.obj = g.obj AND a.approved_on IS NOT NULL AND a.approver <> '') SELECT g.platform, g.ts::DATE AS granted_on, coalesce(g.email, g.login) AS grantee, g.obj AS object, g.privs, coalesce(g.grantor_email, g.grantor_login, '-') AS granted_by_person, c.approver, c.approved_on, CASE WHEN c.gid IS NULL THEN 'NO APPROVAL' WHEN c.approver = coalesce(g.email, g.login) THEN 'SELF-APPROVED' WHEN c.approved_on > g.ts::DATE THEN 'APPROVED AFTER GRANT' ELSE 'approved before grant' END AS status FROM g LEFT JOIN cand c ON c.gid = g.gid AND c.rn = 1 ORDER BY g.ts, g.platform, grantee, g.obj, g.privs""" l1 = population(con, "sox_1_access_grants", "Access approved before it was granted", f"grants of {sorted(set(ACCESS_PRIVILEGES))} on the tables in FINANCIAL_TABLES, granted in the period; one row per person and table", sql, ["platform", "granted_on", "grantee", "object", "privileges", "granted_by_person", "approver", "approved_on", "status"], ok, OUT_DIR) for r in l1: # every row when there are 30 or fewer, otherwise only the ones to explain if len(l1) <= 30 or r[8] != "approved before grant": print(" " + " | ".join("-" if v is None else str(v) for v in r)) # ---- list 2: leavers, per platform, each departure on its own plat_vals = ", ".join(f"('{p}')" for p in exported) sql = f""" WITH lv AS (SELECT email, last_day, lag(last_day) OVER (PARTITION BY email ORDER BY last_day) AS prev_last_day, lead(last_day) OVER (PARTITION BY email ORDER BY last_day) AS next_last_day FROM leavers WHERE email <> '' AND last_day IS NOT NULL), pairs AS (SELECT l.*, p.platform FROM lv l CROSS JOIN (VALUES {plat_vals}) p(platform) WHERE l.last_day BETWEEN DATE '{p0}' AND DATE '{p1}' AND (EXISTS ( SELECT 1 FROM grants_r g WHERE g.email = l.email AND g.platform = p.platform AND g.ts::DATE <= l.last_day AND (l.prev_last_day IS NULL OR g.ts::DATE > l.prev_last_day)) OR EXISTS ( SELECT 1 FROM usage_r u WHERE u.email = l.email AND u.platform = p.platform AND u.ts::DATE > l.last_day + INTERVAL {GRACE_DAYS} DAY AND l.next_last_day IS NULL))), -- a rehire's earlier departure needs a grant record credited AS (SELECT e.email, e.platform, e.ts, coalesce( (SELECT max(l.last_day) FROM lv l WHERE l.email = e.email AND l.last_day <= e.ts::DATE), (SELECT min(l.last_day) FROM lv l WHERE l.email = e.email AND l.last_day - INTERVAL {REMOVAL_WINDOW_DAYS} DAY <= e.ts::DATE)) AS last_day FROM ends_r e WHERE e.email IS NOT NULL), rem AS (SELECT email, platform, last_day, min(ts) AS removed_at FROM credited WHERE last_day IS NOT NULL GROUP BY ALL), act AS (SELECT p.email, p.last_day, p.platform, count(u.ts) FILTER (WHERE u.ts::DATE > p.last_day + INTERVAL {GRACE_DAYS} DAY AND (r.removed_at IS NULL OR u.ts < r.removed_at)) AS n_grace, count(u.ts) FILTER (WHERE r.removed_at IS NOT NULL AND u.ts > r.removed_at AND p.next_last_day IS NULL) AS n_removed FROM pairs p LEFT JOIN rem r ON r.email = p.email AND r.platform = p.platform AND r.last_day = p.last_day LEFT JOIN usage_r u ON u.email = p.email AND u.platform = p.platform GROUP BY ALL) SELECT p.email, p.last_day, p.platform, r.removed_at::DATE AS first_removal_record, date_diff('day', p.last_day, r.removed_at::DATE) AS days_to_removal, a.n_grace AS uses_after_grace, a.n_removed AS uses_after_removal, CASE WHEN r.removed_at IS NULL THEN 'NO REMOVAL RECORD' WHEN a.n_removed > 0 THEN 'ACTIVE AFTER REMOVAL' WHEN date_diff('day', p.last_day, r.removed_at::DATE) > {GRACE_DAYS} THEN 'REMOVED LATE' ELSE 'removed in time' END AS status FROM pairs p LEFT JOIN rem r ON r.email = p.email AND r.platform = p.platform AND r.last_day = p.last_day LEFT JOIN act a ON a.email = p.email AND a.platform = p.platform AND a.last_day = p.last_day ORDER BY p.last_day, p.email, p.platform""" l2 = population(con, "sox_2_leavers", "Leavers", f"HR leavers with last day in the period who held access to a financial table on the platform, or used one after the grace period; removal records credited to one departure; grace {GRACE_DAYS} day(s)", sql, ["email", "last_day", "platform", "first_removal_record", "days_to_removal", "uses_after_grace", "uses_after_removal", "status"], ok, OUT_DIR) for r in l2: print(" " + " | ".join("-" if v is None else str(v) for v in r)) for p in exported: n = con.execute(f"""SELECT count(*) FROM leavers l WHERE l.last_day BETWEEN DATE '{p0}' AND DATE '{p1}' AND l.email <> '' AND NOT EXISTS (SELECT 1 FROM grants_r g WHERE g.email = l.email AND g.platform = '{p}' AND g.ts::DATE <= l.last_day AND NOT EXISTS (SELECT 1 FROM leavers l0 WHERE l0.email = l.email AND l0.last_day < l.last_day AND l0.last_day >= g.ts::DATE)) AND NOT (EXISTS (SELECT 1 FROM usage_r u WHERE u.email = l.email AND u.platform = '{p}' AND u.ts::DATE > l.last_day + INTERVAL {GRACE_DAYS} DAY) AND NOT EXISTS (SELECT 1 FROM leavers l2 WHERE l2.email = l.email AND l2.last_day > l.last_day))""").fetchone()[0] print(f"[info] HR leavers in the period with no grant on a financial table and no use after the grace period seen on {p}, so not listed there: {n}") # ---- list 3: segregation of duties sql = f""" WITH w AS (SELECT email, count(*) AS writes, string_agg(DISTINCT platform, ',' ORDER BY platform) AS wrote_on FROM usage_r WHERE is_write AND email IS NOT NULL AND ts::DATE BETWEEN DATE '{p0}' AND DATE '{p1}' GROUP BY 1), g AS (SELECT grantor_email AS email, count(*) AS grants_run FROM grants_r WHERE grantor_email IS NOT NULL AND ts::DATE BETWEEN DATE '{p0}' AND DATE '{p1}' GROUP BY 1), a AS (SELECT approver AS email, count(*) AS approvals_given FROM approvals WHERE approver <> '' AND obj IN (SELECT obj FROM scope) AND approved_on BETWEEN DATE '{p0}' AND DATE '{p1}' GROUP BY 1) SELECT w.email, w.writes, w.wrote_on, coalesce(g.grants_run, 0) AS grants_run, coalesce(a.approvals_given, 0) AS approvals_given, 'WROTE AND ' || CASE WHEN g.grants_run > 0 AND a.approvals_given > 0 THEN 'GRANTED AND APPROVED' WHEN g.grants_run > 0 THEN 'GRANTED' ELSE 'APPROVED' END AS status FROM w LEFT JOIN g ON g.email = w.email LEFT JOIN a ON a.email = w.email WHERE coalesce(g.grants_run, 0) + coalesce(a.approvals_given, 0) > 0 ORDER BY w.email""" l3 = population(con, "sox_3_segregation_of_duties", "Segregation of duties", "people (matched by email across platforms) who wrote to a financial table and also ran a grant of access to one or approved access to one, in the period", sql, ["email", "writes", "wrote_on", "grants_run", "approvals_given", "status"], ok, OUT_DIR) for r in l3: print(" " + " | ".join(str(v) for v in r)) sql = f""" SELECT platform, login, count(*) FILTER (WHERE kind = 'write') AS writes, count(*) FILTER (WHERE kind = 'grant') AS grants_run FROM (SELECT platform, login, 'write' AS kind FROM usage_r WHERE is_write AND email IS NULL AND ts::DATE BETWEEN DATE '{p0}' AND DATE '{p1}' UNION ALL SELECT platform, grantor_login, 'grant' FROM grants_r WHERE grantor_email IS NULL AND grantor_login IS NOT NULL AND ts::DATE BETWEEN DATE '{p0}' AND DATE '{p1}') GROUP BY 1, 2 ORDER BY 1, 2""" unmatched = con.execute(sql).fetchall() print(f"[info] logins that wrote to a financial table or ran a grant and match no person (add them to login_map): {len(unmatched)}") for r in unmatched: print(" UNMATCHED LOGIN | " + " | ".join(str(v) for v in r)) # ---- result f1 = sum(1 for r in l1 if r[8] != "approved before grant") f2 = sum(1 for r in l2 if r[7] != "removed in time" or r[5] or r[6]) f3 = len(l3) + len(unmatched) + n_unk print("\n== Result") print(f"input checks : {'all passed' if ok else 'FAILED, the lists above are not complete'}") print(f"not checked : {', '.join(not_checked) if not_checked else 'none'}") print(f"follow-up : {f1} grant row(s), {f2} leaver row(s), {f3} segregation or login row(s) need an explanation") if not ok or f1 + f2 + f3: verdict = "EXCEPTIONS FOUND" elif not_checked: verdict = "NOTHING TO FOLLOW UP ON THE PLATFORMS CHECKED; NOT CHECKED: " + ", ".join(not_checked) else: verdict = "NOTHING TO FOLLOW UP" print(f"RESULT: {verdict}") return 1 if strict and verdict != "NOTHING TO FOLLOW UP" else 0 if __name__ == "__main__": sys.exit(main())