"""Rules OBJ-01 to OBJ-06 and FLD-01 to FLD-08, in the order of references/rules-data-model.md. Each rule is one function named after its ID (OBJ-01 is obj_01). It takes the merged inventory (Ctx) and an empty result (Res) from audit_common.py and records findings, NOT CHECKED reasons and notes on the result. audit_crm.py loads this file by path; it is not run on its own. """ import re from audit_common import ( CRITICAL_WRITER_CLASSES, num, alnum, norm_writer, is_automated, pct, fmt_num, ratio, tokens, empty, norm_name, hist_for, unknown_writes_note, permitted_ok, check_ratio) WEAK_PARENT_KEYS = {"name", "companyname", "company", "domain", "website", "email", "companydomainname", "accountname"} RECORD_ID_KEYS = {"hsobjectid", "recordid", "id", "objectid"} PARENT_OBJECTS = {"companies", "accounts", "company", "account"} DATE_TOKENS = {"date", "at", "timestamp", "deadline", "day"} NUMBER_TOKENS = {"amount", "count", "revenue", "arr", "mrr", "price", "seats", "quantity", "qty", "total", "employees"} def obj_01(c, r): if not isinstance(c.objects, list): return r.nc("objects missing from the inventory") for o in c.objects: if o.get("meaning_mismatch"): r.add("FAIL", "object %s holds records of another kind: %s" % (o.get("name"), o["meaning_mismatch"])) for p in c.pipelines or []: if p.get("meaning_mismatch"): r.add("FAIL", "pipeline %r holds records of another kind: %s" % (p.get("label"), p["meaning_mismatch"])) em = c.decisions.get("entity_map_documented") if em is False: r.add("WARN", "no written entity map") elif em is None: r.notes.append("decisions.entity_map_documented not recorded") for o in c.objects: if not o.get("custom"): continue if empty(o.get("owner")): r.add("WARN", "custom object %s has no named person acting on its records" % o.get("name")) age = c.days_since(o.get("created_at")) if num(o.get("record_count")) == 0 and age is not None and age > 90: r.add("WARN", "custom object %s has 0 records and was created %d days ago" % (o.get("name"), age)) if not o.get("unique_keys"): r.add("WARN", "custom object %s has no unique key" % o.get("name")) if not r.items: r.add("PASS", "%s%d objects, no zero-record, ownerless or keyless custom objects" % ("entity map documented; " if em else "", len(c.objects))) def obj_02(c, r): if c.integrations is None: return r.nc("integrations missing from the inventory") objs = {str(o.get("name") or "").lower(): o for o in c.objects or []} writing = [i for i in c.integrations if c.writes_int(i)] if not writing: return r.add("PASS", "no integration creates or updates records") good = [] for i in writing: name = i.get("name") targets = [] for o in list(i.get("creates_records") or []) + list(i.get("objects") or []): if str(o).lower() not in targets: targets.append(str(o).lower()) creates = set(str(o).lower() for o in i.get("creates_records") or []) before = len(r.items) keys_used = [] for o in targets: mk = str((i.get("match_keys") or {}).get(o) or i.get("match_key") or "").strip() keys_used.append("%s on %s" % (o, mk or "?")) ob = objs.get(o) ukeys = set(alnum(k) for k in (ob or {}).get("unique_keys") or []) parent = o in PARENT_OBJECTS or bool((ob or {}).get("custom")) verb = "creates" if o in creates else "updates" if not mk: r.add("WARN", "%r %s %s and no match key is recorded; create-or-update not confirmed" % (name, verb, o)) elif parent and alnum(mk) in WEAK_PARENT_KEYS: r.add("FAIL", "%r %s %s matched on %s; name, domain or email alone is not a unique key for companies or accounts" % (name, verb, o, mk)) elif ob is not None and not ukeys: r.add("FAIL", "%r writes %s, which has no unique key" % (name, o)) elif ob is not None and alnum(mk) not in ukeys | RECORD_ID_KEYS: r.add("FAIL", "%r matches %s on %s, which is not marked unique (unique keys: %s)" % (name, o, mk, ", ".join(ob.get("unique_keys") or []))) else: if o in creates and i.get("searches_before_create") is False: r.add("WARN", "%r creates %s without searching %s first" % (name, o, mk)) if o in ("contacts", "people", "leads") and alnum(mk) == "email" and not i.get("keeps_person_id"): r.add("WARN", "%r matches %s on email only and keeps no person ID" % (name, o)) if len(r.items) == before: good.append("%s (%s)" % (name, ", ".join(keys_used))) if good: r.add("PASS", "match on unique keys: " + ", ".join(good)) def obj_03(c, r): s = c.settings rated = False if c.crm == "hubspot" or "auto_create_associate_companies" in s: v = s.get("auto_create_associate_companies") if v is None: r.nc("HubSpot 'Create and associate companies with contacts' setting not captured") else: rated = True comp = next((o for o in c.objects or [] if str(o.get("name")).lower() == "companies"), {}) keys = [k for k in comp.get("unique_keys") or [] if alnum(k) not in WEAK_PARENT_KEYS | RECORD_ID_KEYS] if v and (keys or s.get("companies_keyed_on_source_id") or s.get("salesforce_connected")): why = ("companies are keyed on " + ", ".join(keys)) if keys else "Salesforce is connected" r.add("FAIL", "'Create and associate companies with contacts' is on while %s" % why) elif v and s.get("auto_association_shared_domain_exclusions") is False: r.add("WARN", "'Create and associate companies with contacts' is on with no exclusions for shared domains") else: r.add("PASS", "auto company association is %s" % ("on, with shared-domain exclusions" if v else "off")) if c.crm == "salesforce" or "account_creation_without_duplicate_matching" in s: v = s.get("account_creation_without_duplicate_matching") if v is not None: rated = True if v: r.add("FAIL", "a flow or lead conversion path creates accounts without duplicate matching while an account key exists") else: r.add("PASS", "account creation paths run duplicate matching") if c.crm == "attio" or "mailbox_sync_creates_companies" in s: v = s.get("mailbox_sync_creates_companies") if v is not None: rated = True if v and s.get("companies_maintained_externally"): r.add("WARN", "mailbox sync creates a company for every non-public email domain while companies are maintained from another system") else: r.add("PASS", "mailbox sync company creation fits how companies are maintained") if not rated and not r.nc_reasons: r.nc("no auto-creation or auto-association settings captured") if rated and c.decisions.get("auto_creation_recorded") is not True: r.notes.append("record the auto-creation settings in decisions.md") def obj_04(c, r): bm = str(c.meta.get("business_model") or "").lower() if bm in ("b2c", "self-serve", "self_serve"): r.na = "business model is %s, so deals may legitimately have no company" % bm return t = c.cfg["OBJ-04"] a = check_ratio(r, c.counts.get("open_deals_without_company"), "open deals without a company", t["open_deals_without_company"]["warn"], t["open_deals_without_company"]["fail"]) b = check_ratio(r, c.counts.get("sales_contacts_without_company"), "sales-eligible contacts without a company", t["sales_contacts_without_company"]["warn"], t["sales_contacts_without_company"]["fail"]) cc = None if c.settings.get("contact_roles_drive_lifecycle"): cc = check_ratio(r, c.counts.get("open_deals_without_contact"), "open deals without a contact", t["open_deals_without_contact_warn"]) if a is None and b is None and cc is None and not r.nc_reasons: r.nc("counts.open_deals_without_company and counts.sales_contacts_without_company missing") def obj_05(c, r): t = c.cfg["OBJ-05"] a = check_ratio(r, c.counts.get("companies_sharing_domain"), "companies sharing a primary domain", t["shared_domain_companies"]["warn"], t["shared_domain_companies"]["fail"]) rising = [i.get("name") for i in c.integrations or [] if str(i.get("duplicate_trend") or "").lower() == "rising"] for n in rising: r.add("FAIL", "%r created duplicates in the last 30 days at a rising rate" % n) v = ratio(c.counts.get("vendor_duplicate_contacts")) if v and v[1] > 0 and v[0] / v[1] > t["vendor_duplicate_contacts_warn"]: r.add("WARN", "vendor duplicate suggestions on %s of contacts (signal only)" % pct(*v)) if a is None and not rising and not r.items and not r.nc_reasons: r.nc("counts.companies_sharing_domain missing") def obj_06(c, r): if c.integrations is None: return r.nc("integrations missing from the inventory") merges = num(c.settings.get("merges_90d")) behaviour = {"hubspot": "HubSpot merges create a new record ID unless the primary ID preservation beta is on", "attio": "Attio merges always create a new record ID and the originals can no longer be read or written", "salesforce": "Salesforce keeps the master and deletes the merged records"}.get(c.crm, "check the CRM's merge behaviour") if c.crm == "hubspot" and c.settings.get("merge_preserves_record_id") is not None: behaviour += " (preservation is %s here)" % ("on" if c.settings.get("merge_preserves_record_id") else "off") keyed = [i for i in c.integrations if i.get("keys_on_record_id")] for i in keyed: errs = i.get("errors_after_merges") or (num(i.get("record_not_found_errors_30d")) or 0) > 0 if merges and merges > 0 and errs: r.add("FAIL", "%r keys on CRM record IDs; %d merges in 90 days were followed by sync errors or orphaned mappings" % (i.get("name"), merges)) elif not i.get("remaps_after_merge"): r.add("WARN", "%r keys on CRM record IDs with no remap step after merges (%s merges in 90 days); %s" % ( i.get("name"), fmt_num(merges) if merges is not None else "unknown", behaviour)) if not r.items: r.add("PASS", "no integration keys on CRM record IDs without a remap step") REGISTER_KEYS = ["meaning", "authoritative_source", "permitted_writers", "blank_means", "consumers"] def fld_01(c, r): if not c.critical: return r.nc("no critical fields defined (audit-config.json critical_fields)") gaps = [] for f in c.critical: miss = [k for k in REGISTER_KEYS if empty(f.get(k))] if miss: gaps.append("%s.%s (%s)" % (f["object"], f["field"], "no row" if len(miss) == len(REGISTER_KEYS) else "no " + ", ".join(miss))) if gaps: r.add("WARN", "%d of %d critical fields are incomplete in the register: %s" % (len(gaps), len(c.critical), "; ".join(gaps))) else: r.add("PASS", "all %d critical fields are complete in the register" % len(c.critical)) def flip_state(c, h): """FAIL at 3+ flip-flop records, or when flip-flops between automated writers reach 1% of records changed; WARN at 1 or 2 otherwise (an automated writer reversing a person's change stays WARN).""" t = c.cfg["FLD-02"] n = int(num(h.get("flip_flop_records")) or 0) na, nh = num(h.get("automated_flip_flop_records")), num(h.get("human_reversal_records")) if na is None and nh is None: # history files without the split: count every flip-flop as automated na, nh = n, 0 na, nh = int(na or 0), int(nh or 0) n = max(n, na + nh) changed = num(h.get("records_changed")) rate = (n / changed) if changed else num(h.get("flip_flop_rate")) auto_rate = (na / changed) if changed else (rate if not nh else None) if n >= t["flip_flop_records_fail"] or (na >= 1 and auto_rate is not None and auto_rate >= t["flip_flop_rate_fail"]): st = "FAIL" elif n >= t["flip_flop_records_warn"]: st = "WARN" else: st = "PASS" kinds = [k for k in ("%d between automated writers" % na if na else "", "%d an automated writer reversing a person's change" % nh if nh else "") if k] return st, n, rate, ", ".join(kinds) def history_gap(c, f, w, h, what): """Why a critical field cannot pass without history, or None when its history was read.""" fid = "%s.%s" % (f["object"], f["field"]) if f.get("history_tracked") is False and not h: return "%s: history tracking is off for this field, so no history exists; the first change request turns it on" % fid if not h: return "%s: no history, %s; export its %s" % (fid, ("%d writer%s from definitions only" % (len(w), "" if len(w) == 1 else "s")) if w else "no writer evidence", what) if num(h.get("records_changed")) == 0: return "%s: the history has no changes inside the %s-day window; widen the window" % ( fid, h.get("window_days", c.cfg["FLD-02"]["window_days"])) return None def pass_only_when_complete(r, gaps): """A rule passes only when every field it covers was rated from history (NOT CHECKED is never a pass). Each gap is a NOT CHECKED part (Res.gap): a PASS turns NOT CHECKED, FAIL and WARN findings stand, and a gate with a gap cannot reach a ready verdict whatever its status.""" for g in gaps: r.gap(g, "rated from history: ") def fld_02(c, r): targets = [f for f in c.critical if str(f.get("class") or "").lower() in CRITICAL_WRITER_CLASSES] if not targets: return r.nc("no critical fields of class routing, ownership, lifecycle, stage, revenue or identity") unknown_writes_note(c, r) gaps, passes = [], [] for f in targets: fid = "%s.%s" % (f["object"], f["field"]) w = c.writers(f["object"], f["field"]) h = hist_for(c, f) auto = [l for l in w if is_automated(l)] before = len(r.items) if len(auto) >= 2 and empty(f.get("precedence_rule")): r.add("FAIL", "%s: %d automated writers (%s) and no written precedence rule" % (fid, len(auto), ", ".join(sorted(auto)))) gap = history_gap(c, f, w, h, "property or field history") if h and not gap: st, n, rate, kinds = flip_state(c, h) if st != "PASS": r.add(st, "%s: flip-flops on %d of %s records changed in %s days (%s)%s; samples %s" % ( fid, n, h.get("records_changed", "?"), h.get("window_days", c.cfg["FLD-02"]["window_days"]), ("%.1f%%" % (100 * rate)) if rate is not None else "rate unknown", (": " + kinds) if kinds else "", ", ".join(str(x) for x in (h.get("sample_record_ids") or [])[:5]) or "none listed")) permitted = set(c.label(p) for p in f.get("permitted_writers") or []) if permitted: outside = [l for l in sorted(w) if not permitted_ok(l, permitted)] if outside: r.add("WARN", "%s: observed writers outside the permitted list: %s" % (fid, ", ".join(outside))) unknown = [l for l in w if l.split(":", 1)[0] in ("id", "unknown", "api-token", "app") or l.partition(":")[2].isdigit()] if unknown: r.notes.append("%s: writers known only by raw ID: %s; name them with --writer-map or give the automation or " "integration its id in the inventory (unmapped user IDs count as automated)" % (fid, ", ".join(sorted(unknown)))) if gap: gaps.append(gap) elif len(r.items) == before: passes.append("%s: writers %s, no flip-flops in %s records changed" % ( fid, ", ".join(sorted(w)) or "none observed", h.get("records_changed", "?"))) for p in passes: r.add("PASS", p) pass_only_when_complete(r, gaps) def fld_03(c, r): targets = [f for f in c.critical if str(f.get("class") or "").lower() == "consent"] if not targets: return r.nc("no consent fields in critical_fields") unknown_writes_note(c, r) gaps = [] for f in targets: # every consent field needs history, whatever its definitions show fid = "%s.%s" % (f["object"], f["field"]) w = c.writers(f["object"], f["field"]) gap = history_gap(c, f, w, hist_for(c, f), "subscription or consent property history") if gap: gaps.append(gap) permitted = set(c.label(p) for p in f.get("permitted_writers") or []) if not permitted: r.add("WARN", "%s has no register entry naming permitted writers; writers seen: %s" % (fid, ", ".join(sorted(w)) or "none")) continue bad_auto = [l for l in sorted(w) if is_automated(l) and not permitted_ok(l, permitted)] bad_human = [l for l in sorted(w) if not is_automated(l) and not permitted_ok(l, permitted)] if bad_auto: r.add("FAIL", "%s is written by %s, not a permitted consent writer (permitted: %s)" % (fid, ", ".join(bad_auto), ", ".join(sorted(permitted)))) if bad_human: r.add("WARN", "%s has manual edits by %s outside the permitted list; confirm the documented process" % (fid, ", ".join(bad_human))) for rev in f.get("opt_out_reversals") or []: who = rev.get("writer") if isinstance(rev, dict) else rev n = rev.get("records") if isinstance(rev, dict) else None r.add("FAIL", "%s: %s cleared opt-outs%s in the window" % (fid, norm_writer(who), (" on %s records" % n) if n else "")) if not bad_auto and not bad_human and not f.get("opt_out_reversals") and not gap: r.add("PASS", "%s: only permitted writers in history and definitions (%s)" % (fid, ", ".join(sorted(w)) or "none observed")) if c.settings.get("forms_set_marketing_contact_default") is True and c.decisions.get("marketing_contact_default_reviewed") is not True: r.add("WARN", "forms set new contacts as marketing contacts by default with no reviewed decision") pass_only_when_complete(r, gaps) def field_written(c, f): name = str(f.get("name") or "").lower() if (num(f.get("fill_rate")) or 0) > 0 or f.get("writers"): return True for a in c.enabled(): if any(str(w.get("field") or "").lower() == name for w in (a.get("writes") or []) + (a.get("owner_writes") or [])): return True return any(n == name for i in c.integrations or [] for _, n in c.int_fields(i)) def fld_04(c, r): if c.fields is None: return r.nc("fields missing (run normalize_exports.py and pass --fields)") groups = {} for f in c.fields: fam = "string" if f.get("type_family") in ("text", "enum") else (f.get("type_family") or "other") groups.setdefault((str(f.get("object") or "").lower(), fam), []).append(f) clusters = [] for key, fs in groups.items(): keys = [norm_name(c, f.get("name")) for f in fs] parent = list(range(len(fs))) def find(x): while parent[x] != x: parent[x] = parent[parent[x]] x = parent[x] return x for i in range(len(fs)): for j in range(i + 1, len(fs)): (ci, ti), (cj, tj) = keys[i], keys[j] if not ci or not cj: continue jac = len(ti & tj) / float(len(ti | tj)) if ti | tj else 0 if ci == cj or jac >= 0.8: parent[find(i)] = find(j) comp = {} for i in range(len(fs)): comp.setdefault(find(i), []).append(fs[i]) clusters.extend([m for m in comp.values() if len(m) >= 2 and any(x.get("custom") for x in m)]) for m in clusters: names = [x.get("name") for x in m] obj = m[0].get("object") masked = set(re.sub(r"\d+", "#", str(n).lower()) for n in names) both = [x for x in m if field_written(c, x) and (num(x.get("usages")) or 0) > 0] desc = ", ".join("%s (usages %s, fill %s)" % (x.get("name"), x.get("usages") if x.get("usages") is not None else "?", ("%.0f%%" % (100 * x["fill_rate"])) if num(x.get("fill_rate")) is not None else "?") for x in m) if len(masked) == 1 and any(re.search(r"\d", str(n)) for n in names): r.add("WARN", "%s: numbered set %s; model the repeated entity as its own object or a list" % (obj, ", ".join(names))) elif len(both) >= 2: r.add("FAIL", "%s: %s split one meaning and %d of them are written and used: %s" % (obj, ", ".join(names), len(both), desc)) else: kind = "custom field duplicates a standard one" if any(x.get("custom") is False for x in m) else "near-duplicate fields" r.add("WARN", "%s: %s: %s" % (obj, kind, desc)) if not r.items: r.add("PASS", "no near-duplicate clusters among %d fields" % len(c.fields)) def fld_05(c, r): if c.fields is None: return r.nc("fields missing (run normalize_exports.py and pass --fields)") t = c.cfg["FLD-05"] custom = [f for f in c.fields if f.get("custom")] if not custom: return r.add("PASS", "no custom fields") has_data = any(f.get("fill_rate") is not None and f.get("usages") is not None for f in custom) if not has_data: r.nc("no custom field has both fill rate and usages") mapped = set(n for i in c.integrations or [] for _, n in c.int_fields(i)) cands, unknown_age, kept = [], 0, [] for f in custom if has_data else []: if num(f.get("fill_rate")) != 0 or num(f.get("usages")) != 0: continue age, basis = c.days_since(f.get("created_at")), "created" if age is None: age, basis = c.days_since(f.get("last_updated")), "last updated" if age is None: unknown_age += 1 continue if age <= t["unused_min_age_days"]: continue if str(f.get("name")).lower() in mapped: kept.append(f.get("name")) continue cands.append("%s.%s (%s %d days ago)" % (f.get("object"), f.get("name"), basis, age)) if cands: r.add("WARN", "archive candidates (0%% fill, zero usages, older than %d days): %s" % (t["unused_min_age_days"], "; ".join(cands))) if kept: r.notes.append("empty but mapped by an integration, keep: " + ", ".join(kept)) if unknown_age: r.notes.append("%d unused custom fields have no created or last updated date" % unknown_age) limits = c.settings.get("custom_field_limits") or {} counts = c.settings.get("custom_field_counts") or {} for obj, limit in limits.items(): lim = num(limit) n = num(counts.get(obj)) if obj in counts else float(sum(1 for f in custom if str(f.get("object")).lower() == obj.lower())) if not lim: continue share = n / lim if share >= t["capacity"]["fail"]: r.add("FAIL", "%s: %d custom fields of a %d limit (%s)" % (obj, n, lim, pct(n, lim))) r.cls = "B" elif share >= t["capacity"]["warn"]: r.add("WARN", "%s: %d custom fields of a %d limit (%s)" % (obj, n, lim, pct(n, lim))) if not r.items and has_data: r.add("PASS", "no unused custom fields older than %d days%s" % ( t["unused_min_age_days"], ("; capacity below %.0f%%" % (100 * t["capacity"]["warn"])) if limits else "; capacity limits not captured")) def fld_06(c, r): if c.fields is None: return r.nc("fields missing (run normalize_exports.py and pass --fields)") known = [f for f in c.fields if f.get("custom") and f.get("description") is not None] if not known: return r.nc("descriptions are not in the fields data; Salesforce describe output has none, so pass the " "FieldDefinition query results with normalize_exports.py salesforce-describe --descriptions") missing = [f for f in known if not str(f.get("description") or "").strip()] share = len(missing) / float(len(known)) if share > c.cfg["FLD-06"]["undocumented_custom_warn"]: r.add("WARN", "%d of %d custom fields (%s) have no description" % (len(missing), len(known), pct(len(missing), len(known)))) crit = [cf["field"] for cf in c.critical if (cf["object"], cf["field"].lower()) in c.field_index and c.field_index[(cf["object"], cf["field"].lower())].get("description") is not None and not str(c.field_index[(cf["object"], cf["field"].lower())].get("description") or "").strip()] if crit: r.add("WARN", "critical fields without a description: " + ", ".join(crit)) if not r.items: r.add("PASS", "%d of %d custom fields have a description" % (len(known) - len(missing), len(known))) def fld_07(c, r): if c.fields is None: return r.nc("fields missing (run normalize_exports.py and pass --fields)") for f in c.fields: fid = "%s.%s" % (f.get("object"), f.get("name")) fam = f.get("type_family") opts = f.get("options") or [] if fam == "enum" and opts: seen = {} for idx, o in enumerate(opts): for k in set([str(o.get("label") or "").strip().lower(), str(o.get("value") or "").strip().lower()]): if not k: continue if k in seen and seen[k] != idx: a = opts[seen[k]].get("raw", opts[seen[k]].get("label")) b = o.get("raw", o.get("label")) r.add("FAIL", "%s has options %r and %r that match after trimming and lowercasing" % (fid, a, b)) break seen.setdefault(k, idx) if len(opts) == 1: r.add("WARN", "%s is an enumeration with one option" % fid) if fam == "text": tk = set(tokens(f.get("name"))) kind = "date" if tk & DATE_TOKENS else ("number" if tk & NUMBER_TOKENS else None) if kind and (f.get("used_in_range_filters") or f.get("used_in_calculations")): r.add("FAIL", "%s stores a %s as text and is used in range filters or calculations" % (fid, kind)) elif kind: r.add("WARN", "%s looks like a %s but is stored as text (%s)" % ( fid, kind, ("usages %s" % f["usages"]) if f.get("usages") is not None else "usages not recorded")) if f.get("used_as_dimension"): r.add("WARN", "%s is free text used as a segmentation dimension" % fid) if not r.items: r.add("PASS", "no duplicate options or type smells among %d fields" % len(c.fields)) def fld_08(c, r): t = c.cfg["FLD-08"] rated = 0 for f in c.critical: if not isinstance(f.get("fill"), dict): continue cls = str(f.get("class") or "").lower() th, kind = (t["routing_inputs"], "routing input") if cls in ("routing", "ownership") else (t["reporting_dimensions"], "reporting dimension") if check_ratio(r, f["fill"], "%s.%s filled (%s)" % (f["object"], f["field"], kind), th["warn"], th["fail"], below=True) == "rated": rated += 1 if not rated and not r.nc_reasons: r.nc("no fill counts for critical fields (critical_fields[].fill)")