#!/usr/bin/env python3 """NIST SP 800-53 AC-2 account evidence from Snowflake, Databricks and Redshift exports, in DuckDB. python3 ac2_account_evidence.py --demo # clean generated records python3 ac2_account_evidence.py --demo --plant # generated records with planted findings python3 ac2_account_evidence.py --demo --plant-input # generated records with planted input faults python3 ac2_account_evidence.py --export-sql # print the export queries for each platform python3 ac2_account_evidence.py # your exports: set the settings below first python3 ac2_account_evidence.py --strict # exit 1 unless nothing needs follow-up AC-2 leaves every time period to the organization (each one is an organization-defined parameter), so INACTIVE_DAYS, TEMP_GRACE_DAYS and TRANSFER_DAYS have no default: a list whose setting is unset does not run. The demo uses 90, 1 and 14 so the demo has something to measure; they are not a recommendation. The script reads CSV exports of each platform's own records, with the platform's published column names, plus lists you keep (HR people and transfers, a login map, temporary accounts, a role-to-department map, change tickets). It prints, each with its period, filters, run time, row count and a SHA-256 of its rows: 1. inactive but enabled (AC-2(3)(d)): an enabled account with no sign-in for INACTIVE_DAYS, counted from the last sign-in the records hold, or from creation if there is none 2. temporary and emergency accounts past their end (AC-2(2), AC-2(3)(a)): accounts on your list, or with a Snowflake EXPIRES_AT, still enabled, used, or ended later than the end date plus TEMP_GRACE_DAYS 3. transfers that kept the old access (AC-2 h.2 and l): roles or groups your map ties to the old department that the person still holds TRANSFER_DAYS after the transfer 4. account actions in the period (AC-2(4)): each successful create, modify, enable, disable and remove the script reads from the platforms' records, and every change ticket with no recorded action (NOT RECORDED). The first is the full set of recorded actions, not a finding. Input checks run first; if one fails, the lists are marked INCOMPLETE. A platform you do not export is printed as NOT CHECKED. The script reads files; it never connects to a platform and never writes to one. The demo records are GENERATED (seed 232): not a company's data. Column names come from: Snowflake https://docs.snowflake.com/en/sql-reference/account-usage/users https://docs.snowflake.com/en/sql-reference/account-usage/grants_to_users https://docs.snowflake.com/en/sql-reference/account-usage/query_history Databricks https://docs.databricks.com/aws/en/admin/system-tables/audit-logs https://docs.databricks.com/aws/en/admin/account-settings/audit-logs https://docs.databricks.com/api/account/accountusers/list https://docs.databricks.com/api/account/accountgroups/list Redshift https://docs.aws.amazon.com/redshift/latest/dg/SYS_USERLOG.html https://docs.aws.amazon.com/redshift/latest/dg/SYS_CONNECTION_LOG.html https://docs.aws.amazon.com/redshift/latest/dg/r_SVV_USER_INFO.html https://docs.aws.amazon.com/redshift/latest/dg/r_SVV_USER_GRANTS.html """ import csv import datetime as dt import hashlib import pathlib import random import re import sys import duckdb # ---- CHANGE THESE FOR YOUR OWN DATA --------------------------------------------------------- PERIOD_START = "2026-09-01" # first day of the review period (UTC, inclusive) PERIOD_END = "2026-09-30" # last day of the review period (UTC, inclusive) AS_OF = "2026-10-05" # 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 INACTIVE_DAYS = None # no default: days without a sign-in before an enabled account is flagged TEMP_GRACE_DAYS = None # no default: days after a temporary account's end date it may still be live TRANSFER_DAYS = None # no default: days after a transfer by which old-department access must go KNOWN_GAP_DAYS = {} # days you can explain, e.g. {"redshift": ["2026-10-03"]} # Databricks `accounts` audit actions (AWS and Azure audit log references): which ones are a sign-in, and which # ones change an account. The kinds apply to account-level rows (workspace_id 0); a workspace-level row of any of # these actions changes the account's place in one workspace, so it is listed as a modify of that workspace. DATABRICKS_SIGNIN_ACTIONS = ["aadBrowserLogin", "aadTokenLogin", "accountInHouseOAuthClientAuthentication", "accountLoginCodeAuthentication", "accountlessToAccountLoginAuthentication", "certLogin", "jwtLogin", "login", "mfaLogin", "multiFactorAuthenticationLogin", "oidcBrowserLogin", "oidcTokenAuthorization", "samlLogin", "tokenLogin", "workspaceLoginCodeAuthentication"] DATABRICKS_ACTION_KIND = {"add": "create", "activateUser": "enable", "deactivateUser": "disable", "delete": "remove", "updateUser": "modify", "setAdmin": "modify", "removeAdmin": "modify", "setAccountAdmin": "modify", "removeAccountAdmin": "modify", "addPrincipalToGroup": "modify", "addPrincipalsToGroup": "modify", "removePrincipalFromGroup": "modify", "removePrincipalsFromGroup": "modify"} # documented actions whose names look like user management or a sign-in but are not an account change we list DATABRICKS_OTHER_ACTIONS = ["createGroup", "removeGroup", "updateGroup", "changePassword", "resetPassword", "usernameDomainDenied", "validateEmail", "deleteUser", "changeServicePrincipalAcls", "multiFactorAuthenticationUpdateUserAuthPolicy"] REDSHIFT_SIGNIN_EVENTS = ["authenticated"] # SYS_CONNECTION_LOG events that count as a sign-in # ---------------------------------------------------------------------------------------------- FILES = { "snowflake_users": ["name", "disabled", "created_on", "deleted_on", "last_success_login", "expires_at"], "snowflake_grants_to_users": ["created_on", "deleted_on", "role", "grantee_name", "granted_by"], "snowflake_query_history": ["query_id", "query_text", "query_type", "user_name", "role_name", "start_time", "execution_status"], "databricks_users": ["id", "username", "active"], "databricks_group_members": ["group_name", "member_id"], "databricks_audit": ["event_time", "event_date", "workspace_id", "user_identity", "service_name", "action_name", "request_params", "response"], "redshift_svv_user_info": ["user_name", "user_id"], "redshift_svv_user_grants": ["user_name", "role_name"], "redshift_sys_userlog": ["user_id", "user_name", "original_user_name", "action", "record_time"], "redshift_sys_connection_log": ["event", "record_time", "user_name"], "hr_people": ["employee_email", "department"], "hr_transfers": ["employee_email", "transfer_date", "old_department", "new_department"], "login_map": ["platform", "login", "employee_email"], "temp_accounts": ["platform", "login", "end_date", "kind"], "role_departments": ["platform", "role_name", "department"], "change_tickets": ["ticket", "platform", "login", "action", "change_date"], } PLATFORM_FILES = { "snowflake": ["snowflake_users", "snowflake_grants_to_users", "snowflake_query_history"], "databricks": ["databricks_users", "databricks_group_members", "databricks_audit"], "redshift": ["redshift_svv_user_info", "redshift_svv_user_grants", "redshift_sys_userlog", "redshift_sys_connection_log"], } PLATFORMS = ["snowflake", "databricks", "redshift"] KINDS = ["create", "modify", "enable", "disable", "remove"] OUT_DIR = pathlib.Path("evidence-out") LIST_CSVS = ["1_inactive_but_enabled.csv", "2_temporary_accounts.csv", "3_transfers.csv", "4a_account_actions.csv", "4b_tickets_not_recorded.csv"] GEN_DIR = pathlib.Path("generated") 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}" RUN_AT = dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") def ts(col): """A timestamp 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 try_strptime(trim({col}), '%Y-%m-%d')::DATE END" def jv(col, key): """One key of a JSON text column, NULL when the text is not valid JSON.""" return f"CASE WHEN json_valid({col}) THEN json_extract_string({col}, '$.{key}') END" def status_ok(col="response"): return f"coalesce({jv(col, 'status_code')}, {jv(col, 'statusCode')}) = '200'" def blank(col): return f"nullif(trim({col}), '')" def quote_list(items): return ", ".join("'" + str(i).replace("'", "''") + "'" for i in items) # --------------------------------------------------------------------------------------------- # Generated demo records (seed 232). Everything below this block to "ENGINE" builds files only. # --------------------------------------------------------------------------------------------- def write_csv(folder, name, rows): with open(folder / f"{name}.csv", "w", newline="") as f: w = csv.writer(f) w.writerow(FILES[name]) w.writerows(rows) def generate(folder, plant, plant_input): rnd = random.Random(232) folder.mkdir(parents=True, exist_ok=True) d0, asof = dt.date(2026, 7, 1), dt.date(2026, 10, 5) days = [d0 + dt.timedelta(i) for i in range((asof - d0).days + 1)] D = lambda s: dt.date.fromisoformat(s) at = lambda d, h=9, m=0: f"{d} {h:02d}:{m:02d}:00" def schedule(first, last, always=False): out = [] for d in days: if not (D(first) <= d <= D(last)): continue if always or (d.weekday() < 5 and (rnd.random() < 0.7 or d >= D(last) - dt.timedelta(days=4))): out.append((d, 8 + rnd.randrange(9), rnd.randrange(60))) return out # ---- Snowflake sf_logins = {"ANA": ("2026-07-01", "2026-10-02"), "BEN": ("2026-07-01", "2026-10-02"), "CHEN": ("2026-07-01", "2026-10-02"), "DARA": ("2026-07-01", "2026-10-02"), "ADMIN_SAM": ("2026-07-01", "2026-10-02"), "ELI_TEMP": ("2026-09-01", "2026-09-18"), "SVC_LOAD": ("2026-07-01", "2026-10-05")} if plant: sf_logins["ELI_TEMP"] = ("2026-09-01", "2026-09-28") qh, qn, last_login = [], 0, {} for login, (a, b) in sf_logins.items(): for d, h, m in schedule(a, b, always=(login == "SVC_LOAD")): qn += 1 qh.append([f"q{qn:05d}", "SELECT 1", "SELECT", login, "ANALYST", at(d, h, m), "success"]) last_login[login] = at(d, h, m) def stmt(day_, text, who="ADMIN_SAM", status="success"): nonlocal qn qn += 1 qh.append([f"q{qn:05d}", text, "UNKNOWN", who, "SECURITYADMIN", at(D(day_), 10), status]) stmt("2026-09-01", "CREATE USER ELI_TEMP MUST_CHANGE_PASSWORD = TRUE COMMENT = 'contractor'") stmt("2026-09-10", "GRANT ROLE DATA_ENGINEER TO USER DARA") stmt("2026-09-12", "DROP USER GHOST", status="fail") if not plant: stmt("2026-08-03", "ALTER USER HAL SET DISABLED = TRUE") stmt("2026-09-10", '/* ticket T-105 */ REVOKE ROLE "FIN_ANALYST" FROM USER "DARA"') users = [[n, "false", "2026-01-15 09:00:00", "", last_login.get(n, ""), ""] for n in ["ANA", "BEN", "CHEN", "DARA", "ADMIN_SAM", "SVC_LOAD"]] users.append(["ELI_TEMP", "false", "2026-09-01 10:00:00", "", last_login["ELI_TEMP"], "" if plant else "2026-09-20 00:00:00"]) users.append(["HAL", "false" if plant else "true", "2025-11-03 09:00:00", "", "2026-05-14 08:00:00", ""]) grants = [["2026-01-15 09:00:00", "", "FIN_ANALYST", "ANA", "SECURITYADMIN"], ["2026-01-15 09:00:00", "", "SALES_READER", "BEN", "SECURITYADMIN"], ["2026-01-15 09:00:00", "", "DATA_ENGINEER", "CHEN", "SECURITYADMIN"], ["2026-09-10 10:00:00", "", "DATA_ENGINEER", "DARA", "SECURITYADMIN"], ["2026-02-01 09:00:00", "" if plant else "2026-09-10 10:00:00", "FIN_ANALYST", "DARA", "SECURITYADMIN"], ["2025-11-03 09:00:00", "", "FIN_ANALYST", "HAL", "SECURITYADMIN"], ["2026-01-15 09:00:00", "", "ACCOUNTADMIN", "ADMIN_SAM", "SECURITYADMIN"]] # ---- Databricks E = lambda n: f"{n}@example.test" db_logins = {"ana": ("2026-07-01", "2026-10-02"), "ben": ("2026-07-01", "2026-10-02"), "chen": ("2026-07-01", "2026-10-02"), "dara": ("2026-07-01", "2026-10-02"), "sam": ("2026-07-01", "2026-10-02"), "eli.contractor": ("2026-09-01", "2026-09-14"), "svc-etl": ("2026-07-01", "2026-10-05")} if plant: db_logins["eli.contractor"] = ("2026-09-01", "2026-09-22") audit, eid = [], 0 WS = "1234567890123456" # one workspace; account-level rows carry workspace_id 0 def aud(d, h, who, action, params, code="200", ws="0"): nonlocal eid eid += 1 import json audit.append([at(d, h, 0) if isinstance(d, dt.date) else d, str(d)[:10], ws, json.dumps({"email": E(who)}), "accounts", action, json.dumps(params), json.dumps({"status_code": int(code)}) if code.isdigit() else code]) for login, (a, b) in db_logins.items(): for d, h, m in schedule(a, b, always=(login == "svc-etl")): aud(f"{d} {h:02d}:{m:02d}:00", 0, login, "tokenLogin" if login == "svc-etl" else "login", {"user": E(login), "authenticationMethod": "saml"}, ws=WS) adm = lambda d, action, ws="0", **p: aud(f"{d} 10:00:00", 0, "sam", action, {**p, "endpoint": "accountsConsole"}, ws=ws) adm("2026-09-01", "add", targetUserName=E("eli.contractor")) adm("2026-09-01", "add", ws=WS, targetUserName=E("eli.contractor")) # added to the workspace: not a creation adm("2026-09-10", "addPrincipalToGroup", targetUserName=E("dara"), targetGroupName="data-engineers") if not plant: adm("2026-08-20", "deactivateUser", targetUserName=E("ivy")) adm("2026-09-10", "removePrincipalFromGroup", targetUserName=E("dara"), targetGroupName="finance-readers") adm("2026-09-15", "deactivateUser", targetUserName=E("eli.contractor")) else: # removed from the workspace only: the account itself stays active adm("2026-09-15", "delete", ws=WS, targetUserName=E("eli.contractor")) ids = {n: str(7001 + i) for i, n in enumerate(["ana", "ben", "chen", "dara", "sam", "svc-etl", "eli.contractor", "ivy"])} dbu = [[ids[n], E(n), "true"] for n in ["ana", "ben", "chen", "dara", "sam", "svc-etl"]] dbu += [[ids["eli.contractor"], E("eli.contractor"), "true" if plant else "false"], [ids["ivy"], E("ivy"), "true" if plant else "false"]] members = [["finance-readers", ids["ana"]], ["sales-readers", ids["ben"]], ["sales-readers", ids["ivy"]], ["data-engineers", ids["chen"]], ["data-engineers", ids["dara"]]] if plant: members.append(["finance-readers", ids["dara"]]) # ---- Redshift rs_logins = {"ana": ("2026-07-01", "2026-10-02"), "chen": ("2026-07-01", "2026-10-02"), "sam": ("2026-07-01", "2026-10-02"), "kim": ("2026-09-09", "2026-10-02"), "eli_temp": ("2026-09-01", "2026-09-14"), "fire_call": ("2026-09-11", "2026-09-12"), "svc_bi": ("2026-07-01", "2026-10-05")} if plant: rs_logins["fire_call"] = ("2026-09-11", "2026-09-20") conn = [] for login, (a, b) in rs_logins.items(): for d, h, m in schedule(a, b, always=(login == "svc_bi")): conn.append(["authenticated", f"{d} {h:02d}:{m:02d}:00", login]) conn.append(["authentication failure", "2026-09-05 03:00:00", "unknown"]) ulog = [[100, "eli_temp", "", "create", "2026-09-01 10:00:00"], [100, "ana", "", "alter", "2026-09-03 11:00:00"], [100, "kim", "", "create", "2026-09-08 10:00:00"], [100, "fire_call", "", "create", "2026-09-11 10:00:00"]] if plant: ulog.append([100, "eli_temp", "", "drop", "2026-09-30 10:00:00"]) else: ulog += [[100, "jon", "", "drop", "2026-09-02 10:00:00"], [100, "fire_call", "", "drop", "2026-09-13 10:00:00"], [100, "eli_temp", "", "drop", "2026-09-16 10:00:00"]] rsu = [[n, i] for i, n in enumerate(["ana", "chen", "kim", "sam", "svc_bi"], 101)] rsg = [["ana", "fin_ro"], ["chen", "data_rw"], ["kim", "data_rw"]] if plant: rsu += [["jon", 110], ["fire_call", 111]] rsg.append(["jon", "data_rw"]) # ---- lists you keep hr = [["ana@example.test", "Finance"], ["ben@example.test", "Sales"], ["chen@example.test", "Data"], ["dara@example.test", "Data"], ["eli@example.test", "Contractors"], ["hal@example.test", "Finance"], ["ivy@example.test", "Sales"], ["jon@example.test", "Data"], ["kim@example.test", "Data"], ["sam@example.test", "IT"]] transfers = [["dara@example.test", "2026-09-10", "Finance", "Data"]] lm = [["snowflake", n, E(e)] for n, e in [("ANA", "ana"), ("BEN", "ben"), ("CHEN", "chen"), ("DARA", "dara"), ("ELI_TEMP", "eli"), ("HAL", "hal"), ("ADMIN_SAM", "sam")]] lm += [["databricks", E(n), E(e)] for n, e in [("ana", "ana"), ("ben", "ben"), ("chen", "chen"), ("dara", "dara"), ("eli.contractor", "eli"), ("ivy", "ivy"), ("sam", "sam")]] lm += [["redshift", n, E(e)] for n, e in [("ana", "ana"), ("chen", "chen"), ("kim", "kim"), ("jon", "jon"), ("eli_temp", "eli"), ("fire_call", "sam"), ("sam", "sam")]] temp = [["snowflake", "ELI_TEMP", "2026-09-20", "temporary"], ["databricks", E("eli.contractor"), "2026-09-15", "temporary"], ["redshift", "eli_temp", "2026-09-15", "temporary"], ["redshift", "fire_call", "2026-09-12", "emergency"]] roles = [["snowflake", "FIN_ANALYST", "Finance"], ["snowflake", "DATA_ENGINEER", "Data"], ["snowflake", "SALES_READER", "Sales"], ["databricks", "finance-readers", "Finance"], ["databricks", "data-engineers", "Data"], ["databricks", "sales-readers", "Sales"], ["redshift", "fin_ro", "Finance"], ["redshift", "data_rw", "Data"]] tickets = [["T-101", "redshift", "kim", "create", "2026-09-08"], ["T-102", "redshift", "jon", "remove", "2026-09-02"], ["T-103", "databricks", E("eli.contractor"), "disable", "2026-09-15"], ["T-104", "snowflake", "ELI_TEMP", "create", "2026-09-01"], ["T-105", "snowflake", "DARA", "modify", "2026-09-10"], ["T-106", "redshift", "fire_call", "remove", "2026-09-13"], ["T-107", "redshift", "eli_temp", "remove", "2026-09-16"]] if plant: tickets.append(["T-108", "redshift", "lee", "create", "2026-09-20"]) if plant_input: users.append(["ANA", "false", "2026-01-15 09:00:00", "", last_login["ANA"], ""]) # duplicate login users[1][1] = "maybe" # BEN: DISABLED not true or false users[2][2] = "2026-13-45 09:00:00" # CHEN: created_on not a timestamp grants_header_fault = True temp[0][2] = "9/20/26" # end date not YYYY-MM-DD temp[1][3] = "tmp" # kind not temporary or emergency roles.append(["snowflake", "GHOST_ROLE", "Finance"]) # key matches no held role roles.append(["oracle", "DBA", "IT"]) # platform not known aud("2026-09-12 10:00:00", 0, "sam", "updateUserRole", {"targetUserName": E("ben")}) # looks like user management aud("2026-09-12 11:00:00", 0, "ben", "passkeyLogin", {"user": E("ben")}) # looks like a sign-in aud("2026-09-12 12:00:00", 0, "ben", "login", {"user": E("ben")}, code="not json") # response unreadable conn[:] = [c for c in conn if not c[1].startswith("2026-09-14")] # a day with no records qh.append(["q99999", "ALTER USER", "UNKNOWN", "ADMIN_SAM", "SECURITYADMIN", "2026-09-12 13:00:00", "success"]) ulog.append([100, "kim", "", "suspend", "2026-09-12 14:00:00"]) # action not known transfers.append(["zed@example.test", "2026-09-12", "Finance", "Data"]) # not in hr_people lm.append(["snowflake", "ANA", "someone.else@example.test"]) # login mapped to two people tickets.append(["T-109", "snowflake", "ANA", "rename", "2026-09-12"]) # action not known else: grants_header_fault = False write_csv(folder, "snowflake_users", users) write_csv(folder, "snowflake_query_history", qh) write_csv(folder, "databricks_users", dbu) write_csv(folder, "databricks_group_members", members) write_csv(folder, "databricks_audit", audit) write_csv(folder, "redshift_svv_user_info", rsu) write_csv(folder, "redshift_svv_user_grants", rsg) write_csv(folder, "redshift_sys_userlog", ulog) write_csv(folder, "redshift_sys_connection_log", conn) write_csv(folder, "hr_people", hr) write_csv(folder, "hr_transfers", transfers) write_csv(folder, "login_map", lm) write_csv(folder, "temp_accounts", temp) write_csv(folder, "role_departments", roles) write_csv(folder, "change_tickets", tickets) write_csv(folder, "snowflake_grants_to_users", grants) if grants_header_fault: # the export came with the column renamed: role_name -> role p = folder / "redshift_svv_user_grants.csv" p.write_text(p.read_text().replace("user_name,role_name", "user_name,role", 1)) EXPORT_SQL = """-- Snowflake (run each as a role that can read SNOWFLAKE.ACCOUNT_USAGE; set the session to UTC first) ALTER SESSION SET TIMEZONE = 'UTC'; SELECT name, disabled, created_on, deleted_on, last_success_login, expires_at FROM SNOWFLAKE.ACCOUNT_USAGE.USERS; SELECT created_on, deleted_on, role, grantee_name, granted_by FROM SNOWFLAKE.ACCOUNT_USAGE.GRANTS_TO_USERS; -- 120 days is a demo window: go back at least as far as your INACTIVE_DAYS SELECT query_id, query_text, query_type, user_name, role_name, start_time, execution_status FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY WHERE start_time >= DATEADD(day, -120, CURRENT_DATE); -- Databricks: system.access.audit (maps and structs written as JSON text); go back at least INACTIVE_DAYS. -- Workspace-level rows are kept in the workspace's region: run this in a workspace in each region you use -- and put all the rows in one file. SELECT event_time, event_date, workspace_id, to_json(user_identity) AS user_identity, service_name, action_name, to_json(request_params) AS request_params, to_json(response) AS response FROM system.access.audit WHERE service_name = 'accounts' AND event_date >= date_sub(current_date(), 120); -- databricks_users.csv (id, username, active): GET /api/2.0/accounts/{account_id}/scim/v2/Users -- -> id, userName, active, one row per user -- databricks_group_members.csv (group_name, member_id): GET .../scim/v2/Groups, then GET .../scim/v2/Groups/{id} -- for each group (the list no longer returns members) -> displayName and each member's value, one row per member -- Redshift (as a superuser, or a user who can see every row of these views) SELECT user_id, user_name, original_user_name, action, record_time FROM sys_userlog; SELECT event, record_time, user_name FROM sys_connection_log; SELECT user_name, user_id FROM svv_user_info; SELECT user_name, role_name FROM svv_user_grants; """ # --------------------------------------------------------------------------------------------- # ENGINE # --------------------------------------------------------------------------------------------- def fingerprint(rows): h = hashlib.sha256() for r in sorted("|".join("" if v is None else str(v) for v in row) for row in rows): h.update(r.encode() + b"\n") return h.hexdigest() class Report: def __init__(self): self.checks, self.follow_up = [], {} def check(self, name, ok, detail=""): self.checks.append((name, bool(ok), detail)) print(f" [{'ok' if ok else 'FAIL'}] {name}" + (f": {detail}" if detail else "")) return ok def main(): args = sys.argv[1:] global EXPORT_DIR, INACTIVE_DAYS, TEMP_GRACE_DAYS, TRANSFER_DAYS global OUT_DIR, GEN_DIR if "--out" in args: # write generated/ and evidence-out/ under this folder instead of the current one base = pathlib.Path(args[args.index("--out") + 1]) OUT_DIR, GEN_DIR = base / "evidence-out", base / "generated" if "--export-sql" in args: print(EXPORT_SQL) return 0 if "--demo" in args: plant, plant_input = "--plant" in args, "--plant-input" in args generate(GEN_DIR, plant, plant_input) EXPORT_DIR = str(GEN_DIR) INACTIVE_DAYS, TEMP_GRACE_DAYS, TRANSFER_DAYS = 90, 1, 14 if plant_input: TRANSFER_DAYS = None # an unset setting: list 3 must not run print(f"Generated records written to {GEN_DIR}/ (seed 232; not a company's data)." f" Demo settings: INACTIVE_DAYS=90, TEMP_GRACE_DAYS=1, TRANSFER_DAYS=14.") if EXPORT_DIR is None: print("Set EXPORT_DIR at the top of the script (or run with --demo).") return 2 rep = Report() con = duckdb.connect() folder = pathlib.Path(EXPORT_DIR) OUT_DIR.mkdir(parents=True, exist_ok=True) for name in LIST_CSVS: # evidence-out/ holds this run's lists only, never an earlier run's (OUT_DIR / name).unlink(missing_ok=True) print(f"\nRun at {RUN_AT}. Period {PERIOD_START} to {PERIOD_END} (UTC); exports taken {AS_OF}; " f"platforms exported: {', '.join(PLATFORMS_EXPORTED) or 'none'}; " f"not checked: {', '.join(p for p in PLATFORMS if p not in PLATFORMS_EXPORTED) or 'none'}.") print("\nInput checks") # --- settings problems = [] for nm, v in [("PERIOD_START", PERIOD_START), ("PERIOD_END", PERIOD_END), ("AS_OF", AS_OF)]: if not (isinstance(v, str) and re.fullmatch(DATE_RE, v)): problems.append(f"{nm} is not written as YYYY-MM-DD") if not problems: try: ps, pe, ao = (dt.date.fromisoformat(x) for x in (PERIOD_START, PERIOD_END, AS_OF)) if not ps <= pe <= ao: problems.append("need PERIOD_START <= PERIOD_END <= AS_OF") except ValueError: problems.append("a date is not a real calendar date") for p in PLATFORMS_EXPORTED: if p not in PLATFORMS: problems.append(f"PLATFORMS_EXPORTED has '{p}', not one of {PLATFORMS}") rep.check("dates and platforms in the settings", not problems, "; ".join(problems)) if problems: return 2 unset = [n for n, v in [("INACTIVE_DAYS", INACTIVE_DAYS), ("TEMP_GRACE_DAYS", TEMP_GRACE_DAYS), ("TRANSFER_DAYS", TRANSFER_DAYS)] if not (isinstance(v, int) and v >= 0)] rep.check("every time period is set (AC-2 leaves each one to you)", not unset, ("not set to a whole number of days: " + ", ".join(unset)) if unset else "") # --- load every file; missing file or column fails bad_files = [] for name, cols in FILES.items(): plat = next((p for p, fs in PLATFORM_FILES.items() if name in fs), None) path = folder / f"{name}.csv" have = [] if (plat is None or plat in PLATFORMS_EXPORTED) and 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() for h in have] missing = [c for c in cols if c not in lowered] if plat is not None and plat not in PLATFORMS_EXPORTED: have, missing = [], [] # not exported: an empty table, reported as NOT CHECKED elif not path.is_file(): bad_files.append(f"{name}.csv is missing") elif missing: bad_files.append(f"{name}.csv has no column {', '.join(missing)}") if have and not missing: sel = ", ".join(f'"{h}" AS "{h.strip().lower()}"' for h in have) 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(c + ' VARCHAR' for c in cols)})") rep.check("every file is there with every column the script reads", not bad_files, "; ".join(bad_files)) for name in FILES: n = con.execute(f"SELECT count(*) FROM t_{name}").fetchone()[0] print(f" read {n} rows into {name}") # --- unreadable values are counted, never skipped must_ts = [("snowflake_query_history", "start_time"), ("databricks_audit", "event_time"), ("redshift_sys_userlog", "record_time"), ("redshift_sys_connection_log", "record_time")] opt_ts = [("snowflake_users", c) for c in ("created_on", "deleted_on", "last_success_login", "expires_at")] + \ [("snowflake_grants_to_users", "created_on"), ("snowflake_grants_to_users", "deleted_on")] must_day = [("hr_transfers", "transfer_date"), ("temp_accounts", "end_date"), ("change_tickets", "change_date")] unreadable = [] for tbl, col in must_ts + opt_ts: need = (tbl, col) in must_ts n = con.execute(f"SELECT count(*) FROM t_{tbl} WHERE ({ts(col)}) IS NULL AND " f"{'TRUE' if need else blank(col) + ' IS NOT NULL'}").fetchone()[0] if n: unreadable.append(f"{tbl}.{col}: {n}") for tbl, col in must_day: n = con.execute(f"SELECT count(*) FROM t_{tbl} WHERE ({day(col)}) IS NULL").fetchone()[0] if n: unreadable.append(f"{tbl}.{col}: {n}") rep.check("every date and time is written as YYYY-MM-DD [HH:MM:SS] (UTC)", not unreadable, "not readable, so left out: " + ", ".join(unreadable) if unreadable else "") # --- values the script knows odd = [] for tbl, col in [("snowflake_users", "disabled"), ("databricks_users", "active")]: n = con.execute(f"SELECT count(*) FROM t_{tbl} WHERE lower(trim({col})) NOT IN ('true','false') " f"OR {col} IS NULL").fetchone()[0] if n: odd.append(f"{tbl}.{col} not true or false: {n}") for tbl, col in [("login_map", "platform"), ("temp_accounts", "platform"), ("role_departments", "platform"), ("change_tickets", "platform")]: n = con.execute(f"SELECT count(*) FROM t_{tbl} WHERE lower(trim({col})) NOT IN ({quote_list(PLATFORMS)}) " f"OR {col} IS NULL").fetchone()[0] if n: odd.append(f"{tbl}.{col} not one of {', '.join(PLATFORMS)}: {n}") n = con.execute(f"SELECT count(*) FROM t_temp_accounts WHERE lower(trim(kind)) NOT IN ('temporary','emergency') " f"OR kind IS NULL").fetchone()[0] if n: odd.append(f"temp_accounts.kind not temporary or emergency: {n}") n = con.execute(f"SELECT count(*) FROM t_change_tickets WHERE lower(trim(action)) NOT IN ({quote_list(KINDS)}) " f"OR action IS NULL").fetchone()[0] if n: odd.append(f"change_tickets.action not one of {', '.join(KINDS)}: {n}") for tbl, col in [("temp_accounts", "login"), ("hr_transfers", "employee_email"), ("change_tickets", "login"), ("databricks_users", "id"), ("databricks_group_members", "member_id")]: n = con.execute(f"SELECT count(*) FROM t_{tbl} WHERE {blank(col)} IS NULL").fetchone()[0] if n: odd.append(f"{tbl}.{col} blank: {n}") rep.check("every value is one the script knows", not odd, "; ".join(odd)) # --- Snowflake statements: read CREATE, ALTER and DROP USER (and role grants to a user) from the text def clean_sql(text): """One pass, left to right, so a quote inside a comment or a comment marker inside a literal is read as what it is: literals and comments (--, // and /* */) become a space. A double-quoted name keeps its text, unless it holds a character an unquoted name cannot (a space, say); then it becomes a mark no pattern reads, and the statement fails the input check below instead of being read as a different user.""" def one(m): if m.group(1) is None: return " " name = m.group(1).replace('""', '"') return name if re.fullmatch(r"[A-Za-z0-9_.@$\-]+", name) else "\x01" t = re.sub(r"'(?:[^']|'')*'|--[^\n]*|//[^\n]*|/\*.*?\*/|\"((?:[^\"]|\"\")*)\"", one, text or "", flags=re.S) return re.sub(r"\s+", " ", t).strip().rstrip("; ") # a statement may end with a semicolon NAMEP = r"([A-Za-z0-9_.@$\-]+)" pat = {"create": re.compile(r"^CREATE (OR REPLACE |OR ALTER )?USER (?:IF NOT EXISTS )?" + NAMEP + r"(?: (.*))?$", re.I), "alter": re.compile(r"^ALTER USER (?:IF EXISTS )?" + NAMEP + r"(?: (.*))?$", re.I), "drop": re.compile(r"^DROP USER (?:IF EXISTS )?" + NAMEP, re.I), "grant": re.compile(r"^GRANT ROLE \S+ TO USER " + NAMEP, re.I), # the role name is not kept "revoke": re.compile(r"^REVOKE ROLE \S+ FROM USER " + NAMEP, re.I)} # a statement that starts as a user change (after comments are stripped) but that the patterns do not read is # counted, never skipped; "CREATE USER" inside a privilege name or a procedure body is not a user change looks = re.compile(r"^((CREATE|ALTER|DROP)( OR (REPLACE|ALTER))? USER\b|(GRANT|REVOKE) ROLE\b.* (TO|FROM) USER\b)", re.I) # an unreadable quoted name where the user's name goes; a quoted option value later in the statement is fine name_unread = re.compile(r"^((CREATE( OR (REPLACE|ALTER))?|ALTER|DROP) USER (IF (NOT )?EXISTS )?\x01" r"|(GRANT|REVOKE) ROLE \S+ (TO|FROM) USER \x01)", re.I) # the words an ALTER USER can start with when it names no user (Snowflake's ALTER USER syntax) ALTER_NO_NAME = ("SET", "UNSET", "RESET", "RENAME", "ADD", "REMOVE", "MODIFY", "ROTATE", "ABORT", "ENROLL") con.execute("CREATE TEMP TABLE sf_actions (platform VARCHAR, t TIMESTAMP, actor VARCHAR, target VARCHAR, " "kind VARCHAR, source VARCHAR)") unread, failed_user_stmts, rows = 0, 0, [] src = con.execute(f"SELECT lower(trim(user_name)), ({ts('start_time')}), lower(trim(execution_status)), " f"query_text, query_id FROM t_snowflake_query_history").fetchall() for who, t, status, text, qid in src: s = clean_sql(text) kind, target = None, None if name_unread.match(s): # a quoted name no pattern can read: never let IF be taken as the name unread += status == "success" continue if m := pat["create"].match(s): kind, target = "create", m.group(2) if m.group(1) and m.group(1).upper().startswith("OR ALTER"): # creates the user or changes it d = re.search(r"\bDISABLED\s*=\s*(TRUE|FALSE)\b", m.group(3) or "", re.I) kind = ("disable" if d.group(1).upper() == "TRUE" else "enable") if d else "modify" elif m := pat["drop"].match(s): kind, target = "remove", m.group(1) elif m := pat["alter"].match(s): target, rest = m.group(1), m.group(2) or "" if target.upper() in ALTER_NO_NAME: # no name: Snowflake alters the logged-in user target, rest = who, (target + " " + rest).strip() if re.match(r"ABORT ALL QUERIES\b", rest, re.I): # stops queries; the account is not changed continue d = re.search(r"\bDISABLED\s*=\s*(TRUE|FALSE)\b", rest, re.I) kind = ("disable" if d.group(1).upper() == "TRUE" else "enable") if d else "modify" elif m := (pat["grant"].match(s) or pat["revoke"].match(s)): kind, target = "modify", m.group(1) elif looks.match(s): if status == "success": unread += 1 continue else: continue if status != "success": failed_user_stmts += 1 continue rows.append(("snowflake", t, who, target.lower(), kind, "query " + qid + (" (create or alter)" if s.upper().startswith("CREATE OR ALTER") else ""))) con.executemany("INSERT INTO sf_actions VALUES (?,?,?,?,?,?)", rows) rep.check("every successful CREATE, ALTER and DROP USER statement and role grant to a user could be read", unread == 0, f"{unread} successful statement(s) look like one but name no user the script can read" if unread else f"{failed_user_stmts} failed user statement(s) not counted" if failed_user_stmts else "") # --- Databricks audit: actions the script does not know known = DATABRICKS_SIGNIN_ACTIONS + list(DATABRICKS_ACTION_KIND) + DATABRICKS_OTHER_ACTIONS unk_mgmt = con.execute(f"SELECT DISTINCT trim(action_name) FROM t_databricks_audit WHERE lower(trim(service_name))=" f"'accounts' AND trim(action_name) NOT IN ({quote_list(known)}) AND " f"regexp_matches(action_name, '(?i)(user|principal|admin|group)') " f"AND NOT regexp_matches(action_name, '(?i)(login|authenticat)') ORDER BY 1").fetchall() unk_sign = con.execute(f"SELECT DISTINCT trim(action_name) FROM t_databricks_audit WHERE lower(trim(service_name))=" f"'accounts' AND trim(action_name) NOT IN ({quote_list(known)}) AND " f"regexp_matches(action_name, '(?i)(login|authenticat)') ORDER BY 1").fetchall() rep.check("every Databricks action that looks like user management is one the script lists", not unk_mgmt, "unknown: " + ", ".join(r[0] for r in unk_mgmt) if unk_mgmt else "") rep.check("every Databricks action that looks like a sign-in is on the sign-in list", not unk_sign, "unknown: " + ", ".join(r[0] for r in unk_sign) if unk_sign else "") inscope = quote_list(DATABRICKS_SIGNIN_ACTIONS + list(DATABRICKS_ACTION_KIND)) nstat = con.execute(f"SELECT count(*) FROM t_databricks_audit WHERE lower(trim(service_name))='accounts' AND " f"trim(action_name) IN ({inscope}) AND coalesce({jv('response','status_code')}, " f"{jv('response','statusCode')}) IS NULL").fetchone()[0] nws = con.execute(f"SELECT count(*) FROM t_databricks_audit WHERE lower(trim(service_name))='accounts' AND " f"trim(action_name) IN ({inscope}) AND {blank('workspace_id')} IS NULL").fetchone()[0] rep.check("every Databricks sign-in and account action has a readable response status and workspace_id", nstat == 0 and nws == 0, "; ".join(x for x in [f"{nstat} row(s) whose response is not JSON with a status code" if nstat else "", f"{nws} row(s) with no workspace_id" if nws else ""] if x)) ulog_unknown = con.execute("SELECT DISTINCT lower(trim(action)) FROM t_redshift_sys_userlog WHERE " "lower(trim(action)) NOT IN ('alter','create','drop','rename') OR action IS NULL " "ORDER BY 1").fetchall() rep.check("every SYS_USERLOG action is alter, create, drop or rename", not ulog_unknown, "unknown: " + ", ".join(str(r[0]) for r in ulog_unknown) if ulog_unknown else "") # --- normalize: accounts, sign-ins, holdings, actions asof_end = f"TIMESTAMP '{AS_OF} 00:00:00' + INTERVAL 1 DAY" con.execute(f"""CREATE TEMP TABLE signins AS SELECT 'snowflake' AS platform, lower(trim(user_name)) AS login, ({ts('start_time')}) AS t FROM t_snowflake_query_history WHERE lower(trim(execution_status)) = 'success' AND {blank('user_name')} IS NOT NULL UNION ALL SELECT 'snowflake', lower(trim(name)), ({ts('last_success_login')}) FROM t_snowflake_users WHERE {blank('last_success_login')} IS NOT NULL UNION ALL SELECT 'databricks', lower(trim(coalesce({blank(jv('request_params', 'user'))}, {jv('user_identity', 'email')}))), ({ts('event_time')}) FROM t_databricks_audit WHERE lower(trim(service_name)) = 'accounts' AND trim(action_name) IN ({quote_list(DATABRICKS_SIGNIN_ACTIONS)}) AND {status_ok()} UNION ALL SELECT 'redshift', lower(trim(user_name)), ({ts('record_time')}) FROM t_redshift_sys_connection_log WHERE trim(event) IN ({quote_list(REDSHIFT_SIGNIN_EVENTS)})""") con.execute("CREATE TEMP TABLE db_kind (action VARCHAR, kind VARCHAR)") con.executemany("INSERT INTO db_kind VALUES (?,?)", list(DATABRICKS_ACTION_KIND.items())) con.execute(f"""CREATE TEMP TABLE actions AS SELECT * FROM sf_actions UNION ALL SELECT 'databricks', ({ts('a.event_time')}), lower(trim({jv('a.user_identity', 'email')})), lower(trim({jv('a.request_params', 'targetUserName')})), CASE WHEN trim(a.workspace_id) = '0' THEN k.kind ELSE 'modify' END, 'audit ' || trim(a.action_name) || CASE WHEN trim(a.workspace_id) = '0' THEN '' ELSE ' (workspace ' || trim(a.workspace_id) || ')' END FROM t_databricks_audit a JOIN db_kind k ON k.action = trim(a.action_name) WHERE lower(trim(a.service_name)) = 'accounts' AND {status_ok('a.response')} AND {blank('a.workspace_id')} IS NOT NULL UNION ALL SELECT 'redshift', ({ts('record_time')}), 'user_id ' || trim(user_id), lower(trim(user_name)), CASE lower(trim(action)) WHEN 'create' THEN 'create' WHEN 'drop' THEN 'remove' ELSE 'modify' END, 'sys_userlog ' || lower(trim(action)) || CASE WHEN {blank('original_user_name')} IS NOT NULL THEN ' (was ' || lower(trim(original_user_name)) || ')' ELSE '' END FROM t_redshift_sys_userlog WHERE lower(trim(action)) IN ('alter','create','drop','rename')""") # A user added from a workspace may have no account-level add in the records. That workspace add stays a modify # (it is also what adding an existing, idle user to a new workspace looks like), but a create ticket may match it. con.execute("ALTER TABLE actions ADD COLUMN ticket_kind VARCHAR") con.execute("""UPDATE actions SET ticket_kind = CASE WHEN platform = 'databricks' AND source LIKE 'audit add (workspace %' AND NOT EXISTS (SELECT 1 FROM actions y WHERE y.platform = 'databricks' AND y.target = actions.target AND y.kind = 'create') THEN 'create' WHEN platform = 'snowflake' AND source LIKE '% (create or alter)' THEN 'create' ELSE kind END""") # Each login's current incarnation starts at the first create after the last remove that precedes its latest # create: a later CREATE USER IF NOT EXISTS (a no-op on an existing user) never moves it, and an earlier # holder of a reused name never counts. con.execute("""CREATE TEMP TABLE since AS WITH lc AS (SELECT platform, target, max(t) AS last_create FROM actions WHERE kind = 'create' OR (platform = 'snowflake' AND ticket_kind = 'create') GROUP BY 1, 2), lr AS (SELECT lc.platform, lc.target, lc.last_create, (SELECT max(r.t) FROM actions r WHERE r.platform = lc.platform AND r.target = lc.target AND r.kind = 'remove' AND r.t < lc.last_create) AS prior_remove FROM lc) SELECT lr.platform, lr.target, (SELECT min(c.t) FROM actions c WHERE c.platform = lr.platform AND c.target = lr.target AND (c.kind = 'create' OR (c.platform = 'snowflake' AND c.ticket_kind = 'create')) AND (lr.prior_remove IS NULL OR c.t > lr.prior_remove)) AS since FROM lr""") con.execute(f"""CREATE TEMP TABLE accounts AS SELECT 'snowflake' AS platform, lower(trim(name)) AS login, (lower(trim(disabled)) <> 'true' AND (({ts('expires_at')}) IS NULL OR ({ts('expires_at')}) > {asof_end})) AS enabled, ({ts('created_on')}) AS created_on, ({ts('expires_at')}) AS expires_at FROM t_snowflake_users WHERE {blank('deleted_on')} IS NULL UNION ALL SELECT 'databricks', lower(trim(username)), lower(trim(active)) = 'true', NULL, NULL FROM t_databricks_users UNION ALL SELECT 'redshift', lower(trim(user_name)), TRUE, NULL, NULL FROM t_redshift_svv_user_info""") con.execute("""CREATE TEMP TABLE accounts2 AS SELECT a.platform, a.login, a.enabled, a.expires_at, coalesce(a.created_on, (SELECT n.since FROM since n WHERE n.platform = a.platform AND n.target = a.login)) AS created_on, (SELECT max(s.t) FROM signins s WHERE s.platform = a.platform AND s.login = a.login) AS last_signin, (SELECT min(x.t) FROM actions x WHERE x.platform = 'databricks' AND a.platform = 'databricks' AND x.target = a.login AND x.kind = 'modify' AND x.ticket_kind = 'create') AS ws_added FROM accounts a""") con.execute(f"""CREATE TEMP TABLE holdings AS SELECT 'snowflake' AS platform, lower(trim(grantee_name)) AS login, lower(trim(role)) AS role_name FROM t_snowflake_grants_to_users WHERE {blank('deleted_on')} IS NULL UNION SELECT 'databricks', lower(trim(u.username)), lower(trim(g.group_name)) FROM t_databricks_group_members g JOIN t_databricks_users u ON trim(u.id) = trim(g.member_id) UNION SELECT 'redshift', lower(trim(user_name)), lower(trim(role_name)) FROM t_redshift_svv_user_grants""") con.execute(f"""CREATE TEMP TABLE ended_sf AS SELECT lower(trim(name)) AS login, max(({ts('deleted_on')})::DATE) AS d FROM t_snowflake_users WHERE {blank('deleted_on')} IS NOT NULL GROUP BY 1""") # --- keys and mappings dups = con.execute("SELECT platform, login, count(*) FROM accounts GROUP BY 1, 2 HAVING count(*) > 1 " "ORDER BY 1, 2").fetchall() rep.check("no login appears twice in one platform's account list", not dups, "; ".join(f"{p}/{l} x{n}" for p, l, n in dups)) conflicts = con.execute("SELECT lower(trim(platform)), lower(trim(login)), count(DISTINCT lower(trim(employee_email))) " "FROM t_login_map GROUP BY 1, 2 HAVING count(DISTINCT lower(trim(employee_email))) > 1 " "ORDER BY 1, 2").fetchall() rep.check("no login is mapped to two people", not conflicts, "; ".join(f"{p}/{l}" for p, l, _ in conflicts)) nokey = con.execute(f"""SELECT lower(trim(platform)), lower(trim(role_name)) FROM t_role_departments r WHERE lower(trim(platform)) IN ({quote_list(PLATFORMS_EXPORTED)}) AND NOT EXISTS (SELECT 1 FROM holdings h WHERE h.platform = lower(trim(r.platform)) AND h.role_name = lower(trim(r.role_name))) ORDER BY 1, 2""").fetchall() rep.check("every role or group in the department map is held by someone", not nokey, "matches no held role: " + ", ".join(f"{p}/{r}" for p, r in nokey) if nokey else "") nohr = con.execute("SELECT DISTINCT coalesce(lower(trim(t.employee_email)), '(blank)') FROM t_hr_transfers t WHERE " "NOT EXISTS (SELECT 1 FROM t_hr_people p WHERE lower(trim(p.employee_email)) = " "lower(trim(t.employee_email))) ORDER BY 1").fetchall() rep.check("every transfer names someone in the HR people list", not nohr, "not in hr_people: " + ", ".join(r[0] for r in nohr) if nohr else "") if "databricks" in PLATFORMS_EXPORTED: nm = con.execute("SELECT count(*) FROM t_databricks_group_members g WHERE NOT EXISTS (SELECT 1 FROM " "t_databricks_users u WHERE trim(u.id) = trim(g.member_id))").fetchone()[0] print(f" Databricks group members that are not account users (groups or service principals; " f"not expanded): {nm}") # --- completeness: a record for every day the lists depend on, per platform span_start = PERIOD_START if isinstance(INACTIVE_DAYS, int) and INACTIVE_DAYS >= 0: span_start = min(PERIOD_START, str(dt.date.fromisoformat(AS_OF) - dt.timedelta(days=INACTIVE_DAYS))) day_src = {"snowflake": f"SELECT ({ts('start_time')})::DATE AS d FROM t_snowflake_query_history", "databricks": f"SELECT ({ts('event_time')})::DATE AS d FROM t_databricks_audit " f"WHERE lower(trim(service_name)) = 'accounts'", "redshift": f"SELECT ({ts('record_time')})::DATE AS d FROM t_redshift_sys_connection_log"} for p in PLATFORMS_EXPORTED: gaps = [str(r[0]) for r in con.execute(f"""SELECT g.d::DATE FROM generate_series(DATE '{span_start}', DATE '{AS_OF}', INTERVAL 1 DAY) g(d) WHERE g.d::DATE NOT IN (SELECT d FROM ({day_src[p]}) WHERE d IS NOT NULL) ORDER BY 1""").fetchall()] gaps = [g for g in gaps if g not in KNOWN_GAP_DAYS.get(p, [])] rep.check(f"{p}: records exist for every day from {span_start} to {AS_OF}", not gaps, f"{len(gaps)} day(s) with none, first {gaps[0]}, last {gaps[-1]}" if gaps else "") failed = [n for n, ok, _ in rep.checks if not ok] flag = "" if not failed else f" ** INCOMPLETE: {len(failed)} input check(s) failed; read the lists with that in mind **" print(f"\n{len(rep.checks)} input checks, {len(failed)} failed.{flag}") def show(title, cols, rows, period, filters, csv_name): print(f"\n== {title}") print(f" period: {period}; filters: {filters}; run at {RUN_AT}; rows: {len(rows)}; " f"sha256: {fingerprint(rows)}" + ("; INCOMPLETE" if failed else "")) if rows: w = [max(len(str(c)), *(len(str(r[i])) for r in rows)) for i, c in enumerate(cols)] print(" " + " | ".join(str(c).ljust(w[i]) for i, c in enumerate(cols))) for r in rows: print(" " + " | ".join(("" if v is None else str(v)).ljust(w[i]) for i, v in enumerate(r))) with open(OUT_DIR / csv_name, "w", newline="") as f: cw = csv.writer(f) cw.writerow(cols) cw.writerows([["" if v is None else v for v in r] for r in rows]) pl = quote_list(PLATFORMS_EXPORTED) need = lambda n: f"NOT RUN: {n} is not set" def not_run(title, setting, csv_name): print(f"\n== {title}\n " + need(setting)) with open(OUT_DIR / csv_name, "w", newline="") as f: csv.writer(f).writerows([["status"], [need(setting) + f"; run at {RUN_AT}"]]) # --- 1. inactive but enabled if isinstance(INACTIVE_DAYS, int) and INACTIVE_DAYS >= 0: rows = con.execute(f"""SELECT platform, login, coalesce(last_signin, created_on, ws_added)::VARCHAR AS basis_time, CASE WHEN last_signin IS NOT NULL THEN 'last sign-in' WHEN created_on IS NOT NULL THEN 'created' WHEN ws_added IS NOT NULL THEN 'no sign-in; added to a workspace, no account-level add' ELSE 'no sign-in in the records' END AS basis, date_diff('day', coalesce(last_signin, created_on, ws_added)::DATE, DATE '{AS_OF}') AS days_idle, CASE WHEN coalesce(last_signin, created_on) IS NULL OR date_diff('day', coalesce(last_signin, created_on)::DATE, DATE '{AS_OF}') >= {INACTIVE_DAYS} THEN 'INACTIVE BUT ENABLED' ELSE 'ok' END AS status FROM accounts2 WHERE enabled AND platform IN ({pl}) ORDER BY platform, login""").fetchall() show("1. Inactive but enabled (AC-2(3)(d))", ["platform", "login", "basis_time", "basis", "days_idle", "status"], rows, f"state on {AS_OF}", f"enabled accounts on {', '.join(PLATFORMS_EXPORTED)}; INACTIVE_DAYS={INACTIVE_DAYS}", "1_inactive_but_enabled.csv") rep.follow_up["inactive"] = sum(r[5] != "ok" for r in rows) else: not_run("1. Inactive but enabled (AC-2(3)(d))", "INACTIVE_DAYS", "1_inactive_but_enabled.csv") # --- 2. temporary and emergency accounts if isinstance(TEMP_GRACE_DAYS, int) and TEMP_GRACE_DAYS >= 0: rows = con.execute(f"""WITH pop AS ( SELECT lower(trim(platform)) AS platform, lower(trim(login)) AS login, ({day('end_date')}) AS end_date, lower(trim(kind)) AS kind, 'your list' AS source FROM t_temp_accounts UNION ALL SELECT platform, login, expires_at::DATE, 'temporary', 'Snowflake EXPIRES_AT' FROM accounts2 x WHERE platform = 'snowflake' AND expires_at IS NOT NULL AND NOT EXISTS (SELECT 1 FROM t_temp_accounts l WHERE lower(trim(l.platform)) = 'snowflake' AND lower(trim(l.login)) = x.login)) SELECT p.platform, p.login, p.kind, p.end_date::VARCHAR AS end_date, p.source, coalesce(a.enabled::VARCHAR, 'not on platform') AS enabled_now, (SELECT max(s.t)::VARCHAR FROM signins s WHERE s.platform = p.platform AND s.login = p.login) AS last_use, least(e.d, a.expires_at::DATE, CASE WHEN a.login IS NULL THEN sd.d END)::VARCHAR AS ended_on, CASE WHEN EXISTS (SELECT 1 FROM signins s WHERE s.platform = p.platform AND s.login = p.login AND s.t::DATE > p.end_date + {TEMP_GRACE_DAYS}) THEN 'USED AFTER END' WHEN a.enabled THEN CASE WHEN DATE '{AS_OF}' > p.end_date + {TEMP_GRACE_DAYS} THEN 'STILL ENABLED PAST END' ELSE 'NOT YET DUE' END WHEN least(e.d, a.expires_at::DATE, CASE WHEN a.login IS NULL THEN sd.d END) > p.end_date + {TEMP_GRACE_DAYS} THEN 'ENDED LATE' WHEN least(e.d, a.expires_at::DATE, CASE WHEN a.login IS NULL THEN sd.d END) IS NOT NULL THEN 'ended on time' WHEN a.login IS NULL THEN 'NOT FOUND ON PLATFORM' ELSE 'ENDED, DATE NOT RECORDED' END AS status FROM pop p LEFT JOIN accounts2 a ON a.platform = p.platform AND a.login = p.login LEFT JOIN (SELECT x.platform, x.target, min(x.t)::DATE AS d FROM actions x LEFT JOIN accounts2 c ON c.platform = x.platform AND c.login = x.target WHERE x.kind IN ('remove', 'disable') AND x.t >= coalesce(c.created_on, (SELECT n.since FROM since n WHERE n.platform = x.platform AND n.target = x.target), TIMESTAMP '1900-01-01') GROUP BY 1, 2) e ON e.platform = p.platform AND e.target = p.login LEFT JOIN ended_sf sd ON p.platform = 'snowflake' AND sd.login = p.login WHERE p.platform IN ({pl}) AND p.end_date <= DATE '{PERIOD_END}' ORDER BY 1, 2""").fetchall() show("2. Temporary and emergency accounts past their end (AC-2(2), AC-2(3)(a))", ["platform", "login", "kind", "end_date", "source", "enabled_now", "last_use", "ended_on", "status"], rows, f"end dates up to {PERIOD_END}; state on {AS_OF}", f"your temp_accounts list plus Snowflake EXPIRES_AT; TEMP_GRACE_DAYS={TEMP_GRACE_DAYS}", "2_temporary_accounts.csv") rep.follow_up["temporary"] = sum(r[8] in ("USED AFTER END", "STILL ENABLED PAST END", "ENDED LATE", "NOT FOUND ON PLATFORM", "ENDED, DATE NOT RECORDED") for r in rows) else: not_run("2. Temporary and emergency accounts past their end (AC-2(2), AC-2(3)(a))", "TEMP_GRACE_DAYS", "2_temporary_accounts.csv") # --- 3. transfers that kept the old access if isinstance(TRANSFER_DAYS, int) and TRANSFER_DAYS >= 0: rows = con.execute(f"""WITH t AS (SELECT lower(trim(employee_email)) AS email, ({day('transfer_date')}) AS tdate, lower(trim(old_department)) AS old_dept, lower(trim(new_department)) AS new_dept, trim(old_department) AS old_name FROM t_hr_transfers), tp AS (SELECT t.*, p.platform FROM t, (SELECT unnest([{pl}]) AS platform) p WHERE t.tdate BETWEEN DATE '{PERIOD_START}' AND DATE '{PERIOD_END}') SELECT tp.email, tp.tdate::VARCHAR AS transfer_date, tp.platform, coalesce((SELECT string_agg(DISTINCT lower(trim(m.login)), ', ' ORDER BY lower(trim(m.login))) FROM t_login_map m WHERE lower(trim(m.platform)) = tp.platform AND lower(trim(m.employee_email)) = tp.email), '') AS logins, coalesce((SELECT string_agg(DISTINCT h.role_name, ', ' ORDER BY h.role_name) FROM t_login_map m JOIN holdings h ON h.platform = tp.platform AND h.login = lower(trim(m.login)) JOIN t_role_departments r ON lower(trim(r.platform)) = tp.platform AND lower(trim(r.role_name)) = h.role_name WHERE lower(trim(m.platform)) = tp.platform AND lower(trim(m.employee_email)) = tp.email AND lower(trim(r.department)) = tp.old_dept AND NOT EXISTS (SELECT 1 FROM t_role_departments r2 WHERE lower(trim(r2.platform)) = tp.platform AND lower(trim(r2.role_name)) = h.role_name AND lower(trim(r2.department)) = tp.new_dept)), '') AS old_access_held, CASE WHEN NOT EXISTS (SELECT 1 FROM t_login_map m WHERE lower(trim(m.platform)) = tp.platform AND lower(trim(m.employee_email)) = tp.email) THEN 'no login on this platform' WHEN tp.tdate + {TRANSFER_DAYS} > DATE '{AS_OF}' THEN 'NOT YET DUE' WHEN EXISTS (SELECT 1 FROM t_login_map m JOIN holdings h ON h.platform = tp.platform AND h.login = lower(trim(m.login)) JOIN t_role_departments r ON lower(trim(r.platform)) = tp.platform AND lower(trim(r.role_name)) = h.role_name WHERE lower(trim(m.platform)) = tp.platform AND lower(trim(m.employee_email)) = tp.email AND lower(trim(r.department)) = tp.old_dept AND NOT EXISTS (SELECT 1 FROM t_role_departments r2 WHERE lower(trim(r2.platform)) = tp.platform AND lower(trim(r2.role_name)) = h.role_name AND lower(trim(r2.department)) = tp.new_dept)) THEN 'KEPT OLD ACCESS' ELSE 'clean' END AS status FROM tp ORDER BY 1, 2, 3""").fetchall() show("3. Transfers that kept the old access (AC-2 h.2 and l)", ["employee", "transfer_date", "platform", "logins", "old_access_held", "status"], rows, f"transfers dated {PERIOD_START} to {PERIOD_END}; holdings on {AS_OF}", f"roles and groups your department map ties to the old department; TRANSFER_DAYS={TRANSFER_DAYS}", "3_transfers.csv") rep.follow_up["transfers"] = sum(r[5] == "KEPT OLD ACCESS" for r in rows) un = con.execute(f"""SELECT count(*) FROM (SELECT DISTINCT h.platform, h.login, h.role_name FROM (SELECT lower(trim(employee_email)) AS email FROM t_hr_transfers WHERE ({day('transfer_date')}) BETWEEN DATE '{PERIOD_START}' AND DATE '{PERIOD_END}') t JOIN t_login_map m ON lower(trim(m.employee_email)) = t.email JOIN holdings h ON h.platform = lower(trim(m.platform)) AND h.login = lower(trim(m.login)) WHERE h.platform IN ({pl}) AND NOT EXISTS (SELECT 1 FROM t_role_departments r WHERE lower(trim(r.platform)) = h.platform AND lower(trim(r.role_name)) = h.role_name))""").fetchone()[0] print(f" held roles and groups of transferred people that your department map does not list: {un}") else: not_run("3. Transfers that kept the old access (AC-2 h.2 and l)", "TRANSFER_DAYS", "3_transfers.csv") # --- 4. account actions acts = con.execute(f"""SELECT a.platform, a.t::VARCHAR AS at_utc, a.kind, a.target, coalesce({blank('a.actor')}, '(not in the record)') AS actor, a.source, coalesce((SELECT string_agg(k.ticket, ', ' ORDER BY k.ticket) FROM t_change_tickets k WHERE lower(trim(k.platform)) = a.platform AND lower(trim(k.login)) = a.target AND lower(trim(k.action)) IN (a.kind, a.ticket_kind) AND ({day('k.change_date')}) = a.t::DATE), 'no ticket') AS ticket FROM actions a WHERE a.t::DATE BETWEEN DATE '{PERIOD_START}' AND DATE '{PERIOD_END}' AND a.platform IN ({pl}) ORDER BY a.platform, a.t, a.kind, a.target, a.source""").fetchall() show("4a. Account actions the platforms recorded (AC-2(4))", ["platform", "at_utc", "kind", "target", "actor", "source", "ticket"], acts, f"{PERIOD_START} to {PERIOD_END} (UTC)", "every create, modify, enable, disable and remove; successful only", "4a_account_actions.csv") by = con.execute("SELECT platform, kind, count(*) FROM (SELECT * FROM actions WHERE t::DATE BETWEEN DATE '" + PERIOD_START + "' AND DATE '" + PERIOD_END + "' AND platform IN (" + pl + ")) GROUP BY 1, 2 ORDER BY 1, 2").fetchall() print(" counts: " + "; ".join(f"{p} {k} {n}" for p, k, n in by)) miss = con.execute(f"""SELECT k.ticket, lower(trim(k.platform)) AS platform, lower(trim(k.login)) AS login, lower(trim(k.action)) AS action, ({day('k.change_date')})::VARCHAR AS change_date, 'NOT RECORDED' AS status FROM t_change_tickets k WHERE ({day('k.change_date')}) BETWEEN DATE '{PERIOD_START}' AND DATE '{PERIOD_END}' AND lower(trim(k.platform)) IN ({pl}) AND lower(trim(k.action)) IN ({quote_list(KINDS)}) AND NOT EXISTS (SELECT 1 FROM actions a WHERE a.platform = lower(trim(k.platform)) AND a.target = lower(trim(k.login)) AND lower(trim(k.action)) IN (a.kind, a.ticket_kind) AND a.t::DATE = ({day('k.change_date')})) ORDER BY 1""").fetchall() show("4b. Change tickets with no recorded action (AC-2(4))", ["ticket", "platform", "login", "action", "change_date", "status"], miss, f"{PERIOD_START} to {PERIOD_END} (UTC)", "your change_tickets, matched on platform, login, action and day", "4b_tickets_not_recorded.csv") rep.follow_up["not_recorded"] = len(miss) nt = sum(1 for r in acts if r[6] == "no ticket") print(f" recorded actions with no ticket: {nt} of {len(acts)} (a count to look at, not a finding)") total = sum(rep.follow_up.values()) parts = [f"{k.replace('_', ' ')} {rep.follow_up[k]}" if k in rep.follow_up else f"{k.replace('_', ' ')} not run" for k in ["inactive", "temporary", "transfers", "not_recorded"]] print(f"\nNeeds follow-up: {', '.join(parts)}; input checks failed: {len(failed)}. " "The script produces evidence; it does not say a control is met.") if "--strict" in args and (total or failed): return 1 return 0 if __name__ == "__main__": sys.exit(main())