#!/usr/bin/env python3
"""Rate every standard and implementation specification of the HIPAA Security Rule for what a data
platform's own logs can show.

The text is 45 CFR Part 164 Subpart C (Security Standards for the Protection of Electronic Protected Health
Information) from the eCFR, pinned to the 2026-10-01 version. The script lists every standard and
implementation specification in sections 164.308 to 164.316, checks that list against the rule's own
Appendix A matrix, then joins it to the rating table RATINGS below (our judgment, written out so you can
change it) and counts.

    python3 hipaa_log_evidence.py --out hipaa_log_evidence.csv
    python3 hipaa_log_evidence.py --out planted.csv --plant duplicate-row,renamed-item,drop-item

Needs Python 3 and the duckdb module (tested on Python 3.12, DuckDB 1.5.6). It reads only the public rule text.
It exits 1, and writes no CSV, if one of its own checks fails.

Ratings (the words are ours; the criteria are the three lines below):
  Direct   a platform's own log, by itself, shows the thing the text asks for happened or is in place
  Partial  the log shows part of it; another record (a policy, a request, an HR date, a sign-off) shows the rest
  None     what the text asks for is not something a data platform's logs record
"Direct" is evidence a person can put in front of an assessor, not a finding that the entity complies.
"""
import argparse
import csv
import gzip
import hashlib
import html
import json
import os
import re
import sys
import urllib.request

import duckdb

DATE = "2026-10-01"
BASE = "https://www.ecfr.gov/api/versioner/v1"
RULE_URL = f"{BASE}/full/{DATE}/title-45.xml?part=164&subpart=C"
VERSIONS_URL = f"{BASE}/versions/title-45.json?part=164&subpart=C"
UA = os.environ.get("ECFR_USER_AGENT", "hipaa-log-evidence-script (https://www.datameaning.com/guides/)")
SECTIONS = {"164.308": "Administrative", "164.310": "Physical", "164.312": "Technical",
            "164.314": "Organizational", "164.316": "Documentation"}
MATRIX_SECTIONS = ("164.308", "164.310", "164.312")  # Appendix A covers these three only
PLANTS = {"duplicate-row", "renamed-item", "drop-item"}
ROMAN = {"i", "ii", "iii", "iv", "v", "vi", "vii", "viii"}

# citation -> (name as the rule words it, rating, what a log shows, what it cannot show).
# These ratings are our judgment. Change them to yours and re-run.
R = "Direct", "Partial", "None"
NONE = ("None", "", "")
RATINGS = {
    "164.308(a)(1)": ("Security management process", "Partial", "access and login logs are the records used to detect violations", "the policies, and proof that violations were contained and corrected, are not in a log"),
    "164.308(a)(1)(ii)(A)": ("Risk analysis", "None", "", "an assessment document, not a platform event"),
    "164.308(a)(1)(ii)(B)": ("Risk management", "None", "", "decisions on which measures to take are not logged"),
    "164.308(a)(1)(ii)(C)": ("Sanction policy", "None", "", "an HR action"),
    "164.308(a)(1)(ii)(D)": ("Information system activity review", "Partial", "audit logs are the records to be reviewed, and a scheduled review query leaves a run record", "who reviewed what, and what they decided, is a sign-off"),
    "164.308(a)(2)": ("Assigned security responsibility", "None", "", "naming the security official is a document"),
    "164.308(a)(3)": ("Workforce security", "Partial", "grant history shows who has access to what and when it changed", "whether that access is appropriate for each role is a policy decision"),
    "164.308(a)(3)(ii)(A)": ("Authorization and/or supervision", "Partial", "access history shows what each user read", "the authorization or supervision is a manager's record"),
    "164.308(a)(3)(ii)(B)": ("Workforce clearance procedure", "None", "", "clearance is an HR decision"),
    "164.308(a)(3)(ii)(C)": ("Termination procedures", "Partial", "grant history shows when each access ended", "the termination date comes from HR, so on-time needs that record"),
    "164.308(a)(4)": ("Information access management", "Partial", "grant history shows access given and removed", "the policies for authorizing access are not in a log"),
    "164.308(a)(4)(ii)(A)": ("Isolating health care clearinghouse functions", "Partial", "grants show whether clearinghouse data is reachable from other roles", "whether the separation was designed and approved is a document"),
    "164.308(a)(4)(ii)(B)": ("Access authorization", "Partial", "grant history shows who granted access, to whom and when", "the approval sits in a request or ticket system"),
    "164.308(a)(4)(ii)(C)": ("Access establishment and modification", "Partial", "grant, revoke and role-change records show who changed access and when", "the review the text asks for is a sign-off"),
    "164.308(a)(5)": ("Security awareness and training", "None", "", "training records are not platform logs"),
    "164.308(a)(5)(ii)(A)": ("Security reminders", "None", "", "reminders are sent outside the data platform"),
    "164.308(a)(5)(ii)(B)": ("Protection from malicious software", "None", "", "endpoint tools, not the data platform, record this"),
    "164.308(a)(5)(ii)(C)": ("Log-in monitoring", "Partial", "login history records each attempt, successes and failures", "monitoring and reporting discrepancies needs an alert and a response record"),
    "164.308(a)(5)(ii)(D)": ("Password management", "Partial", "login history shows which authentication method each login used", "password rules are a setting, not a log"),
    "164.308(a)(6)": ("Security incident procedures", "Partial", "logs show what happened during an incident", "the procedure is a document"),
    "164.308(a)(6)(ii)": ("Response and reporting", "Partial", "logs let a team identify suspected incidents and their reach", "response, mitigation and outcome records are kept elsewhere"),
    "164.308(a)(7)": ("Contingency plan", "None", "", "a plan is a document"),
    "164.308(a)(7)(ii)(A)": ("Data backup plan", "Partial", "if backups run as jobs, job history shows each run", "that a copy is retrievable and exact is shown by a restore test, not a log"),
    "164.308(a)(7)(ii)(B)": ("Disaster recovery plan", "None", "", "restore procedures are documents and tests"),
    "164.308(a)(7)(ii)(C)": ("Emergency mode operation plan", "None", "", "a plan is a document"),
    "164.308(a)(7)(ii)(D)": ("Testing and revision procedures", "None", "", "test results are kept outside platform logs"),
    "164.308(a)(7)(ii)(E)": ("Applications and data criticality analysis", "None", "", "an analysis is a document"),
    "164.308(a)(8)": ("Evaluation", "None", "", "an evaluation report is a document"),
    "164.308(b)(1)": ("Business associate contracts and other arrangements", "None", "", "contracts are not platform logs"),
    "164.308(b)(3)": ("Written contract or other arrangement", "None", "", "a contract is a document"),
    "164.310(a)(1)": ("Facility access controls", "None", "", "physical access is not a data platform event"),
    "164.310(a)(2)(i)": ("Contingency operations", "None", "", "physical access is not a data platform event"),
    "164.310(a)(2)(ii)": ("Facility security plan", "None", "", "physical safeguards"),
    "164.310(a)(2)(iii)": ("Access control and validation procedures", "None", "", "physical access is not a data platform event"),
    "164.310(a)(2)(iv)": ("Maintenance records", "None", "", "repairs to physical components are not logged by the platform"),
    "164.310(b)": ("Workstation use", "None", "", "workstation rules are policy"),
    "164.310(c)": ("Workstation security", "None", "", "physical safeguards"),
    "164.310(d)(1)": ("Device and media controls", "None", "", "hardware and media movements are not platform events"),
    "164.310(d)(2)(i)": ("Disposal", "None", "", "disposal of hardware or media is physical"),
    "164.310(d)(2)(ii)": ("Media re-use", "None", "", "physical media"),
    "164.310(d)(2)(iii)": ("Accountability", "None", "", "movement of hardware is physical"),
    "164.310(d)(2)(iv)": ("Data backup and storage", "None", "", "copies made before moving equipment are not logged by the platform"),
    "164.312(a)(1)": ("Access control", "Partial", "grant history and login history show who can reach the data and who did", "whether each person should have that access is not in a log"),
    "164.312(a)(2)(i)": ("Unique user identification", "Partial", "the user list and login history tie every action to a named user", "a login shared by several people looks like one user"),
    "164.312(a)(2)(ii)": ("Emergency access procedure", "Partial", "use of an emergency account shows in login and access history", "that the procedure exists and was approved is a document"),
    "164.312(a)(2)(iii)": ("Automatic logoff", "None", "", "an idle timeout is a setting; a log of it being applied is rarely kept"),
    "164.312(a)(2)(iv)": ("Encryption and decryption", "None", "", "encryption is a setting, not a logged event"),
    "164.312(b)": ("Audit controls", "Direct", "the platform's access and query history is a mechanism that records activity", "it covers only what the platform records, for as long as it keeps it"),
    "164.312(c)(1)": ("Integrity", "Partial", "query and access history show who changed or deleted data", "whether a change was improper is not in a log"),
    "164.312(c)(2)": ("Mechanism to authenticate electronic protected health information", "None", "", "checksums and signatures are a separate mechanism"),
    "164.312(d)": ("Person or entity authentication", "Partial", "login history records each attempt, the method used and whether it succeeded", "it does not show that the person is who they claim; the identity provider does that"),
    "164.312(e)(1)": ("Transmission security", "None", "", "network protection is not a platform log"),
    "164.312(e)(2)(i)": ("Integrity controls", "None", "", "not a platform log"),
    "164.312(e)(2)(ii)": ("Encryption", "None", "", "encryption in transit is a setting, not a logged event"),
    "164.314(a)(1)": ("Business associate contracts or other arrangements", "None", "", "contracts are not platform logs"),
    "164.314(a)(2)": ("Business associate contracts or other arrangements (implementation specifications)", "None", "", "contract terms are documents"),
    "164.314(b)(1)": ("Requirements for group health plans", "None", "", "plan documents are not platform logs"),
    "164.314(b)(2)": ("Requirements for group health plans (implementation specifications)", "None", "", "plan documents are not platform logs"),
    "164.316(a)": ("Policies and procedures", "None", "", "policies are documents"),
    "164.316(b)(1)": ("Documentation", "Partial", "an electronic log can be the record of an activity the rule requires to be documented", "the policies and assessments must be documented too"),
    "164.316(b)(2)(i)": ("Time limit", "Partial", "the log's own retention period shows how long records are kept", "the rule asks 6 years for documentation, and a log's retention is a separate setting"),
    "164.316(b)(2)(ii)": ("Availability", "None", "", "availability of documents to staff is not a platform event"),
    "164.316(b)(2)(iii)": ("Updates", "None", "", "document reviews are not platform events"),
}
RATING_WORDS = ["Direct", "Partial", "None"]


def get(url):
    req = urllib.request.Request(url, headers={"Accept-Encoding": "gzip", "User-Agent": UA})
    with urllib.request.urlopen(req, timeout=120) as r:
        raw = r.read()
        return gzip.decompress(raw) if r.headers.get("Content-Encoding") == "gzip" else raw


def strip(s):
    return html.unescape(re.sub(r"<(?!/?I>)[^>]+>", "", s)).strip()


def parse_rule(xml):
    """Return the standards and implementation specifications of 164.308-164.316 in rule order."""
    items, std_name = [], ""
    for sec in re.finditer(r'<DIV8 N="(164\.\d+)"[^>]*>(.*?)</DIV8>', xml, re.S):
        number, body = sec.group(1), sec.group(2)
        if number not in SECTIONS:
            continue
        path = {}
        for p in re.findall(r"<P>(.*?)</P>", body, re.S):
            p = strip(p)
            m = re.match(r"^((?:\([A-Za-z0-9]+\))+)", p)
            labels = re.findall(r"\(([A-Za-z0-9]+)\)", m.group(1)) if m else []
            for lab in labels:
                if lab.isdigit():
                    lvl = 2
                elif lab.isupper():
                    lvl = 4
                elif lab in ROMAN and 2 in path:
                    lvl = 3
                else:
                    lvl = 1
                path[lvl] = lab
                for deeper in [k for k in path if k > lvl]:
                    del path[deeper]
            full = number + "".join(f"({path[k]})" for k in sorted(path))
            short = number + "".join(f"({path[k]})" for k in sorted(path) if k <= 2)
            std = re.search(r"<I>Standard: ([^<]+?)\.</I>", p)
            ba = re.search(r"<I>(Business associate contracts and other arrangements)\.</I>", p)
            spec = re.search(r"<I>(?:Implementation specifications?: )?([^<]+?) \((Required|Addressable)\)\.?</I>", p)
            if std or ba:
                std_name = (std or ba).group(1)
                items.append({"citation": short, "kind": "Standard", "name": std_name, "designation": "Standard",
                              "section": number})
            if spec:
                name = spec.group(1)
                if name == "Implementation specifications":
                    name = f"{std_name.rstrip('.')} (implementation specifications)"
                items.append({"citation": full, "kind": "Specification", "name": name,
                              "designation": spec.group(2), "section": number})
    return items


def parse_matrix(xml):
    """Appendix A rows -> {citation: (standard name, [(spec name, 'R'/'A'/None)], standard marker)}."""
    app = xml[xml.index('N="Appendix A to Subpart C'):]
    out, cur = {}, None
    for tr in re.findall(r"<TR>(.*?)</TR>", app, re.S):
        cells = [strip(c) for c in re.findall(r"<TD[^>]*?(?:/>|>(.*?)</TD>)", tr, re.S)]
        if len(cells) != 3:
            continue
        name, cit, spec = cells
        if cit:
            cur = cit
            out[cur] = {"name": name, "specs": [], "marker": None}
        if cur is None:
            continue
        m = re.match(r"^(.*?)\s*\(([RA])\)$", spec)
        if m and m.group(1) == "":
            out[cur]["marker"] = m.group(2)
        elif m:
            out[cur]["specs"].append((m.group(1), m.group(2)))
        else:
            out[cur]["specs"].append((spec, None))
    return out


def norm(s):
    return re.sub(r"[^a-z0-9]+", " ", s.lower()).strip()


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--out", required=True, help="CSV to write")
    ap.add_argument("--plant", default="", help="comma list: duplicate-row, renamed-item, drop-item (test faults)")
    a = ap.parse_args()
    plant = {x for x in a.plant.split(",") if x}
    if plant - PLANTS:
        ap.error(f"unknown --plant value(s) {sorted(plant - PLANTS)}; use {sorted(PLANTS)}")
    print(f"duckdb {duckdb.__version__}, python {sys.version.split()[0]}")

    versions = json.loads(get(VERSIONS_URL))["content_versions"]
    latest = max(v["amendment_date"] for v in versions if v["amendment_date"] <= DATE)
    xml_bytes = get(RULE_URL)
    print(f"source  {RULE_URL}")
    print(f"        bytes={len(xml_bytes)} sha256={hashlib.sha256(xml_bytes).hexdigest()[:16]}... "
          f"latest amendment of the subpart={latest}")
    xml = xml_bytes.decode("utf-8")

    items = parse_rule(xml)
    matrix = parse_matrix(xml)
    unmarked = [f"{c} {s}" for c, m in matrix.items() for s, d in m["specs"] if d is None and s]
    print(f"Appendix A rows with no (R) or (A) marker, where the rule text gives one: {unmarked or 'none'}")
    if "duplicate-row" in plant:
        items.append(dict(next(i for i in items if i["citation"] == "164.308(a)(3)(ii)(C)")))
        print("PLANTED: 'Termination procedures' listed twice")
    if "renamed-item" in plant:
        next(i for i in items if i["citation"] == "164.312(b)")["name"] = "Audit control"
        print("PLANTED: 164.312(b) 'Audit controls' renamed 'Audit control'")
    if "drop-item" in plant:
        items = [i for i in items if i["citation"] != "164.312(d)"]
        print("PLANTED: 164.312(d) 'Person or entity authentication' dropped from the parsed list")

    con = duckdb.connect()
    con.execute("CREATE TABLE parsed (pos INTEGER, citation VARCHAR, kind VARCHAR, name VARCHAR, "
                "designation VARCHAR, section VARCHAR)")
    con.executemany("INSERT INTO parsed VALUES (?, ?, ?, ?, ?, ?)",
                    [(n, i["citation"], i["kind"], i["name"], i["designation"], i["section"])
                     for n, i in enumerate(items)])
    con.execute("CREATE TABLE ratings (citation VARCHAR, name VARCHAR, rating VARCHAR, evidence VARCHAR, gap VARCHAR)")
    con.executemany("INSERT INTO ratings VALUES (?, ?, ?, ?, ?)", [(k, *v) for k, v in RATINGS.items()])
    con.execute("CREATE TABLE sections (section VARCHAR, safeguard VARCHAR)")
    con.executemany("INSERT INTO sections VALUES (?, ?)", list(SECTIONS.items()))

    joined = con.execute("""
        SELECT p.pos, p.citation, p.kind, p.name, p.designation, s.safeguard,
               COALESCE(r.rating, 'Unmapped') AS rating, COALESCE(r.evidence, '') AS evidence,
               COALESCE(r.gap, '') AS gap
        FROM parsed p JOIN sections s USING (section)
        LEFT JOIN ratings r ON r.citation = p.citation
        ORDER BY p.pos""").fetchall()

    # The script's own checks. Each one names the planted fault that trips it.
    failures = []
    n = len(joined)
    # 1. one row per citation (trips: duplicate-row)
    dup = con.execute("SELECT citation, COUNT(*) FROM parsed GROUP BY 1 HAVING COUNT(*) > 1").fetchall()
    for c, k in dup:
        failures.append(f"{c} appears {k} times in the parsed rule")
    # 2. parsed list equals Appendix A for sections 308-312: same standards, same number of
    #    specifications under each, same Required/Addressable letter where the matrix gives one
    #    (trips: duplicate-row, drop-item)
    for cit, mrow in matrix.items():
        sec = cit.split("(")[0]
        if sec not in MATRIX_SECTIONS:
            continue
        std = [i for i in items if i["citation"] == cit and i["kind"] == "Standard"]
        if not std:
            failures.append(f"Appendix A lists {cit} ({mrow['name']}) but the parsed rule has no such standard")
            continue
        specs = [i for i in items if i["kind"] == "Specification" and spec_parent(i["citation"]) == cit]
        if len(specs) != len(mrow["specs"]):
            failures.append(f"{cit}: parsed rule has {len(specs)} specifications, Appendix A has {len(mrow['specs'])}")
        for sp, (mname, mdes) in zip(specs, mrow["specs"]):
            if mdes and mdes != sp["designation"][0]:
                failures.append(f"{sp['citation']}: rule says {sp['designation']}, Appendix A says ({mdes})")
    parsed_m = sum(1 for i in items if i["section"] in MATRIX_SECTIONS and i["kind"] == "Standard")
    matrix_std = sum(1 for c in matrix if c.split("(")[0] in MATRIX_SECTIONS)
    if parsed_m != matrix_std and not any("Appendix A lists" in f for f in failures):
        failures.append(f"parsed {parsed_m} standards in 164.308-164.312, Appendix A lists {matrix_std}")
    # 3. every parsed item is rated (trips: an item the parser finds that RATINGS has no key for)
    for r in joined:
        if r[6] == "Unmapped":
            failures.append(f"{r[1]} '{r[3]}' has no rating in RATINGS")
    # 4. the rating's key matches an item with the same name; a key that matches nothing is an error
    #    (trips: drop-item, renamed-item)
    by_cit = {r[1]: r for r in joined}
    for cit, (name, *_rest) in RATINGS.items():
        if cit not in by_cit:
            failures.append(f"RATINGS key {cit} ('{name}') matches no item in the parsed rule")
        elif norm(by_cit[cit][3]) != norm(name):
            failures.append(f"RATINGS key {cit} expects '{name}' but the rule says '{by_cit[cit][3]}'")
    # 5. ratings are valid words and the three shares add up to 100% (trips: a typo in RATINGS)
    bad = [r[1] for r in joined if r[6] not in RATING_WORDS and r[6] != "Unmapped"]
    if bad:
        failures.append(f"rating not one of {RATING_WORDS}: {bad}")
    share = con.execute("""SELECT SUM(100.0 * cnt / (SELECT COUNT(*) FROM parsed))
        FROM (SELECT rating, COUNT(*) cnt FROM ratings r JOIN parsed p USING (citation) GROUP BY 1)""").fetchone()[0]
    if share is None or abs(share + 100.0 * sum(1 for r in joined if r[6] == "Unmapped") / n - 100.0) > 1e-9:
        failures.append(f"rating shares add up to {share}, not 100%")
    # 6. every row of the table is counted once in the summary (trips: duplicate-row)
    if n != len(RATINGS):
        failures.append(f"{n} parsed items but {len(RATINGS)} rows in RATINGS")

    def table(sql, head):
        rows = con.execute(sql).fetchall()
        print(head)
        for r in rows:
            print("  " + "  ".join(str(x).rjust(w) if isinstance(x, (int, float)) else str(x).ljust(w)
                                   for x, w in zip(r, WIDTHS)))
        print()

    q = """FROM parsed p JOIN ratings r USING (citation) JOIN sections s USING (section)"""
    global WIDTHS
    print()
    print(f"items rated: {n} ({sum(1 for r in joined if r[2] == 'Standard')} standards, "
          f"{sum(1 for r in joined if r[2] == 'Specification')} implementation specifications)")
    print()
    WIDTHS = [28, 7, 7, 7, 7, 8]
    table(f"""SELECT 'all items', COUNT(*), COUNT(*) FILTER (rating='Direct'), COUNT(*) FILTER (rating='Partial'),
              COUNT(*) FILTER (rating='None'), ROUND(100.0 * COUNT(*) FILTER (rating <> 'None') / COUNT(*), 1) {q}""",
          "by what                         items  Direct  Partial  None  % with some log evidence")
    table(f"""SELECT safeguard || ' (' || section || ')', COUNT(*), COUNT(*) FILTER (rating='Direct'),
              COUNT(*) FILTER (rating='Partial'), COUNT(*) FILTER (rating='None'),
              ROUND(100.0 * COUNT(*) FILTER (rating <> 'None') / COUNT(*), 1) {q}
              GROUP BY safeguard, section ORDER BY section""", "by safeguard")
    table(f"""SELECT designation, COUNT(*), COUNT(*) FILTER (rating='Direct'), COUNT(*) FILTER (rating='Partial'),
              COUNT(*) FILTER (rating='None'),
              ROUND(100.0 * COUNT(*) FILTER (rating <> 'None') / COUNT(*), 1) {q}
              GROUP BY designation ORDER BY designation""", "by designation in the rule")
    print("Direct and Partial items:")
    for r in joined:
        if r[6] in ("Direct", "Partial"):
            print(f"  {r[6]:<8}{r[1]:<24}{r[3]}")
    print()
    if failures:
        print("SELF-CHECK FAILED: no CSV written")
        for f in failures:
            print("  -", f)
        sys.exit(1)
    print(f"SELF-CHECK PASSED: {n} items parsed, no citation twice, parsed list agrees with Appendix A "
          f"({matrix_std} standards in 164.308-164.312), every item rated, every rating key matched, "
          f"shares add up to 100%")
    with open(a.out, "w", newline="") as f:
        w = csv.writer(f, lineterminator="\n")
        w.writerow(["citation", "kind", "name", "designation", "safeguard", "rating", "log_shows", "log_cannot_show"])
        for r in joined:
            w.writerow([r[1], r[2], r[3], r[4], r[5], r[6], r[7], r[8]])
    print(f"wrote {a.out}")


def spec_parent(citation):
    """164.308(a)(1)(ii)(A) -> 164.308(a)(1); 164.310(d)(2)(i) -> 164.310(d)(1); 164.312(c)(2) -> 164.312(c)(1);
    164.312(b) stays itself. The rule numbers a standard (x)(1) and its specifications (x)(2) in some places."""
    base = citation.split("(")[0]
    parts = re.findall(r"\([A-Za-z0-9]+\)", citation)
    if base == "164.308" and parts[:2] == ["(b)", "(3)"]:
        return "164.308(b)(1)"
    if base != "164.308" and len(parts) >= 2 and parts[1] == "(2)":
        return f"{base}{parts[0]}(1)"
    return base + "".join(parts[:2])


if __name__ == "__main__":
    main()
