#!/usr/bin/env python3 """Rule engine and scoring for the crm-audit skill. Usage (from a private working folder; SKILL_DIR is this skill's folder): python3 -I "$SKILL_DIR/scripts/audit_crm.py" [--fields fields.json] [--automations automations.json] [--history history.json] [--config audit-config.json] [--as-of YYYY-MM-DD] [--scope full|gates] [--json] [--md audit-work/report.md] The inventory is the JSON the agent fills read-only (assets/crm-inventory-template.json, references/inventory-format.md). Fragments from normalize_exports.py and writer_history.py are merged in memory; the inventory file is never rewritten. A rule whose inputs are missing is NOT CHECKED and names the missing key. A rule that also has a NOT CHECKED part (a critical field without history, an automation whose writes or actions were never read) cannot pass, and a gate with one cannot reach a ready verdict. Agent judgements in manual_results fill or override script results; a downgrade whose decision is not a dated decisions.items entry is rejected with a warning. Reading map: this file DEFAULTS, RULES and SECTIONS (thresholds, the 56 rules with class and gate, the inventory sections each reads); validate_inventory (a problem stops the run with exit code 2); build_inventory and evaluate (fragment merge, coverage, not_applicable, manual_results); score and verdicts (references/scoring.md); result, text_report and main (file reads and the one optional write, --md, through write_report.py) audit_common.py value helpers, Res (one rule's result) and Ctx (the merged inventory; Ctx.writers() gathers every writer of one field) rules_data_model.py OBJ and FLD rules one function per rule, named after its ID (OBJ-01 rules_process.py PIPE, LIFE and AUTO is obj_01), in the order of the reference file rules_access_integrations_reporting.py with the same name: ACC, INT and REP check_fixture.py the regression cases in assets/regression-fixture.json Reads local files only: no network, no third-party modules, Python 3.9 or later; run with python3 -I. Exit code: 1 when any rule is FAIL, 2 on bad input (including an inventory that fails validation or a rule that cannot run on it), 0 otherwise. """ import argparse import copy import datetime as dt import importlib.util import json import math import os import sys HERE = os.path.dirname(os.path.abspath(__file__)) def load_sibling(name): """Import another script from this folder by path and register it under its file name, so the rule modules can import audit_common (python3 -I keeps the script folder off sys.path). No bytecode cache is written, so the skill folder stays unchanged.""" path = os.path.join(HERE, name + ".py") mod = sys.modules.get(name) if mod is not None and getattr(mod, "__file__", None) == path: return mod sys.dont_write_bytecode = True spec = importlib.util.spec_from_file_location(name, path) mod = importlib.util.module_from_spec(spec) sys.modules[name] = mod spec.loader.exec_module(mod) return mod load_sibling("audit_common") from audit_common import (STATUS_ORDER, ENUM_VALUES, alnum, ratio, parse_date, empty, action_label, Res, Ctx, # noqa: E402 hist_for) DEFAULTS = { "OBJ-04": {"open_deals_without_company": {"warn": 0.02, "fail": 0.10}, "sales_contacts_without_company": {"warn": 0.10, "fail": 0.25}, "open_deals_without_contact_warn": 0.10}, "OBJ-05": {"shared_domain_companies": {"warn": 0.02, "fail": 0.05}, "vendor_duplicate_contacts_warn": 0.01}, "FLD-02": {"window_days": 30, "flip_flop_records_fail": 3, "flip_flop_rate_fail": 0.01, "flip_flop_records_warn": 1}, "FLD-05": {"unused_min_age_days": 90, "capacity": {"warn": 0.80, "fail": 0.95}}, "FLD-06": {"undocumented_custom_warn": 0.25}, "FLD-08": {"routing_inputs": {"warn": 0.95, "fail": 0.80}, "reporting_dimensions": {"warn": 0.90, "fail": 0.70}}, "PIPE-02": {"max_open_stages": 8}, "PIPE-04": {"missing_required": {"warn": 0.05, "fail": 0.15}}, "PIPE-05": {"unused_stage_days": 180, "dwell_min_days": 90, "dwell_median_multiple": 2, "stuck_share_warn": 0.20}, "LIFE-06": {"customer_without_evidence": {"warn": 0.02, "fail": 0.05}, "opportunity_without_open_deal_warn": 0.10, "recent_empty_stage_warn": 0.05, "recent_days": 90}, "AUTO-01": {"undocumented_active_warn": 0.20}, "AUTO-08": {"dormant_days": 90}, "ACC-01": {"max_privileged_min": 3, "max_privileged_share": 0.10}, "ACC-03": {"inactive_days": 90}, "ACC-04": {"bulk_delete_share_warn": 0.20}, "INT-03": {"error_rate": {"warn": 0.01, "fail": 0.05}, "open_days_warn": 7, "suspension_lookback_days": 30}, "INT-04": {"salesforce_daily_api": {"warn": 0.70, "fail": 0.90}, "hubspot_private_app_daily_warn": 0.80}, "INT-05": {"unused_token_days": 90}, "REP-02": {"reconcile_tolerance": 0.01}, "REP-05": {"past_close_date": {"warn": 0.10, "fail": 0.25}, "missing_amount": {"warn": 0.10, "fail": 0.30}, "lost_reason_fill": {"warn": 0.80, "fail": 0.50}, "lookback_days": 180}, "REP-06": {"unknown_source": {"warn": 0.30, "fail": 0.50}, "lookback_days": 90}, "scoring": {"weights": {"A": 3, "B": 2, "C": 1}, "credit": {"PASS": 1.0, "WARN": 0.5, "FAIL": 0.0}, "coverage_min": 0.60, "coverage_full": 0.85, "class_a_cap": 60, "round_to": 5, "min_domains_for_overall": 6}, } DEFAULT_LISTS = { "field_prefixes_to_strip": ["hs_", "hubspot_", "sf_", "salesforce_", "stripe_", "clay_", "zi_", "zoominfo_", "apollo_", "clearbit_", "attio_"], "field_suffixes_to_strip": ["__c", "__pc", "__r"], "field_words_to_strip": ["current", "new", "old", "copy", "latest", "primary"], "owner_fields": ["hubspot_owner_id", "ownerid", "owner_id", "owner"], } DOMAINS = [("OBJ", "Objects, associations and identity"), ("FLD", "Properties and field governance"), ("PIPE", "Pipelines and stages"), ("LIFE", "Lifecycle and lead status"), ("AUTO", "Automation and workflows"), ("ACC", "Users, permissions and teams"), ("INT", "Integrations and sync ownership"), ("REP", "Reporting readiness")] RULES = [ ("OBJ-01", "Each object and pipeline has one meaning", "B", None), ("OBJ-02", "Every integration that creates or updates records matches on a stable unique key", "A", "automation"), ("OBJ-03", "Auto-creation and auto-association settings fit the identity design", "A", None), ("OBJ-04", "Required associations exist", "B", None), ("OBJ-05", "Duplicate creation is under control", "B", None), ("OBJ-06", "Merges cannot silently break integration mappings", "B", None), ("FLD-01", "Critical fields are registered", "B", None), ("FLD-02", "One automated writer per critical field, or a written precedence rule", "A", "automation"), ("FLD-03", "Consent and marketing status have only permitted writers", "A", "automation"), ("FLD-04", "No near-duplicate or numbered fields split one meaning", "B", None), ("FLD-05", "Unused fields are known and capacity has headroom", "C", None), ("FLD-06", "Custom fields are documented", "C", None), ("FLD-07", "Field types and options fit their use", "B", None), ("FLD-08", "Decision-driving fields are filled for their population", "B", None), ("PIPE-01", "Each pipeline represents one sales motion", "C", None), ("PIPE-02", "Stages have entry evidence and a workable count", "B", None), ("PIPE-03", "Won and lost meanings are unambiguous", "A", None), ("PIPE-04", "Stage-required fields are filled whatever the writer", "B", None), ("PIPE-05", "No dead or stuck stages", "C", None), ("PIPE-06", "Integrations reference stage IDs, and stage changes run as migrations", "B", None), ("LIFE-01", "Lifecycle stages describe the relationship and are defined", "B", None), ("LIFE-02", "One field answers one question", "B", None), ("LIFE-03", "Lifecycle writers are known and ordered", "A", "automation"), ("LIFE-04", "Forms and defaults never set late stages", "A", None), ("LIFE-05", "Backward moves are governed", "A", None), ("LIFE-06", "Stage evidence matches the facts", "B", None), ("AUTO-01", "Every active automation has an owner and a purpose", "C", None), ("AUTO-02", "Automations run without errors", "B", None), ("AUTO-03", "No two automations write conflicting values to one field", "A", "automation"), ("AUTO-04", "No broken references", "B", None), ("AUTO-05", "No re-enrolment loops", "B", None), ("AUTO-06", "Destructive and consent-changing actions have a dated decision", "A", "automation"), ("AUTO-07", "Owner assignment protects existing ownership and has a fallback", "B", None), ("AUTO-08", "Dormant automations are reviewed", "C", None), ("AUTO-09", "Automation layers do not overlap on one object", "B", None), ("AUTO-10", "Essential steps do not depend on credits or send limits without a fallback", "B", None), ("ACC-01", "Privileged access is small and personal", "A", "automation"), ("ACC-02", "Integrations use dedicated, least-privilege identities", "A", "automation"), ("ACC-03", "Departed and inactive users do not own live work", "B", None), ("ACC-04", "Bulk destructive rights are limited", "B", None), ("ACC-05", "Sensitive fields and records are restricted on purpose", "A", None), ("ACC-06", "Authentication and team structure support the permission model", "B", None), ("INT-01", "The integration inventory is complete and owned", "A", "automation"), ("INT-02", "Each mapped field has one direction and defined blank and clear behaviour", "A", None), ("INT-03", "Syncs are healthy", "B", None), ("INT-04", "API capacity has headroom", "B", None), ("INT-05", "Credentials are current and owned", "B", None), ("INT-06", "Known conflicting combinations are absent", "A", None), ("INT-07", "Enrichment and reverse ETL cannot overwrite owned fields", "A", None), ("REP-01", "Headline metrics are defined", "B", "reporting"), ("REP-02", "Headline numbers reconcile to record lists", "A", "reporting"), ("REP-03", "Reports count at the right grain", "A", "reporting"), ("REP-04", "One currency policy", "A", "reporting"), ("REP-05", "Pipeline hygiene supports forecasting", "B", None), ("REP-06", "Source attribution coverage", "B", None), ("REP-07", "Exclusions and funnel timestamps are consistent", "C", None), ] RULE = {r[0]: {"id": r[0], "name": r[1], "class": r[2], "gate": r[3], "domain": r[0].split("-")[0]} for r in RULES} # Inventory sections each rule reads. An "unavailable" section makes the rule NOT CHECKED. SECTIONS = { "OBJ-01": ["objects"], "OBJ-02": ["integrations", "objects"], "OBJ-03": ["settings"], "OBJ-04": ["counts"], "OBJ-05": ["counts"], "OBJ-06": ["integrations"], "FLD-01": ["critical_fields"], "FLD-02": ["critical_fields"], "FLD-03": ["critical_fields"], "FLD-04": ["fields"], "FLD-05": ["fields"], "FLD-06": ["fields"], "FLD-07": ["fields"], "FLD-08": ["critical_fields"], "PIPE-01": ["pipelines"], "PIPE-02": ["pipelines"], "PIPE-03": ["pipelines"], "PIPE-04": ["pipelines"], "PIPE-05": ["pipelines"], "PIPE-06": ["pipelines", "integrations"], "LIFE-01": ["lifecycle"], "LIFE-02": ["lifecycle"], "LIFE-03": ["lifecycle"], "LIFE-04": ["forms"], "LIFE-05": ["automations"], "LIFE-06": ["counts"], "AUTO-01": ["automations"], "AUTO-02": ["automations"], "AUTO-03": ["automations"], "AUTO-04": ["automations", "fields"], "AUTO-05": ["automations"], "AUTO-06": ["automations"], "AUTO-07": ["automations"], "AUTO-08": ["automations"], "AUTO-09": ["automations"], "AUTO-10": ["automations"], "ACC-01": ["users"], "ACC-02": ["integrations"], "ACC-03": ["users"], "ACC-04": ["users"], "ACC-05": ["access"], "ACC-06": ["settings"], "INT-01": ["integrations"], "INT-02": ["integrations"], "INT-03": ["integrations"], "INT-04": ["api_capacity"], "INT-05": ["integrations"], "INT-06": ["integrations"], "INT-07": ["integrations"], "REP-01": ["reporting"], "REP-02": ["reporting"], "REP-03": ["reporting"], "REP-04": ["settings"], "REP-05": ["counts"], "REP-06": ["counts"], "REP-07": ["reporting"], } # Each rule's function lives in the module for its domain and is named after the rule ID (OBJ-01 is obj_01). RULE_MODULES = {"OBJ": "rules_data_model", "FLD": "rules_data_model", "PIPE": "rules_process", "LIFE": "rules_process", "AUTO": "rules_process", "ACC": "rules_access_integrations_reporting", "INT": "rules_access_integrations_reporting", "REP": "rules_access_integrations_reporting"} EVALUATORS = {rid: getattr(load_sibling(RULE_MODULES[RULE[rid]["domain"]]), rid.lower().replace("-", "_")) for rid in RULE} # Rules whose writer or action lists can miss what an unclassified workflow action does. UNCLASSIFIED_RULES = ("FLD-02", "FLD-03", "LIFE-03", "AUTO-03", "AUTO-06", "AUTO-09") class BadInput(Exception): pass # ---------- configuration and inventory validation ---------- def deep_merge(base, over): out = copy.deepcopy(base) for k, v in (over or {}).items(): if isinstance(v, dict) and isinstance(out.get(k), dict): out[k] = deep_merge(out[k], v) else: out[k] = copy.deepcopy(v) return out def leaf_paths(d, prefix=""): for k, v in d.items(): p = prefix + "." + k if prefix else k if isinstance(v, dict): for x in leaf_paths(v, p): yield x else: yield p, v def get_path(d, path): cur = d for part in path.split("."): if not isinstance(cur, dict) or part not in cur: return None cur = cur[part] return cur SECTION_TYPES = { "meta": dict, "coverage": dict, "objects": list, "settings": dict, "counts": dict, "fields": list, "critical_fields": list, "pipelines": list, "lifecycle": dict, "forms": list, "automations": list, "users": list, "integrations": list, "api_capacity": dict, "reporting": dict, "decisions": dict, "access_checks": list, "record_access": list, "archive_candidates": list, "gate_owners": dict, "manual_results": list, "evidence": list, "history": list, } def is_placeholder(sec, x): """Template entries that sanitize() drops are not validated.""" if sec == "users": return x.get("active") is None and x.get("privileged") is None if sec == "automations": return empty(x.get("id")) and empty(x.get("name")) key = PLACEHOLDER_KEYS.get(sec) return key is not None and (empty(x.get(key)) or str(x.get(key)).startswith("<")) def validate_inventory(inv): """Problems (template hints left in place, wrong types) stop the run; unknown enum values only warn. Enum spellings are rewritten in place to the forms in ENUM_VALUES.""" problems, warns = [], [] def name(x, key): return repr(x.get(key)) if not empty(x.get(key)) else "?" def enum(path, holder, key, kind): v = holder.get(key) if v is None or (isinstance(v, str) and not v.strip()): return if not isinstance(v, str): return problems.append("%s must be a string, got %s" % (path, type(v).__name__)) if " | " in v: return problems.append("%s still holds the template hint %r; record one value or null" % (path, v)) hit = next((a for a in ENUM_VALUES[kind] if alnum(a) == alnum(v)), None) if hit is None: warns.append("%s %r is not one of: %s" % (path, v, ", ".join(ENUM_VALUES[kind]))) else: holder[key] = hit def need(path, v, typ): if v is not None and not isinstance(v, typ): problems.append("%s must be %s, got %s" % (path, "a list" if typ is list else "an object", type(v).__name__)) return False return True def entries(sec): for x in inv.get(sec) or []: if not isinstance(x, dict): problems.append("%s holds an entry that is not an object: %r" % (sec, x)) elif not is_placeholder(sec, x): yield x for sec, typ in SECTION_TYPES.items(): need(sec, inv.get(sec), typ) if problems: return problems, warns meta = inv.get("meta") or {} for key in ("crm", "business_model", "scope"): enum("meta." + key, meta, key, key) if isinstance(meta.get("environment"), str) and " | " in meta["environment"]: problems.append("meta.environment still holds the template hint %r; record one value" % meta["environment"]) for sec, cov in (inv.get("coverage") or {}).items(): if isinstance(cov, dict): enum("coverage.%s.status" % sec, cov, "status", "coverage_status") for o in entries("objects"): need("objects[%s].unique_keys" % name(o, "name"), o.get("unique_keys"), list) for f in entries("fields"): need("fields[%s].writers" % name(f, "name"), f.get("writers"), list) for p in entries("pipelines"): pn = "pipelines[%s]" % name(p, "label") if need(pn + ".stages", p.get("stages"), list): for s in p.get("stages") or []: if isinstance(s, dict): enum("%s.stages[%s].type" % (pn, name(s, "label")), s, "type", "stage_type") need("%s.stages[%s].missing_required" % (pn, name(s, "label")), s.get("missing_required"), dict) for u in entries("users"): enum("users[%s].account_type" % name(u, "id"), u, "account_type", "account_type") for i in entries("integrations"): n = "integrations[%s]" % name(i, "name") for key, kind in (("type", "integration_type"), ("identity_type", "identity_type"), ("direction", "direction"), ("blank_semantics", "blank_semantics"), ("stage_mapping", "stage_mapping")): enum("%s.%s" % (n, key), i, key, kind) need(n + ".match_keys", i.get("match_keys"), dict) for key in ("objects", "creates_records", "fields_written", "stage_refs", "two_way_fields_with_split_owners"): need("%s.%s" % (n, key), i.get(key), list) lc = inv.get("lifecycle") or {} for key in ("options", "writers"): need("lifecycle." + key, lc.get(key), list) for w in lc.get("writers") or [] if isinstance(lc.get("writers"), list) else []: if isinstance(w, dict) and not empty(w.get("writer")): enum("lifecycle.writers[%s].kind" % name(w, "writer"), w, "kind", "writer_kind") need("lifecycle.writers[%s].sets" % name(w, "writer"), w.get("sets"), list) for cf in inv.get("critical_fields") or []: if isinstance(cf, dict) and not empty(cf.get("field")): cn = "critical_fields[%s]" % name(cf, "field") for key in ("permitted_writers", "observed_writers", "consumers", "opt_out_reversals"): need("%s.%s" % (cn, key), cf.get(key), list) need(cn + ".fill", cf.get("fill"), dict) for a in entries("automations"): for key in ("writes", "owner_writes", "mutually_exclusive_with", "archived_list_refs", "unknown_action_types"): need("automations[%s].%s" % (name(a, "id"), key), a.get(key), list) for m in entries("manual_results"): enum("manual_results[%s].status" % name(m, "rule"), m, "status", "manual_status") for e in entries("evidence"): enum("evidence[%s].kind" % name(e, "id"), e, "kind", "evidence_kind") for k, v in (inv.get("counts") or {}).items(): need("counts." + k, v, dict) dec, rep = inv.get("decisions") or {}, inv.get("reporting") or {} for key in ("restricted_fields", "items", "automation_order_documented"): need("decisions." + key, dec.get(key), list) for key in ("metrics", "headline_numbers", "reconciliations", "reports"): need("reporting." + key, rep.get(key), list) return problems, warns # ---------- evaluation, manual results, scoring ---------- def merge_by_key(base, overlay, keyfn): out, index = [copy.deepcopy(x) for x in base], {} for n, x in enumerate(out): index[keyfn(x)] = n for x in overlay or []: k = keyfn(x) if k in index: for key, v in x.items(): if v is not None and v != "": # null and "" never override; a list, even [], does out[index[k]][key] = copy.deepcopy(v) else: out.append(copy.deepcopy(x)) return out PLACEHOLDER_KEYS = {"objects": "name", "integrations": "name", "forms": "name", "pipelines": "label", "users": "id", "fields": "name", "automations": "id", "access_checks": "field", "record_access": "object", "manual_results": "rule", "evidence": "rule"} ALWAYS_LISTS = ("manual_results", "access_checks", "record_access", "evidence") def sanitize(inv): """Drop blank template entries, so an unfilled section reads as missing rather than as 'none found'.""" cov = inv.get("coverage") or {} for sec, key in PLACEHOLDER_KEYS.items(): lst = inv.get(sec) if not isinstance(lst, list): continue keep = [x for x in lst if isinstance(x, dict) and not empty(x.get(key)) and not str(x.get(key)).startswith("<")] if sec == "users": keep = [u for u in keep if u.get("active") is not None or u.get("privileged") is not None] if sec == "automations": keep = [a for a in lst if isinstance(a, dict) and not (empty(a.get("id")) and empty(a.get("name")))] status = str((cov.get(sec) or {}).get("status") if isinstance(cov.get(sec), dict) else cov.get(sec) or "").lower() inv[sec] = keep if (keep or status == "complete" or sec in ALWAYS_LISTS) else None for sec, sub, key in (("lifecycle", "options", "label"), ("lifecycle", "writers", "writer"), ("reporting", "metrics", "name"), ("reporting", "reconciliations", "metric"), ("reporting", "reports", "name"), ("decisions", "items", "id")): part = inv.get(sec) if isinstance(part, dict) and isinstance(part.get(sub), list): part[sub] = [x for x in part[sub] if isinstance(x, dict) and not empty(x.get(key))] return inv def build_inventory(inv, fields_frag=None, autos_frag=None, hist_frag=None): inv = copy.deepcopy(inv) fkey = lambda f: (str(f.get("object") or "").lower(), str(f.get("name") or "").lower()) if fields_frag is not None: inv["fields"] = merge_by_key(fields_frag, inv.get("fields"), fkey) if autos_frag is not None: inv["automations"] = merge_by_key(autos_frag, inv.get("automations"), lambda a: str(a.get("id") or a.get("name"))) if hist_frag is not None: inv["history"] = list(inv.get("history") or []) + list(hist_frag) return sanitize(inv) def evaluate(inv, cfg, as_of): try: c = Ctx(inv, cfg, as_of) except Exception as e: # a shape validate_inventory does not cover; report it as bad input, not a crash raise BadInput("the inventory could not be read (%s: %s)" % (type(e).__name__, e)) results, warnings = {}, [] unclassified = c.unclassified() for rid, fn in EVALUATORS.items(): r = Res() blocked = [(s, c.section_state(s)) for s in SECTIONS.get(rid, [])] unavailable = [(s, why) for s, (st, why) in blocked if st == "unavailable"] if unavailable: r.nc("; ".join("%s unavailable%s" % (s, (": " + why) if why else "") for s, why in unavailable)) else: try: fn(c, r) except Exception as e: raise BadInput("rule %s could not run on this inventory (%s: %s); check the keys it reads in " "references/inventory-format.md" % (rid, type(e).__name__, e)) r.close() # NOT CHECKED parts recorded with r.gap: a PASS becomes NOT CHECKED partial = [(s, why) for s, (st, why) in blocked if st == "partial"] for s, why in partial: r.notes.append("partial coverage of %s%s" % (s, (": " + why) if why else "")) if rid in UNCLASSIFIED_RULES and unclassified: r.part("unclassified action types in enabled automations, not checked: %s; open each, add its field " "writes, deletes and consent actions to the inventory overlay, then set unknown_action_types to []" % "; ".join("%r (%s)" % (n, ", ".join(action_label(x) for x in t)) for n, t in unclassified)) results[rid] = r na = cfg.get("not_applicable") or {} for rid, why in na.items(): if rid not in results: continue if RULE[rid]["gate"]: warnings.append("config not_applicable ignored for gate %s: a gate cannot be switched off" % rid) continue results[rid].na = str(why) for m in inv.get("manual_results") or []: rid, st = m.get("rule"), str(m.get("status") or "").upper().replace("NOT_CHECKED", "NOT CHECKED") if rid not in results or st not in ("PASS", "WARN", "FAIL", "NOT CHECKED", "N/A"): warnings.append("manual result ignored: unknown rule or status %r" % m) continue cur = results[rid].status() down = (cur == "FAIL" and st != "FAIL") or (cur == "WARN" and st in ("PASS", "N/A", "NOT CHECKED")) if down and str(m.get("decision") or "").strip() not in c.dated_decisions: warnings.append("manual result for %s would change %s to %s without a decision reference that matches a dated " "decisions.items entry; kept %s" % (rid, cur, st, cur)) continue r = results[rid] text = str(m.get("finding") or "agent judgement") + ((" Evidence: " + str(m["evidence"])) if m.get("evidence") else "") if down: text += " (downgraded from %s per decision %s)" % (cur, m["decision"]) if st == "N/A": r.na = text elif st == "NOT CHECKED": r.items, r.nc_reasons = [], [text] else: if cur == "NOT CHECKED": # the agent rated what the script could not: the judgement replaces the result r.items, r.nc_reasons, r.partial = [], [], [] elif down: # the decision replaces the findings; parts never checked stay NOT CHECKED r.items = [] r.items.insert(0, (st, text)) if str(m.get("class") or "").upper() in ("A", "B", "C"): r.cls = str(m["class"]).upper() r.notes.append("agent judgement recorded in manual_results") return c, results, warnings def round_to(x, step): return int(math.floor(x / float(step) + 0.5) * step) def score(results, cfg): sc = cfg["scoring"] domains = [] for code, name in DOMAINS: rids = [rid for rid in RULE if RULE[rid]["domain"] == code] appl = [rid for rid in rids if results[rid].status() != "N/A"] rated = [rid for rid in appl if results[rid].status() in ("PASS", "WARN", "FAIL")] wt = lambda rid: sc["weights"][results[rid].cls or RULE[rid]["class"]] aw, rw = sum(wt(x) for x in appl), sum(wt(x) for x in rated) cov = (rw / float(aw)) if aw else 0.0 d = {"code": code, "name": name, "coverage": round(cov, 3), "rules": len(rids), "rated": len(rated), "score": None, "band": None} if rated: raw = 100.0 * sum(wt(x) * sc["credit"][results[x].status()] for x in rated) / rw s = round_to(raw, sc["round_to"]) a_fail = any(results[x].status() == "FAIL" and (results[x].cls or RULE[x]["class"]) == "A" for x in rated) capped = a_fail and s > sc["class_a_cap"] if capped: s = sc["class_a_cap"] d.update({"raw": round(raw, 1), "score": s, "class_a_fail": a_fail, "capped": capped}) if not rated or cov < sc["coverage_min"]: d["band"], d["show"] = "Insufficient evidence", False else: s = d["score"] d["band"] = "Healthy" if s >= 85 else "Needs work" if s >= 70 else "At risk" if s >= 50 else "Failing" d["show"], d["partial"] = True, cov < sc["coverage_full"] domains.append(d) shown = [d for d in domains if d.get("show")] weights = cfg.get("domain_weights") or {} overall = None if len(shown) >= sc["min_domains_for_overall"]: tw = sum(float(weights.get(d["code"], 1)) for d in shown) overall = int(round(sum(float(weights.get(d["code"], 1)) * d["score"] for d in shown) / tw)) return domains, overall def verdicts(results, inv): """references/scoring.md. A gate counts as NOT CHECKED when its status is, and also when a PASS or WARN carries a NOT CHECKED part: a reason beside its findings (a critical field without history, an automation whose writes or actions were never read) or a partial entry (an unclassified workflow action).""" owners = inv.get("gate_owners") or {} out = {} for kind, ready in (("automation", "Ready to automate"), ("reporting", "Ready to report from")): gates = [rid for rid in RULE if RULE[rid]["gate"] == kind] st = {g: results[g].status() for g in gates} fails = [g for g in gates if st[g] == "FAIL"] ncs = [g for g in gates if st[g] == "NOT CHECKED"] parts = [g for g in gates if st[g] in ("PASS", "WARN") and (results[g].nc_reasons or results[g].partial)] warns = [g for g in gates if st[g] == "WARN"] unowned = [g for g in warns if not (isinstance(owners.get(g), dict) and owners[g].get("owner") and owners[g].get("date"))] if fails: v = ("Fix gates first", "FAIL: " + ", ".join(fails)) elif ncs or parts: v = ("Not assessable", "; ".join(x for x in ( ("NOT CHECKED: %s (what each needs is listed with the rule)" % ", ".join(ncs)) if ncs else "", ("NOT CHECKED parts: " + "; ".join("%s %s, to collect: %s" % ( g, st[g], " / ".join(dict.fromkeys(results[g].nc_reasons + results[g].partial))) for g in parts)) if parts else "") if x)) elif unowned: v = ("Not ready until owned", "WARN gates need an owner and a date in the change list: " + ", ".join(unowned)) else: v = (ready, "every gate PASS or WARN" + ((", WARN gates owned: " + ", ".join(warns)) if warns else "")) out[kind] = {"verdict": v[0], "detail": v[1], "gates": st} return out def overrides(cfg_user): out = [] reasons = cfg_user.get("override_reasons") or {} for path, default in leaf_paths(DEFAULTS): v = get_path(cfg_user, path) if v is not None and v != default: out.append({"path": path, "default": default, "value": v, "reason": reasons.get(path) or "reason missing"}) return out # ---------- inventories and the result ---------- def writer_inventory(c): rows = [] for f in c.critical: w = c.writers(f["object"], f["field"]) h = hist_for(c, f) fill = f.get("fill") if isinstance(f.get("fill"), dict) and ratio(f.get("fill")) else None rows.append({"object": f["object"], "field": f["field"], "class": f.get("class"), "permitted": [c.label(p) for p in f.get("permitted_writers") or []], "observed": {k: sorted(v) for k, v in sorted(w.items())}, "precedence_rule": f.get("precedence_rule") or "", "flip_flop_records": (h or {}).get("flip_flop_records"), "fill": {k: fill.get(k) for k in ("numerator", "denominator", "population")} if fill else None}) return rows def integration_inventory(c): keys = ["name", "type", "identity", "identity_type", "identity_privilege", "owner", "direction", "objects", "match_key", "match_keys", "creates_records", "fields_written", "blank_semantics", "in_stack_md"] return [{k: i.get(k) for k in keys} for i in c.integrations or []] def ordered(results, scope): rids = [rid for rid in RULE if scope != "gates" or RULE[rid]["gate"]] rank = {"FAIL": 0, "WARN": 1, "PASS": 2, "NOT CHECKED": 3, "N/A": 4} return sorted(rids, key=lambda rid: (rank[results[rid].status()], list(RULE).index(rid))) def summary(results, scope): rids = [rid for rid in RULE if scope != "gates" or RULE[rid]["gate"]] cnt = {k: sum(1 for rid in rids if results[rid].status() == k) for k in ("FAIL", "WARN", "PASS", "NOT CHECKED", "N/A")} return cnt, "%d FAIL, %d WARN, %d PASS, %d NOT CHECKED, %d N/A (%d rules). Nothing was changed." % ( cnt["FAIL"], cnt["WARN"], cnt["PASS"], cnt["NOT CHECKED"], cnt["N/A"], len(rids)) def result(c, results, cfg, cfg_user, warns, scope, inputs): """Everything the reports need, as JSON-ready data (printed by --json, rendered by write_report.py).""" domains, overall = score(results, cfg) cnt, line = summary(results, scope) return { "meta": c.meta, "as_of": c.as_of.isoformat(), "scope": scope, "inputs": inputs, "rules": [{"id": rid, "name": RULE[rid]["name"], "domain": RULE[rid]["domain"], "class": RULE[rid]["class"], "class_used": results[rid].cls or RULE[rid]["class"], "gate": RULE[rid]["gate"], "status": results[rid].status(), "findings": [{"status": s, "text": t} for s, t in results[rid].items], "not_checked": results[rid].nc_reasons, "na": results[rid].na, "notes": results[rid].notes, "partial": results[rid].partial} for rid in ordered(results, scope)], "domains": domains if scope != "gates" else [], "overall": overall if scope != "gates" else None, "verdicts": verdicts(results, c.inv), "overrides": overrides(cfg_user), "warnings": warns, "writer_inventory": writer_inventory(c), "integration_inventory": integration_inventory(c), "counts": c.counts, "evidence": [e for e in c.inv.get("evidence") or [] if isinstance(e, dict)], "unclassified_actions": [{"automation": n, "action_types": t, "labels": [action_label(x) for x in t]} for n, t in c.unclassified()], "summary": cnt, "summary_line": line, } # ---------- output ---------- def finding_lines(rule): st, rid, name = rule["status"], rule["id"], rule["name"] head = "%s %s %s" % (st, rid, name) if st == "N/A": return [head + ": " + rule["na"]] if st == "NOT CHECKED": return [head + ": " + ("; ".join(rule["not_checked"]) or "no evidence")] items = sorted(rule["findings"], key=lambda x: -STATUS_ORDER.get(x["status"], 0)) lines = ["%s %s %s: %s" % (x["status"], rid, name, x["text"]) for x in items] extra = ["not checked: " + x for x in rule["not_checked"]] + rule["notes"] if extra: lines[-1] += " [" + "; ".join(extra) + "]" return lines def text_report(res): m, out = res["meta"], [] out.append("CRM audit: %s, %s (%s), read %s, scope %s, as of %s" % ( m.get("crm", "?"), m.get("instance_label", "?"), m.get("environment", "?"), m.get("read_at", "?"), res["scope"], res["as_of"])) out.append("Inputs: " + "; ".join(res["inputs"])) out.append("") for rule in res["rules"]: out.extend(finding_lines(rule)) out.append("") if res["scope"] != "gates": out.append("Scorecard (the score ranks where to look; the verdicts decide readiness)") for d in res["domains"]: if d.get("show"): out.append("- %s %s: %d %s, coverage %.0f%%%s%s" % (d["code"], d["name"], d["score"], d["band"], 100 * d["coverage"], " (partial)" if d.get("partial") else "", ", capped by a class A FAIL" if d.get("capped") else "")) else: out.append("- %s %s: insufficient evidence, coverage %.0f%%" % (d["code"], d["name"], 100 * d["coverage"])) shown = sum(1 for d in res["domains"] if d.get("show")) out.append("- Overall: %s" % (("%d (mean of %d domain scores)" % (res["overall"], shown)) if res["overall"] is not None else "not computed (fewer than six domains have a score)")) out.append("") out.append("Verdicts") for k, lab in (("automation", "Automation readiness"), ("reporting", "Reporting readiness")): out.append("- %s: %s (%s)" % (lab, res["verdicts"][k]["verdict"], res["verdicts"][k]["detail"])) out.append("") ovr = res["overrides"] out.append("Overrides: " + ("none" if not ovr else "; ".join("%s %s -> %s (%s)" % (o["path"], o["default"], o["value"], o["reason"]) for o in ovr))) for w in res["warnings"]: out.append("warning: " + w) out.append(res["summary_line"]) return "\n".join(out) def load(path, what): try: with open(path, encoding="utf-8") as f: return json.load(f) except (OSError, ValueError) as e: raise BadInput("cannot read %s %s: %s" % (what, path, e)) def main(): p = argparse.ArgumentParser(description="Read-only CRM audit rule engine and scoring.") p.add_argument("inventory") p.add_argument("--fields") p.add_argument("--automations") p.add_argument("--history", action="append", help="writer_history.py JSON (repeatable)") p.add_argument("--config") p.add_argument("--as-of") p.add_argument("--scope", choices=["full", "gates"], default=None) p.add_argument("--json", action="store_true", help="print the JSON result instead of the text report") p.add_argument("--md", help="write the Markdown report (write_report.py) that follows assets/audit-report-template.md") a = p.parse_args() try: inv_raw = load(a.inventory, "inventory") if not isinstance(inv_raw, dict): raise BadInput("the inventory must be a JSON object") problems, warns = validate_inventory(inv_raw) if problems: raise BadInput("the inventory needs fixing before the rules can run:\n " + "\n ".join(problems)) cfg_user = load(a.config, "config") if a.config else {} cfg = deep_merge(deep_merge(DEFAULTS, DEFAULT_LISTS), cfg_user) as_of = parse_date(a.as_of) if a.as_of else dt.datetime.now(dt.timezone.utc).date() if a.as_of and as_of is None: raise BadInput("--as-of must be YYYY-MM-DD, got %r" % a.as_of) inputs = ["inventory " + os.path.basename(a.inventory)] ff = af = hf = None if a.fields: ff = load(a.fields, "fields").get("fields") inputs.append("fields %s (%d)" % (os.path.basename(a.fields), len(ff or []))) if a.automations: af = load(a.automations, "automations").get("automations") inputs.append("automations %s (%d)" % (os.path.basename(a.automations), len(af or []))) for h in a.history or []: part = load(h, "history").get("fields") or [] hf = (hf or []) + part inputs.append("history %s (%d fields)" % (os.path.basename(h), len(part))) if a.config: inputs.append("config " + os.path.basename(a.config)) inv = build_inventory(inv_raw, ff, af, hf) scope = a.scope or str((inv.get("meta") or {}).get("scope") or "full") c, results, rule_warns = evaluate(inv, cfg, as_of) res = result(c, results, cfg, cfg_user, warns + rule_warns, scope, inputs) report = load_sibling("write_report").render(res) if a.md else None except BadInput as e: print("audit_crm: %s" % e, file=sys.stderr) sys.exit(2) except (AttributeError, KeyError, TypeError, ValueError) as e: print("audit_crm: the inputs could not be read (%s: %s); check them against references/inventory-format.md" % (type(e).__name__, e), file=sys.stderr) sys.exit(2) print(json.dumps(res, indent=1) if a.json else text_report(res)) if report is not None: os.makedirs(os.path.dirname(os.path.abspath(a.md)), exist_ok=True) with open(a.md, "w", encoding="utf-8") as f: f.write(report) sys.exit(1 if res["summary"]["FAIL"] else 0) if __name__ == "__main__": main()