#!/usr/bin/env python3 """Deletion and masking evidence from Snowflake, Databricks and Redshift, in DuckDB. python3 deletion_masking_evidence.py --demo # clean generated records python3 deletion_masking_evidence.py --demo --plant # generated records with planted findings python3 deletion_masking_evidence.py --demo --plant-input # generated records with planted input faults python3 deletion_masking_evidence.py --queries # print the row-age query for every table in YOUR register python3 deletion_masking_evidence.py --demo --queries # the same for the generated register python3 deletion_masking_evidence.py # your exports: set the settings below first python3 deletion_masking_evidence.py --strict # exit 1 unless nothing needs follow-up You keep two lists: a retention register (table, data category, retention days, the date column that ages a row, and how many days your policy lets deleted data stay recoverable) and a masking policy (data category and the roles that may see it in the clear). The script reads each platform's own records, exported as CSV with the platform's published column names, and prints three lists, each with its period, filters, run time, row count and a SHA-256 of its rows: 1. rows past retention: any register table with rows older than its retention (PAST RETENTION), or with no row-age export (NOT MEASURED) 2. deleted but still recoverable: how long deleted data stays recoverable on that platform, against what your register allows (RECOVERABLE BEYOND POLICY), and a Redshift manual snapshot kept indefinitely that was taken before a delete (INDEFINITE SNAPSHOT, flagged whatever the register says) 3. masking against your policy: every column tagged with a category your masking policy covers must have a mask the platform lists on it (Snowflake also reports a status; the script counts only ACTIVE), and any role you list as reading it in the clear must be on the policy's list (NOT MASKED, NOT MASKED FOR PUBLIC for a Redshift mask attached only to named roles or users, MASK NOT ACTIVE, CLEAR READER NOT ON POLICY LIST) 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. The script reads files; it never connects to a platform and never writes to one. Input checks: --demo --plant-input plants one fault for each of the 26, and every one reports FAIL, for example a table twice in the register, a masking-policy category ('financial') with no tagged column, a row-age line for a table not in the register ('sales.old_staging'), row ages made with 365 days against the register's 730, a Databricks VACUUM operation the script does not know ('VACUUM FULL'), Databricks history that starts after the period, a Redshift day with no query history (2026-10-03), a Redshift DELETE behind a CTE that cannot be read, an unreadable snapshot, automated snapshots with no retention setting, an empty Redshift masking export, and a Databricks deletion_vectors value that is not true or false. The demo records are GENERATED (seed 231): not a company's data. Column names come from: Snowflake https://docs.snowflake.com/en/sql-reference/account-usage/tables https://docs.snowflake.com/en/sql-reference/account-usage/policy_references Databricks https://docs.databricks.com/aws/en/tables/history-schema https://docs.databricks.com/aws/en/tables/features/deletion-vectors https://docs.databricks.com/aws/en/sql/language-manual/information-schema/column_masks Redshift https://docs.aws.amazon.com/redshift/latest/dg/SYS_QUERY_HISTORY.html https://docs.aws.amazon.com/redshift/latest/dg/SYS_VACUUM_HISTORY.html https://docs.aws.amazon.com/redshift/latest/dg/r_SVV_ATTACHED_MASKING_POLICY.html https://docs.aws.amazon.com/cli/latest/reference/redshift/describe-cluster-snapshots.html Our assumptions, to check against one of your own rows: Databricks DESCRIBE HISTORY names a delete DELETE (the history-schema page says so), a finished VACUUM VACUUM END, and REORG TABLE ... APPLY (PURGE) is the operation REORG (the history-schema page names neither VACUUM's nor REORG's operation names, so both are settings); Snowflake POLICY_KIND is MASKING_POLICY for a masking policy; a Redshift DELETE is a statement that starts DELETE (FROM is optional) once comments and double quotes are removed. """ 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-16" # first day of the review 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 SNOWFLAKE_COUNT_FAILSAFE = True # judge Snowflake on Time Travel plus the 7-day Fail-safe (False: Time Travel only) DATABRICKS_DEFAULT_RETENTION_DAYS = 7 # delta.deletedFileRetentionDuration when the table does not set it DATABRICKS_VACUUM_DONE = "VACUUM END" # the DESCRIBE HISTORY operation that is a finished VACUUM DATABRICKS_VACUUM_OTHER = ["VACUUM START"] # other VACUUM operations you know and ignore DATABRICKS_PURGE_OPS = ["REORG"] # the operation that rewrites every file with deletion vectors (REORG ... APPLY (PURGE)) REDSHIFT_AUTOMATED_RETENTION_DAYS = None # no default: your cluster's automated snapshot retention in days ROW_AGES_MAX_STALE_DAYS = 3 # row ages must have been measured within this many days of AS_OF KNOWN_GAPS = {} # days you can explain, e.g. {"redshift": ["2026-10-03"]} # ---------------------------------------------------------------------------------------------- FAILSAFE_DAYS = 7 # Snowflake: "a (non-configurable) 7-day period" (docs.snowflake.com/en/user-guide/data-failsafe) FILES = { "retention_register": "retention_register.csv", "row_ages": "row_ages.csv", "masking_policy": "masking_policy.csv", "column_tags": "column_tags.csv", "clear_readers": "clear_readers.csv", "snowflake_tables": "snowflake_tables.csv", "snowflake_policy_references": "snowflake_policy_references.csv", "databricks_history": "databricks_history.csv", "databricks_table_properties": "databricks_table_properties.csv", "databricks_column_masks": "databricks_column_masks.csv", "redshift_sys_query_history": "redshift_sys_query_history.csv", "redshift_sys_vacuum_history": "redshift_sys_vacuum_history.csv", "redshift_snapshots": "redshift_snapshots.csv", "redshift_attached_masking_policy": "redshift_attached_masking_policy.csv", } PLATFORM_FILES = { "snowflake": ["snowflake_tables", "snowflake_policy_references"], "databricks": ["databricks_history", "databricks_table_properties", "databricks_column_masks"], "redshift": ["redshift_sys_query_history", "redshift_sys_vacuum_history", "redshift_snapshots", "redshift_attached_masking_policy"], } COLUMNS = { # the platforms' published column names, lower-cased here; the file's own case does not matter "retention_register": ["platform", "table_name", "data_category", "retention_days", "age_column", "recoverable_days_allowed"], "row_ages": ["platform", "table_name", "retention_days_used", "rows_total", "rows_past_retention", "rows_no_age", "oldest_date", "measured_on"], "masking_policy": ["data_category", "roles_in_clear"], "column_tags": ["platform", "table_name", "column_name", "data_category"], "clear_readers": ["platform", "table_name", "column_name", "role_name"], "snowflake_tables": ["table_catalog", "table_schema", "table_name", "table_type", "is_transient", "retention_time", "deleted"], "snowflake_policy_references": ["policy_name", "policy_kind", "ref_database_name", "ref_schema_name", "ref_entity_name", "ref_entity_domain", "ref_column_name", "policy_status"], "databricks_history": ["table_name", "version", "timestamp", "operation"], "databricks_table_properties": ["table_name", "deleted_file_retention_days", "deletion_vectors"], "databricks_column_masks": ["catalog_name", "schema_name", "table_name", "column_name", "mask_catalog", "mask_schema", "mask_name"], "redshift_sys_query_history": ["user_id", "query_id", "query_type", "status", "start_time", "query_text"], "redshift_sys_vacuum_history": ["schema_name", "table_name", "vacuum_type", "is_automatic", "status", "start_time", "end_time"], "redshift_snapshots": ["snapshotidentifier", "snapshotcreatetime", "snapshottype", "manualsnapshotretentionperiod"], "redshift_attached_masking_policy": ["policy_name", "schema_name", "table_name", "grantee", "grantee_type", "output_columns"], } PLATFORMS = ["snowflake", "databricks", "redshift"] OUT_DIR = pathlib.Path("evidence-out") 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}" IDENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_$]*$") NAME = re.compile(r"^[A-Za-z0-9_$]+(\.[A-Za-z0-9_$]+)*$") 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({col}, '{TS_RE}') THEN " f"try_strptime(substr(replace({col}, 'T', ' '), 1, 19), '%Y-%m-%d %H:%M:%S') END") def day(col): return (f"CASE WHEN regexp_full_match({col}, '{DATE_RE}') THEN try_strptime({col}, '%Y-%m-%d')::DATE END") def whole(col): """A whole number, or NULL (so '7.5', '1e3' and 'seven' are counted as unreadable, not guessed at).""" return f"CASE WHEN regexp_full_match({col}, '-?\\d+') THEN {col}::BIGINT END" def nm(col): """A table or column name compared whole, case-insensitively, without double quotes or backticks.""" return f"lower(replace(replace({col}, '\"', ''), '`', ''))" def fingerprint(rows): body = "\n".join("|".join("" if v is None else str(v) for v in r) for r in rows) return hashlib.sha256(body.encode()).hexdigest() def die(msg): print(f"STOP: {msg}") sys.exit(2) def row_age_query(plat, table, col, days): """The query that measures row ages: no platform view holds them. Run it on each register table and save every answer as a row of row_ages.csv.""" cutoff = {"snowflake": f"DATEADD(day, -{days}, CURRENT_DATE)", "redshift": f"DATEADD(day, -{days}, CURRENT_DATE)", "databricks": f"date_sub(current_date(), {days})"}[plat] return (f"SELECT '{plat}' AS platform, '{table}' AS table_name, {days} AS retention_days_used,\n" f" COUNT(*) AS rows_total,\n" f" COALESCE(SUM(CASE WHEN {col} < {cutoff} THEN 1 ELSE 0 END), 0) AS rows_past_retention,\n" f" COALESCE(SUM(CASE WHEN {col} IS NULL THEN 1 ELSE 0 END), 0) AS rows_no_age,\n" f" CAST(MIN({col}) AS DATE) AS oldest_date, CURRENT_DATE AS measured_on\n" f"FROM {table};") # --------------------------------------------------------------------------------------------- # Demo records: generated, in each platform's published column names # --------------------------------------------------------------------------------------------- def generate(plant, plant_input): rnd = random.Random(231) out = GEN_DIR / ("planted-input" if plant_input else "planted" if plant else "clean") out.mkdir(parents=True, exist_ok=True) D, T = dt.date, dt.datetime start, as_of = D(2026, 9, 16), D(2026, 10, 15) rows = {k: [] for k in FILES} stamp = lambda t: t.strftime("%Y-%m-%d %H:%M:%S") at = lambda y, m, d, h=0, mi=0, s=0: T(y, m, d, h, mi, s) reg = [("snowflake", "sf_prod.sales.customer_contacts", "contact", 730, "created_at", 14), ("snowflake", "sf_prod.sales.web_events", "event", 90, "event_date", 14), ("snowflake", "sf_prod.hr.employee_notes", "hr", 365, "noted_on", 14), ("databricks", "main.sales.customer_contacts", "contact", 730, "created_at", 14), ("databricks", "main.sales.web_events", "event", 90, "event_date", 14), ("redshift", "sales.customer_contacts", "contact", 730, "created_at", 14), ("redshift", "sales.web_events", "event", 90, "event_date", 14)] rows["retention_register"] = [list(r) for r in reg] # row ages: measured on AS_OF, with the register's own retention oldest = {"sf_prod.sales.customer_contacts": "2024-10-19", "sf_prod.sales.web_events": "2026-07-20", "sf_prod.hr.employee_notes": "2025-10-18", "main.sales.customer_contacts": "2024-10-21", "main.sales.web_events": "2026-07-22", "sales.customer_contacts": "2024-10-22", "sales.web_events": "2026-07-19"} for p, t, c, ret, col, allowed in reg: total = rnd.randrange(40_000, 400_000) rows["row_ages"].append([p, t, ret, total, 0, 0, oldest[t], "2026-10-15"]) if plant: for r in rows["row_ages"]: if r[1] == "sf_prod.sales.web_events": r[4], r[6] = 1204, "2026-06-30" if r[1] == "main.sales.web_events": r[4], r[6] = 37, "2026-07-02" rows["row_ages"] = [r for r in rows["row_ages"] if r[1] != "sales.customer_contacts"] # never measured rows["masking_policy"] = [["contact", "SUPPORT_LEAD;DATA_STEWARD"], ["hr", "HR_ADMIN"], ["network", "SECURITY_ANALYST"]] tags = [("snowflake", "sf_prod.sales.customer_contacts", "EMAIL", "contact"), ("snowflake", "sf_prod.sales.customer_contacts", "PHONE", "contact"), ("snowflake", "sf_prod.hr.employee_notes", "NOTE_TEXT", "hr"), ("snowflake", "sf_prod.sales.web_events", "IP_ADDRESS", "network"), ("databricks", "main.sales.customer_contacts", "email", "contact"), ("databricks", "main.sales.customer_contacts", "phone", "contact"), ("databricks", "main.sales.web_events", "ip_address", "network"), ("redshift", "sales.customer_contacts", "email", "contact"), ("redshift", "sales.customer_contacts", "phone", "contact"), ("redshift", "sales.web_events", "ip_address", "network")] rows["column_tags"] = [list(t) for t in tags] rows["clear_readers"] = [["snowflake", "sf_prod.sales.customer_contacts", "EMAIL", "SUPPORT_LEAD"], ["snowflake", "sf_prod.hr.employee_notes", "NOTE_TEXT", "HR_ADMIN"], ["databricks", "main.sales.customer_contacts", "email", "data_steward"], ["redshift", "sales.web_events", "ip_address", "security_analyst"]] if plant: rows["clear_readers"].append(["redshift", "sales.customer_contacts", "email", "marketing_analyst"]) # Snowflake: ACCOUNT_USAGE.TABLES and POLICY_REFERENCES sf = [["SF_PROD", "SALES", "CUSTOMER_CONTACTS", "BASE TABLE", "NO", 7, ""], ["SF_PROD", "SALES", "WEB_EVENTS", "BASE TABLE", "YES", 1, ""], ["SF_PROD", "HR", "EMPLOYEE_NOTES", "BASE TABLE", "NO", 90 if plant else 1, ""], ["SF_PROD", "SALES", "AUDIT_LOG", "BASE TABLE", "NO", 1, ""], ["SF_PROD", "SALES", "WEB_EVENTS", "BASE TABLE", "NO", 90, "2026-08-02 09:00:00.000 +0000"]] # dropped copy rows["snowflake_tables"] = sf pr = [["PII_MASK", "MASKING_POLICY", "SF_PROD", "SALES", "CUSTOMER_CONTACTS", "TABLE", "EMAIL", "ACTIVE"], ["PII_MASK", "MASKING_POLICY", "SF_PROD", "SALES", "CUSTOMER_CONTACTS", "TABLE", "PHONE", "ACTIVE"], ["HR_MASK", "MASKING_POLICY", "SF_PROD", "HR", "EMPLOYEE_NOTES", "TABLE", "NOTE_TEXT", "ACTIVE"], ["IP_MASK", "MASKING_POLICY", "SF_PROD", "SALES", "WEB_EVENTS", "TABLE", "IP_ADDRESS", "ACTIVE"], ["RAP_REGION", "ROW_ACCESS_POLICY", "SF_PROD", "SALES", "WEB_EVENTS", "TABLE", "", "ACTIVE"]] if plant: pr = [r for r in pr if r[6] != "PHONE"] rows["snowflake_policy_references"] = pr # Databricks: DESCRIBE HISTORY per table, table properties, information_schema.column_masks props = {"main.sales.customer_contacts": "30" if plant else "7", "main.sales.web_events": ""} for t in ("main.sales.customer_contacts", "main.sales.web_events"): h = [(0, at(2026, 6, 1, 8), "CREATE TABLE"), (1, at(2026, 9, 1, 8), "WRITE"), (2, at(2026, 9, 20, 10), "DELETE"), (3, at(2026, 9, 28, 3), "VACUUM START"), (4, at(2026, 9, 28, 3, 4, 10), "VACUUM END"), (5, at(2026, 10, 1, 8), "WRITE"), (6, at(2026, 10, 8, 10), "DELETE"), (7, at(2026, 10, 12, 4), "OPTIMIZE")] if plant_input and t.endswith("web_events"): h = [x for x in h if x[0] >= 2] # history starts after the period began h.append((8, at(2026, 10, 13, 5), "VACUUM FULL")) for v, when, op in h: rows["databricks_history"].append([t, v, stamp(when) + ".000", op]) rows["databricks_table_properties"] = [[t, v, "false"] for t, v in props.items()] # delta.enableDeletionVectors cm = [["main", "sales", "customer_contacts", "email", "main", "gov", "mask_email"], ["main", "sales", "customer_contacts", "phone", "main", "gov", "mask_phone"], ["main", "sales", "web_events", "ip_address", "main", "gov", "mask_ip"]] if plant: cm = [r for r in cm if r[3] != "ip_address"] rows["databricks_column_masks"] = cm # Redshift: SYS_QUERY_HISTORY, SYS_VACUUM_HISTORY, snapshots, SVV_ATTACHED_MASKING_POLICY q, qid = [], 1000 d = start while d <= as_of: if not (plant_input and d == D(2026, 10, 3)): for h in (7, 11, 15): qid += 1 q.append([100 + rnd.randrange(5), qid, "SELECT", "success", stamp(T(d.year, d.month, d.day, h, rnd.randrange(60), rnd.randrange(60))), "select count(*) from sales.web_events"]) d += dt.timedelta(days=1) q.append([101, 2001, "DELETE", "success", "2026-09-20 11:00:00", "delete from sales.customer_contacts where created_at < '2024-09-20'"]) q.append([101, 2002, "DELETE", "success", "2026-09-20 11:05:00", '/* nightly purge */ DELETE FROM "sales"."web_events" WHERE event_date < \'2026-06-20\' -- 90 days']) q.append([101, 2003, "DELETE", "failed", "2026-09-22 11:00:00", "delete from sales.web_events where event_date < '2026-06-22'"]) if plant_input: q.append([101, 2004, "OTHER", "success", "2026-10-06 11:00:00", "with old as (select 1) delete from sales.web_events where 1 = 1"]) rows["redshift_sys_query_history"] = q rows["redshift_sys_vacuum_history"] = [ ["sales", "customer_contacts", "Delete", "true", "Complete", "2026-09-21 02:00:00", "2026-09-21 02:20:00"], ["sales", "web_events", "Delete", "true", "Complete", "2026-09-21 02:20:00", "2026-09-21 02:41:00"], ["sales", "web_events", "Sort", "true", "Complete", "2026-10-02 02:00:00", "2026-10-02 02:30:00"]] snaps = [["rs-auto-1014", "2026-10-14 03:00:00", "automated", ""], ["rs-auto-1015", "2026-10-15 03:00:00", "automated", ""], ["rs-manual-0918", "2026-09-18 09:00:00", "manual", "-1" if plant else "10"]] if plant_input: snaps.append(["rs-manual-bad", "yesterday", "manual", "ten"]) rows["redshift_snapshots"] = snaps mp = [["pii_mask", "sales", "customer_contacts", "public", "public", "{email}"], ["pii_mask", "sales", "customer_contacts", "public", "public", "{phone}"], ["ip_mask", "sales", "web_events", "public", "public", "{ip_address}"]] rows["redshift_attached_masking_policy"] = [] if plant_input else mp if plant_input: rows["masking_policy"].append(["financial", "FINANCE_LEAD"]) rows["column_tags"].append(["snowflake", "sf_prod.sales.customer_contacts", "ssn", ""]) rows["clear_readers"].append(["snowflake", "sf_prod.sales.customer_contacts", "DOB", "SUPPORT_LEAD"]) rows["clear_readers"].append(["snowflake", "sf_prod.sales.customer_contacts", "EMAIL", ""]) rows["row_ages"].append(["redshift", "sales.old_staging", 90, 10, 0, 0, "2026-09-01", "2026-10-15"]) for r in rows["row_ages"]: if r[1] == "sf_prod.sales.customer_contacts": r[2] = 365 if r[1] == "sf_prod.hr.employee_notes": r[7], r[5] = "2026-09-01", 4 rows["retention_register"].append(["redshift", "sales.customer_contacts", "contact", 730, "created_at", "14.5"]) rows["column_tags"].append(["oracle", "erp.hr.pay", "salary", "hr"]) rows["row_ages"].append(["snowflake", "sf_prod.sales.web_events", 90, 105751, 0, 0, "2026-07-20", "2026-10-15"]) for r in rows["row_ages"]: if r[1] == "main.sales.customer_contacts": r[3] = "many" rows["snowflake_tables"].append(["SF_PROD", "SALES", "AUDIT_LOG_OLD", "BASE TABLE", "MAYBE", "seven", ""]) rows["snowflake_tables"].append(["SF_PROD", "SALES", "WEB_EVENTS", "BASE TABLE", "YES", 1, ""]) rows["databricks_history"].append(["main.sales.customer_contacts", 9, "last tuesday", "WRITE"]) rows["databricks_table_properties"][1][1] = "7 days" rows["databricks_table_properties"][0][2] = "maybe" for key, data in rows.items(): with open(out / FILES[key], "w", newline="") as f: w = csv.writer(f) w.writerow(COLUMNS[key] if key not in ("snowflake_tables",) else [c.upper() for c in COLUMNS[key]]) w.writerows(data) return out # --------------------------------------------------------------------------------------------- # Reading the exports # --------------------------------------------------------------------------------------------- def load(con, key, path): if not path.is_file(): die(f"{path} is missing (set EXPORT_DIR, or change FILES['{key}'])") with open(path, newline="", encoding="utf-8-sig") as f: header = next(csv.reader(f), []) have = {h.strip().lower(): h for h in header} missing = [c for c in COLUMNS[key] if c not in have] if missing: die(f"{path.name} has no column {', '.join(missing)} (it has: {', '.join(header)})") sel = ", ".join(f"NULLIF(trim(\"{have[c]}\"), '') AS {c}" for c in COLUMNS[key]) con.execute(f"CREATE OR REPLACE TABLE {key} AS SELECT {sel} FROM " f"read_csv(?, header=true, all_varchar=true, sample_size=-1)", [str(path)]) def build(con, directory): for key in ["retention_register", "row_ages", "masking_policy", "column_tags", "clear_readers"] + \ [k for p in PLATFORMS_EXPORTED for k in PLATFORM_FILES[p]]: load(con, key, directory / FILES[key]) P = ", ".join(f"'{p}'" for p in PLATFORMS_EXPORTED) run = lambda sql: con.execute(sql) run(f"""CREATE TABLE reg AS SELECT lower(platform) AS platform, {nm('table_name')} AS t, table_name AS raw_name, lower(data_category) AS category, {whole('retention_days')} AS ret, {whole('recoverable_days_allowed')} AS allowed, age_column FROM retention_register WHERE lower(platform) IN ({P})""") run(f"""CREATE TABLE ages AS SELECT lower(platform) AS platform, {nm('table_name')} AS t, {whole('retention_days_used')} AS ret_used, {whole('rows_total')} AS total, {whole('rows_past_retention')} AS past, {whole('rows_no_age')} AS no_age, {day('oldest_date')} AS oldest, {day('measured_on')} AS measured, oldest_date AS oldest_raw FROM row_ages WHERE lower(platform) IN ({P})""") # one line per category: roles from every line for the same category are merged run("""CREATE TABLE pol AS SELECT lower(data_category) AS category, string_agg(coalesce(roles_in_clear, ''), ';' ORDER BY coalesce(roles_in_clear, '')) AS roles FROM masking_policy GROUP BY 1""") run(f"""CREATE TABLE tags AS SELECT DISTINCT lower(platform) AS platform, {nm('table_name')} AS t, {nm('column_name')} AS col, lower(data_category) AS category FROM column_tags WHERE lower(platform) IN ({P})""") run(f"""CREATE TABLE clr AS SELECT lower(platform) AS platform, {nm('table_name')} AS t, {nm('column_name')} AS col, {nm('role_name')} AS role FROM clear_readers WHERE lower(platform) IN ({P})""") if "snowflake" in PLATFORMS_EXPORTED: run(f"""CREATE TABLE sf_t AS SELECT {nm("table_catalog || '.' || table_schema || '.' || table_name")} AS t, upper(is_transient) AS transient, upper(coalesce(table_type, '')) AS ttype, {whole('retention_time')} AS rt, deleted FROM snowflake_tables""") run(f"""CREATE TABLE sf_pr AS SELECT upper(policy_kind) AS kind, upper(policy_status) AS status, {nm("ref_database_name || '.' || ref_schema_name || '.' || ref_entity_name")} AS t, {nm('ref_column_name')} AS col FROM snowflake_policy_references""") if "databricks" in PLATFORMS_EXPORTED: run(f"""CREATE TABLE db_h AS SELECT {nm('table_name')} AS t, {whole('version')} AS version, {ts('timestamp')} AS ts, timestamp AS ts_raw, upper(operation) AS op FROM databricks_history""") run(f"""CREATE TABLE db_p AS SELECT {nm('table_name')} AS t, deleted_file_retention_days AS raw, {whole('deleted_file_retention_days')} AS days, lower(deletion_vectors) AS dv FROM databricks_table_properties""") run(f"""CREATE TABLE db_m AS SELECT {nm("catalog_name || '.' || schema_name || '.' || table_name")} AS t, {nm('column_name')} AS col FROM databricks_column_masks""") if "redshift" in PLATFORMS_EXPORTED: run(f"""CREATE TABLE rs_q AS SELECT lower(status) AS status, {ts('start_time')} AS ts, start_time AS ts_raw, query_text FROM redshift_sys_query_history""") run(f"""CREATE TABLE rs_v AS SELECT {nm("schema_name || '.' || table_name")} AS t, lower(vacuum_type) AS vtype, lower(status) AS status, {ts('start_time')} AS s_ts, {ts('end_time')} AS e_ts FROM redshift_sys_vacuum_history""") run(f"""CREATE TABLE rs_s AS SELECT {ts('snapshotcreatetime')} AS created, lower(snapshottype) AS stype, {whole('manualsnapshotretentionperiod')} AS ret, snapshotcreatetime AS created_raw, manualsnapshotretentionperiod AS ret_raw FROM redshift_snapshots""") run(f"""CREATE TABLE rs_m AS SELECT {nm("schema_name || '.' || table_name")} AS t, output_columns AS outc, lower(grantee) AS grantee, lower(grantee_type) AS gtype FROM redshift_attached_masking_policy""") def input_checks(con): ds, de = f"DATE '{PERIOD_START}'", f"DATE '{AS_OF}'" n = lambda sql: con.execute(sql).fetchone()[0] out = [] # (label, number of bad rows) add = lambda label, sql: out.append((label, n(sql))) known = "('snowflake', 'databricks', 'redshift')" add("every platform name in your lists is snowflake, databricks or redshift", f"""SELECT count(*) FROM (SELECT lower(platform) p FROM retention_register UNION ALL SELECT lower(platform) FROM row_ages UNION ALL SELECT lower(platform) FROM column_tags UNION ALL SELECT lower(platform) FROM clear_readers) WHERE p IS NULL OR p NOT IN {known}""") add("no table appears twice in the retention register", "SELECT count(*) FROM (SELECT platform, t FROM reg GROUP BY 1, 2 HAVING count(*) > 1)") add("every register line has a category, whole-number days, and a date column that is a plain name", """SELECT count(*) FROM reg WHERE category IS NULL OR ret IS NULL OR allowed IS NULL OR ret < 0 OR allowed < 0 OR age_column IS NULL OR NOT regexp_full_match(age_column, '[A-Za-z_][A-Za-z0-9_$]*')""") add("every column tag has a table, a column and a category", "SELECT count(*) FROM tags WHERE t IS NULL OR col IS NULL OR category IS NULL") add("every category in the masking policy is on at least one tagged column", "SELECT count(*) FROM pol WHERE category NOT IN (SELECT category FROM tags WHERE category IS NOT NULL)") add("every clear reader names a table, a column and a role", "SELECT count(*) FROM clr WHERE t IS NULL OR col IS NULL OR role IS NULL") add("every clear reader is on a column that is tagged", "SELECT count(*) FROM clr WHERE NOT EXISTS (SELECT 1 FROM tags g WHERE g.platform = clr.platform AND g.t = clr.t AND g.col = clr.col)") add("every row-age line is for a table in the register", "SELECT count(*) FROM ages WHERE NOT EXISTS (SELECT 1 FROM reg WHERE reg.platform = ages.platform AND reg.t = ages.t)") add("no table has two row-age lines", "SELECT count(*) FROM (SELECT platform, t FROM ages GROUP BY 1, 2 HAVING count(*) > 1)") add("row-age counts are whole numbers and the oldest date is YYYY-MM-DD (or blank for an empty table)", """SELECT count(*) FROM ages WHERE total IS NULL OR past IS NULL OR no_age IS NULL OR measured IS NULL OR (oldest IS NULL AND (oldest_raw IS NOT NULL OR total > 0))""") add("row ages were made with the same retention days as the register", "SELECT count(*) FROM ages a JOIN reg r ON r.platform = a.platform AND r.t = a.t WHERE a.ret_used IS DISTINCT FROM r.ret") add(f"row ages were measured within {ROW_AGES_MAX_STALE_DAYS} days before AS_OF, not after it", f"SELECT count(*) FROM ages WHERE measured IS NOT NULL AND (measured > {de} OR measured < {de} - {int(ROW_AGES_MAX_STALE_DAYS)})") add("no register table has rows with no date to age them by", "SELECT count(*) FROM ages WHERE no_age > 0") if "snowflake" in PLATFORMS_EXPORTED: add("Snowflake: retention_time is a whole number and is_transient is YES or NO", "SELECT count(*) FROM sf_t WHERE rt IS NULL OR transient IS NULL OR transient NOT IN ('YES', 'NO')") add("Snowflake: no register table has two live (not dropped) rows", "SELECT count(*) FROM (SELECT t FROM sf_t WHERE deleted IS NULL AND t IN (SELECT t FROM reg WHERE platform = 'snowflake') GROUP BY 1 HAVING count(*) > 1)") if "databricks" in PLATFORMS_EXPORTED: add("Databricks: every history timestamp is YYYY-MM-DD HH:MM:SS", "SELECT count(*) FROM db_h WHERE ts IS NULL OR version IS NULL") done, other = DATABRICKS_VACUUM_DONE.upper(), [o.upper() for o in DATABRICKS_VACUUM_OTHER] known_ops = ", ".join(f"'{o}'" for o in [done] + other) add("Databricks: every VACUUM operation in the history is one the settings name", f"SELECT count(*) FROM db_h WHERE op LIKE 'VACUUM%' AND op NOT IN ({known_ops})") add("Databricks: each register table's history reaches back to the period's start (or to its first version)", f"""SELECT count(*) FROM (SELECT t, min(ts) mn, min(version) v FROM db_h WHERE t IN (SELECT t FROM reg WHERE platform = 'databricks') GROUP BY t) WHERE mn > {ds} AND v <> 0""") add("Databricks: a table's deleted-file retention is a whole number of days or blank", "SELECT count(*) FROM db_p WHERE raw IS NOT NULL AND (days IS NULL OR days < 0)") add("Databricks: every register table says whether deletion vectors are on (true or false)", """SELECT count(*) FROM reg r LEFT JOIN db_p p ON p.t = r.t WHERE r.platform = 'databricks' AND (p.dv IS NULL OR p.dv NOT IN ('true', 'false'))""") if "redshift" in PLATFORMS_EXPORTED: add("Redshift: every query-history, vacuum and snapshot timestamp is YYYY-MM-DD HH:MM:SS", """SELECT (SELECT count(*) FROM rs_q WHERE ts IS NULL) + (SELECT count(*) FROM rs_v WHERE s_ts IS NULL OR (status = 'complete' AND e_ts IS NULL)) + (SELECT count(*) FROM rs_s WHERE created IS NULL)""") gaps = KNOWN_GAPS.get("redshift", []) skip = ", ".join(f"DATE '{g}'" for g in gaps) or "DATE '1900-01-01'" add("Redshift: every day from the period's start to AS_OF has query history, except KNOWN_GAPS", f"""SELECT count(*) FROM generate_series({ds}, {de}, INTERVAL 1 DAY) g(d) WHERE d::DATE NOT IN ({skip}) AND NOT EXISTS (SELECT 1 FROM rs_q WHERE ts::DATE = g.d::DATE)""") add("Redshift: snapshot type is manual or automated, and a manual retention is -1 or whole days from 1", """SELECT count(*) FROM rs_s WHERE stype IS NULL OR stype NOT IN ('manual', 'automated') OR (stype = 'manual' AND (ret IS NULL OR (ret <> -1 AND ret < 1)))""") add("Redshift: REDSHIFT_AUTOMATED_RETENTION_DAYS is set when automated snapshots exist", f"SELECT CASE WHEN {REDSHIFT_AUTOMATED_RETENTION_DAYS is None} AND EXISTS (SELECT 1 FROM rs_s WHERE stype = 'automated') THEN 1 ELSE 0 END") add("Redshift: the masking-policy export has rows (a user who is not a superuser or sys:secadmin sees 0)", "SELECT CASE WHEN (SELECT count(*) FROM rs_m) = 0 THEN 1 ELSE 0 END") return out # --------------------------------------------------------------------------------------------- # The three lists # --------------------------------------------------------------------------------------------- def clean_sql(text): text = re.sub(r"/\*.*?\*/", " ", text or "", flags=re.S) text = re.sub(r"--[^\n]*", " ", text) return text.replace('"', "").replace("`", "") def redshift_deletes(con): """Successful DELETE statements by target (schema.table); FROM is optional in Redshift. A DELETE that does not start the statement (a CTE in front), or whose target names no schema, cannot be matched to the register for certain; when its table name is the table part of a Redshift register entry it is counted for the input check, never skipped. One on a table the register cannot hold (a staging table) is left out.""" found, unreadable = [], 0 bare = {r[0].split(".")[-1] for r in con.execute("SELECT t FROM reg WHERE platform = 'redshift' AND t IS NOT NULL").fetchall()} for status, when, text in con.execute("SELECT status, ts, query_text FROM rs_q ORDER BY ts, query_text").fetchall(): c = re.sub(r"'(?:[^']|'')*'", "''", clean_sql(text)) # string literals cannot name a table m = re.match(r"\s*delete\s+(?:from\s+)?(?:only\s+)?([\w$.]+)", c, re.I) if status != "success" or when is None: # an unreadable time is counted by the timestamp check continue if m and "." in m.group(1): parts = m.group(1).lower().split(".") found.append((".".join(parts[-2:]), when)) else: # GRANT ... DELETE ON, REVOKE DELETE and the like are not deletes w = re.search(r"\bdelete\s+(?:from\s+)?(?:only\s+)?([\w$.]+)", c, re.I) if re.match(r"\s*with\b", c, re.I) else None target = (m or w).group(1).lower() if (m or w) else None if target and target.split(".")[-1] in bare: unreadable += 1 return found, unreadable def list_rows_past(con): rows = con.execute("""SELECT r.platform, r.raw_name, r.category, r.ret, a.total, a.past, a.oldest, CASE WHEN a.t IS NULL OR a.past IS NULL THEN 'NOT MEASURED' WHEN a.past > 0 THEN 'PAST RETENTION' ELSE 'OK' END FROM reg r LEFT JOIN ages a ON a.platform = r.platform AND a.t = r.t ORDER BY 1, 2""").fetchall() return ["platform", "table", "category", "retention_days", "rows", "rows_past_retention", "oldest_date", "status"], rows def list_recoverable(con, as_of_end, redshift_dels): rows = [] reg = con.execute("SELECT platform, t, raw_name, category, allowed FROM reg ORDER BY 1, 2").fetchall() rank = {"INDEFINITE SNAPSHOT": 4, "RECOVERABLE BEYOND POLICY": 3, "OPEN, WITHIN SO FAR": 2, "WITHIN POLICY": 1} for plat, t, raw, cat, allowed in reg: days, basis, status = None, "", "" if allowed is None: rows.append((plat, raw, cat, allowed, None, "recoverable_days_allowed is not a whole number", "NOT MEASURED")) continue if plat == "snowflake": hit = con.execute("SELECT rt, transient, ttype FROM sf_t WHERE t = ? AND deleted IS NULL", [t]).fetchall() if not hit: status, basis = "NOT MEASURED", "table not in the ACCOUNT_USAGE.TABLES export" elif hit[0][0] is None: status, basis = "NOT MEASURED", "retention_time is not a whole number" else: rt, transient, ttype = hit[0] fs = FAILSAFE_DAYS if (transient == "NO" and "TEMPORARY" not in ttype and SNOWFLAKE_COUNT_FAILSAFE) else 0 days = rt + fs basis = (f"Time Travel {rt} d + Fail-safe {FAILSAFE_DAYS} d" if fs else f"Time Travel {rt} d, transient: no Fail-safe" if transient == "YES" else f"Time Travel {rt} d, temporary: no Fail-safe" if "TEMPORARY" in ttype else f"Time Travel {rt} d only") status = "RECOVERABLE BEYOND POLICY" if days > allowed else "WITHIN POLICY" elif plat == "databricks": if not con.execute("SELECT count(*) FROM db_h WHERE t = ?", [t]).fetchone()[0]: status, basis = "NOT MEASURED", "table not in the DESCRIBE HISTORY export" else: prop = con.execute("SELECT days, dv FROM db_p WHERE t = ?", [t]).fetchone() keep = prop[0] if prop and prop[0] is not None else DATABRICKS_DEFAULT_RETENTION_DAYS dv = prop[1] if prop else None done = DATABRICKS_VACUUM_DONE.upper() purge = [o.upper() for o in DATABRICKS_PURGE_OPS] dels = con.execute( "SELECT ts FROM db_h WHERE t = ? AND op = 'DELETE' AND ts >= ? AND ts <= ? ORDER BY ts", [t, PERIOD_START + " 00:00:00", as_of_end.strftime("%Y-%m-%d %H:%M:%S")]).fetchall() best = None for (d,) in dels: # with deletion vectors a DELETE only marks rows: a REORG ... APPLY (PURGE) must rewrite the files # first, and the VACUUM that removes the old files must come at least the retention after that start = d if dv == "true": hit = con.execute(f"SELECT ts, op FROM db_h WHERE t = ? AND op IN ({', '.join('?' for _ in purge)}) AND ts >= ? ORDER BY ts LIMIT 1", [t, *purge, d]).fetchone() start, purge_op = (hit[0], hit[1]) if hit else (None, None) v = None if start is None else con.execute("SELECT min(ts) FROM db_h WHERE t = ? AND op = ? AND ts >= ?", [t, done, start + dt.timedelta(days=keep)]).fetchone()[0] if dv not in ("true", "false"): val, st, b = None, "unknown", f"delete {d:%Y-%m-%d}; deletion_vectors not stated, so removal cannot be judged" elif v: val, st = (v - d).total_seconds() / 86400, None b = f"delete {d:%Y-%m-%d}, " + (f"{purge_op} {start:%Y-%m-%d}, " if dv == "true" else "") + f"VACUUM finished {v:%Y-%m-%d}, files kept {keep} d" elif dv == "true" and start is None: val, st, b = (as_of_end - d).total_seconds() / 86400, "open", f"delete {d:%Y-%m-%d}, deletion vectors on and no " + "/".join(purge) + " since: rows still in live files" else: val, st, b = (as_of_end - d).total_seconds() / 86400, "open", f"delete {d:%Y-%m-%d}, no VACUUM yet old enough to remove its files (kept {keep} d)" if st == "unknown": cand = (0, 0.0, "NOT MEASURED", b) best = cand if best is None else best continue st = ("RECOVERABLE BEYOND POLICY" if val > allowed else "OPEN, WITHIN SO FAR" if st else "WITHIN POLICY") cand = (rank[st], val, st, b) best = cand if best is None or cand[:2] > best[:2] else best if best is None: status, basis = "NO DELETE IN PERIOD", "no DELETE in the history during the period" else: _, days, status, basis = best else: mine = [d for tt, d in redshift_dels if tt == t and PERIOD_START <= f"{d:%Y-%m-%d}" <= AS_OF] if not mine: status, basis = "NO DELETE IN PERIOD", "no successful DELETE in the query history during the period" else: auto = REDSHIFT_AUTOMATED_RETENTION_DAYS or 0 # a snapshot whose time or manual retention cannot be read is counted by an input check, not used here snaps = con.execute("""SELECT created, stype, ret FROM rs_s WHERE created IS NOT NULL AND NOT (stype = 'manual' AND ret IS NULL)""").fetchall() best = None for d in sorted(mine): v = con.execute("SELECT min(e_ts) FROM rs_v WHERE t = ? AND status = 'complete' AND vtype IN ('delete', 'full') AND s_ts >= ?", [t, d]).fetchone()[0] indef = [c for c, ty, r in snaps if ty == "manual" and r == -1 and c < d] exp = [c + dt.timedelta(days=(r if ty == "manual" else auto)) for c, ty, r in snaps if c < d and not (ty == "manual" and r == -1) and c + dt.timedelta(days=(r if ty == "manual" else auto)) > d] if indef: cand = (rank["INDEFINITE SNAPSHOT"], 0.0, "INDEFINITE SNAPSHOT", f"delete {d:%Y-%m-%d}; a manual snapshot from {min(indef):%Y-%m-%d} is kept indefinitely") else: if v: end = max([v] + exp) val, st = (end - d).total_seconds() / 86400, None b = f"delete {d:%Y-%m-%d}, VACUUM DELETE finished {v:%Y-%m-%d}" + (f", snapshot kept to {max(exp):%Y-%m-%d}" if exp else "") else: val, st = (max([as_of_end] + exp) - d).total_seconds() / 86400, "open" b = f"delete {d:%Y-%m-%d}, no VACUUM DELETE since" st = ("RECOVERABLE BEYOND POLICY" if val > allowed else "OPEN, WITHIN SO FAR" if st else "WITHIN POLICY") cand = (rank[st], val, st, b) best = cand if best is None or cand[:2] > best[:2] else best _, days, status, basis = best if status == "NOT MEASURED": days = None shown = None if days is None or status == "INDEFINITE SNAPSHOT" else round(days, 1) rows.append((plat, raw, cat, allowed, shown, basis, status)) return ["platform", "table", "category", "allowed_days", "recoverable_days", "basis", "status"], rows def list_masking(con): rows = [] for plat, t, col, cat, roles in con.execute("""SELECT g.platform, g.t, g.col, g.category, p.roles FROM tags g LEFT JOIN pol p ON p.category = g.category WHERE g.category IS NOT NULL AND g.t IS NOT NULL AND g.col IS NOT NULL ORDER BY 1, 2, 3""").fetchall(): if roles is None: rows.append((plat, t, col, cat, "", "NO POLICY FOR CATEGORY")) continue active, inactive, partial = False, [], [] if plat == "snowflake": for st, in con.execute("SELECT status FROM sf_pr WHERE kind = 'MASKING_POLICY' AND t = ? AND col = ?", [t, col]).fetchall(): active, inactive = active or st == "ACTIVE", inactive + ([st] if st != "ACTIVE" else []) elif plat == "databricks": active = con.execute("SELECT count(*) FROM db_m WHERE t = ? AND col = ?", [t, col]).fetchone()[0] > 0 else: # a policy attached to one role or user masks the column only for that grantee: PUBLIC covers everyone for outc, grantee, gtype in con.execute("SELECT outc, grantee, gtype FROM rs_m WHERE t = ?", [t]).fetchall(): if col in [x.lower() for x in re.findall(r"[A-Za-z0-9_$]+", outc or "")]: if gtype == "public" or grantee == "public": active = True else: partial.append(grantee or "?") allowed = {r.strip().lower() for r in roles.split(";") if r.strip()} clear = sorted({r for (r,) in con.execute("SELECT role FROM clr WHERE platform = ? AND t = ? AND col = ? AND role IS NOT NULL", [plat, t, col]).fetchall()}) extra = [r for r in clear if r not in allowed] flags = [] if not active: flags.append("MASK NOT ACTIVE (" + ", ".join(sorted(set(inactive))) + ")" if inactive else "NOT MASKED FOR PUBLIC (attached only to " + ", ".join(sorted(set(partial))) + ")" if partial else "NOT MASKED") if extra: flags.append("CLEAR READER NOT ON POLICY LIST: " + ", ".join(extra)) rows.append((plat, t, col, cat, ", ".join(clear), "; ".join(flags) or "OK")) return ["platform", "table", "column", "category", "roles_reading_in_clear", "status"], rows def show(name, title, cols, rows, filters, incomplete, run_at): rows = sorted(rows, key=lambda r: tuple("" if v is None else str(v) for v in r)) OUT_DIR.mkdir(exist_ok=True) with open(OUT_DIR / f"{name}.csv", "w", newline="") as f: w = csv.writer(f) w.writerow(cols) w.writerows(rows) print(f"\n== {title}: {name}" + (" ** INCOMPLETE: see input checks **" if incomplete else "")) print(f" period {PERIOD_START} to {AS_OF} | {filters} | run {run_at} | rows {len(rows)} | sha256 {fingerprint(rows)}") cells = [cols] + [["" if v is None else str(v) for v in r] for r in rows] width = [max(len(c[i]) for c in cells) for i in range(len(cols))] for k, c in enumerate(cells): print(" " + " ".join(v.ljust(width[i]) for i, v in enumerate(c)).rstrip()) if k == 0: print(" " + " ".join("-" * w for w in width)) def main(): global REDSHIFT_AUTOMATED_RETENTION_DAYS argv = sys.argv[1:] if "--queries" in argv: d = generate(False, False) if "--demo" in argv else pathlib.Path(EXPORT_DIR or ".") with open(d / FILES["retention_register"], newline="", encoding="utf-8-sig") as f: for r in csv.DictReader(f): r = {k.strip().lower(): (v or "").strip() for k, v in r.items()} p, t, c, n = r["platform"].lower(), r["table_name"], r["age_column"], r["retention_days"] if p not in PLATFORMS or not NAME.match(t) or not IDENT.match(c) or not n.isdigit(): die(f"cannot write a query for {r}: platform, a plain table name, a plain column name and whole days are needed") print(row_age_query(p, t, c, int(n)), "\n") return 0 plant_input = "--plant-input" in argv plant = "--plant" in argv or plant_input run_at = dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%d %H:%M:%SZ") if "--demo" in argv: directory = generate("--plant" in argv, plant_input) REDSHIFT_AUTOMATED_RETENTION_DAYS = None if plant_input else 1 print(f"Demo: GENERATED records (seed 231) in {directory}/, not a company's data.") else: if EXPORT_DIR is None: die("set EXPORT_DIR (and the other settings) at the top of this file, or run with --demo") directory = pathlib.Path(EXPORT_DIR) bad_dates = [x for x in (PERIOD_START, AS_OF) if not re.fullmatch(DATE_RE, x)] if bad_dates or PERIOD_START > AS_OF: die("PERIOD_START and AS_OF must be YYYY-MM-DD, with PERIOD_START not after AS_OF") as_of_end = dt.datetime.strptime(AS_OF, "%Y-%m-%d") + dt.timedelta(hours=23, minutes=59, seconds=59) con = duckdb.connect() build(con, directory) print("\n== Input checks") checks = input_checks(con) redshift_dels, unreadable = [], 0 if "redshift" in PLATFORMS_EXPORTED: redshift_dels, unreadable = redshift_deletes(con) checks.append(("Redshift: every DELETE that could touch a register table starts the statement and names schema.table", unreadable)) failed = [c for c in checks if c[1]] for label, bad in checks: print(f" [{'FAIL' if bad else 'ok'}] {label}" + (f": {bad} row(s)" if bad else "")) incomplete = bool(failed) not_checked = [p for p in PLATFORMS if p not in PLATFORMS_EXPORTED] for p in not_checked: print(f" NOT CHECKED: {p} is not in PLATFORMS_EXPORTED") c1, r1 = list_rows_past(con) show("iso_1_rows_past_retention", "Rows past retention", c1, r1, "filters: register tables on exported platforms", incomplete, run_at) c2, r2 = list_recoverable(con, as_of_end, redshift_dels) show("iso_2_deleted_but_recoverable", "Deleted but still recoverable", c2, r2, "filters: register tables; Snowflake from table settings, the others from deletes in the period", incomplete, run_at) c3, r3 = list_masking(con) show("iso_3_masking_against_policy", "Masking against your policy", c3, r3, "filters: columns tagged with a category", incomplete, run_at) count = lambda rows, i, s: sum(1 for r in rows if str(r[i]).startswith(s)) flags = {"past retention": count(r1, 7, "PAST RETENTION"), "not measured": count(r1, 7, "NOT MEASURED") + count(r2, 6, "NOT MEASURED"), "recoverable beyond policy": count(r2, 6, "RECOVERABLE BEYOND"), "indefinite snapshot": count(r2, 6, "INDEFINITE"), "not masked or mask not active": sum(1 for r in r3 if "NOT MASKED" in r[5] or "MASK NOT ACTIVE" in r[5]), "clear reader not on policy list": sum(1 for r in r3 if "CLEAR READER" in r[5]), "no policy for category": count(r3, 5, "NO POLICY")} total = sum(flags.values()) print("\n== Result") print(" follow-up: " + (", ".join(f"{v} {k}" for k, v in flags.items() if v) or "none")) print(f" input checks: {len(checks) - len(failed)} of {len(checks)} passed" + ( "; the lists above are INCOMPLETE" if failed else "")) if not_checked: print(" NOT CHECKED: " + ", ".join(not_checked)) print(f" RESULT: {total} item(s) need follow-up; {len(failed)} input check(s) failed; " f"{len(not_checked)} platform(s) not checked.") return 1 if "--strict" in argv and (total or failed or not_checked) else 0 if __name__ == "__main__": sys.exit(main())