|
#!/usr/bin/env python3 |
|
# flake8: noqa: E501 |
|
# (file-wide E501 disable: long docstring/section strings stay <100 per CLAUDE.md threshold.) |
|
"""Promotion-status toil dashboard for zentral_config. |
|
|
|
SANITIZED PUBLIC COPY — track/audience tag names are genericized (testing/prod) and internal |
|
resource names removed. The tag labels here (e.g. `testing_track`) do NOT match the live repo's |
|
real tags, so this copy is for sharing/illustration, not for running against production terraform; |
|
see toil.py for the operational version. |
|
|
|
Sibling in spirit to a companion msc_repo toil.py: that one watches the munki/S3 side for |
|
pkgs stuck pre-prod; this one watches the *Zentral* side — terraform that gates an artifact to a |
|
subset of the fleet and never opens it up. The classic trap is a resource left at default_shard 0 |
|
with only testing track tag_shards: it reaches testing machines forever but no untagged ("prod"/ |
|
default) machine ever gets it. |
|
|
|
This file is the thin entrypoint/orchestrator (mirrors how msc_repo's toil.py stays an entrypoint). |
|
For now it's self-contained — as sections deepen, split the parse/classify/render weight into a |
|
toil_lib/ package the way msc_repo did. |
|
|
|
Shard model (provider default for default_shard AND shard_modulo is 100, verified against the |
|
terraform-provider-zentral docs). Availability to an *untagged* prod machine = default_shard/shard_modulo: |
|
omitted / >= modulo -> 100% = prod (fully promoted) |
|
0 with track tags -> 0% of prod, only the testing pipeline tags (THE hitlist case) |
|
0 with audience tags -> targeted population (specific audience tags) — usually intentional, verify |
|
0 with no tags -> reaches nobody (effectively disabled) |
|
1..modulo-1 -> partial/mid-rollout (e.g. an artifact at 7/10 = 70%) |
|
|
|
Sections written to TOIL.md (overwritten each run; git is the history for now): |
|
Hit list — track-gated-never-prod + partial rollouts + disabled, the resourceable promotions |
|
A. Munki — zentral_monolith_sub_manifest_pkg_info shard gaps |
|
B. MDM — newest version per artifact + blueprint_artifact assignment shard gaps |
|
C. Santa — inventory plus Default-vs-Lockdown+ rule drift |
|
D. Osquery— cursory inventory (queries/packs/ATC/enrollments) [refine later] |
|
E. Audience-gated — non-fleet targeting parked out of the hit list (verify intent, not toil) |
|
|
|
Control flow (main()): |
|
parse_tf_blocks -> classify_munki -> classify_mdm -> summarize_santa -> summarize_osquery -> |
|
partition findings into hit list vs audience-gated -> write_toil |
|
|
|
Map / non-success exit codes (ascending in control-flow order): |
|
1: no *.tf found in REPO_ROOT, unreadable/undecodable *.tf file, or TOIL.md write failure |
|
""" |
|
|
|
import datetime |
|
import re |
|
import sys |
|
from pathlib import Path |
|
from typing import Any |
|
|
|
REPO_ROOT = Path(__file__).resolve().parent |
|
TOIL_PATH = REPO_ROOT / "TOIL.md" |
|
|
|
# Tags that ARE the promotion pipeline; gating to only these = "stuck on the way to prod". |
|
TRACK_TAGS: set[str] = {"testing_track", "temp"} |
|
|
|
MUNKI_TYPE = "zentral_monolith_sub_manifest_pkg_info" |
|
MDM_ARTIFACT_TYPE = "zentral_mdm_artifact" |
|
MDM_VERSION_TYPES = { |
|
"zentral_mdm_profile": "Profile", |
|
"zentral_mdm_enterprise_app": "Enterprise App", |
|
"zentral_mdm_store_app": "Store App", |
|
"zentral_mdm_declaration": "Declaration", |
|
} |
|
MDM_BLUEPRINT_TYPE = "zentral_mdm_blueprint_artifact" |
|
SANTA_DEFAULT_CONFIG = "default" |
|
SANTA_LOCKDOWN_CONFIG = "lockdown_plus" |
|
|
|
HEADER_RE = re.compile(r'^resource\s+"([a-z0-9_]+)"\s+"([^"]+)"\s*\{') |
|
SCALAR_RE = re.compile(r"^\s*([a-z0-9_]+)\s*=\s*(.+?)\s*$") |
|
# `.id` optional: one source stanza writes `zentral_tag.testing_track` (no .id), capture it anyway. |
|
TAG_REF_RE = re.compile(r"zentral_tag\.([A-Za-z0-9_-]+)(?:\.id)?") |
|
ARTIFACT_REF_RE = re.compile(r"zentral_mdm_artifact\.([A-Za-z0-9_-]+)\.id") |
|
TAG_SHARDS_RE = re.compile(r"^\s*tag_shards\s*=") |
|
|
|
|
|
class ToilError(Exception): |
|
"""Expected fatal error that should be reported without a traceback.""" |
|
|
|
|
|
def main() -> int: |
|
"""Parse the repo's terraform, classify shard-gated artifacts, write TOIL.md.""" |
|
now = datetime.datetime.now(datetime.timezone.utc) |
|
print(f"[{now.strftime('%Y-%m-%dT%H:%M:%SZ')}] toil.py start ({REPO_ROOT})") |
|
try: |
|
blocks = parse_tf_blocks(REPO_ROOT) |
|
except ToilError as err: |
|
print(f"FATAL: {err}", file=sys.stderr) |
|
return 1 |
|
if not blocks: |
|
print(f"FATAL: no *.tf resources found under {REPO_ROOT}") |
|
return 1 |
|
print(f" parsed {len(blocks)} resource blocks") |
|
munki_findings = classify_munki(blocks) |
|
mdm_findings = classify_mdm(blocks) |
|
santa = summarize_santa(blocks) |
|
osquery = summarize_osquery(blocks) |
|
findings = munki_findings + mdm_findings |
|
# Hit list = the promotion toil; audience-gated is intentional targeting, parked separately. |
|
hit_list = [item for item in findings if item["severity"] != "audience_gated"] |
|
audience = [item for item in findings if item["severity"] == "audience_gated"] |
|
hit_list.sort( |
|
key=lambda item: (SEVERITY_ORDER.get(item["severity"], 9), item["name"]) |
|
) |
|
print( |
|
f" munki gaps: {len(munki_findings)} | mdm gaps: {len(mdm_findings)} | " |
|
f"hit list: {len(hit_list)} | audience-gated: {len(audience)}" |
|
) |
|
try: |
|
write_toil( |
|
TOIL_PATH, |
|
now, |
|
hit_list, |
|
munki_findings, |
|
mdm_findings, |
|
santa, |
|
osquery, |
|
audience, |
|
) |
|
except ToilError as err: |
|
print(f"FATAL: {err}", file=sys.stderr) |
|
return 1 |
|
end = datetime.datetime.now(datetime.timezone.utc) |
|
print( |
|
f"[{end.strftime('%Y-%m-%dT%H:%M:%SZ')}] done in {(end - now).total_seconds():.1f}s, " |
|
f"wrote {TOIL_PATH.name}" |
|
) |
|
return 0 |
|
|
|
|
|
def parse_tf_blocks(tf_dir: Path) -> list[dict[str, Any]]: |
|
"""Block-parse every top-level `resource` in *.tf. Relies on terraform fmt: a resource opens |
|
with `resource "type" "label" {` and closes with `}`. Returns one dict per block |
|
with its scalar assignments, whether it has a tag_shards array, and the tag labels inside it.""" |
|
blocks: list[dict[str, Any]] = [] |
|
for tf_path in sorted(tf_dir.glob("*.tf")): |
|
try: |
|
lines = tf_path.read_text(encoding="utf-8").splitlines() |
|
except UnicodeDecodeError as err: |
|
raise ToilError( |
|
f"could not decode Terraform file {tf_path}: {err}" |
|
) from err |
|
except OSError as err: |
|
raise ToilError(f"could not read Terraform file {tf_path}: {err}") from err |
|
index, total = 0, len(lines) |
|
while index < total: |
|
header = HEADER_RE.match(lines[index]) |
|
if not header: |
|
index += 1 |
|
continue |
|
body: list[str] = [] |
|
index += 1 |
|
while index < total and lines[index].strip() != "}": |
|
body.append(lines[index]) |
|
index += 1 |
|
index += 1 # step past the column-0 closing brace |
|
blocks.append( |
|
_block_from_body(header.group(1), header.group(2), tf_path.name, body) |
|
) |
|
return blocks |
|
|
|
|
|
def _block_from_body( |
|
res_type: str, label: str, file_name: str, body: list[str] |
|
) -> dict[str, Any]: |
|
"""Pull scalar assignments + tag_shards tag labels out of a resource body.""" |
|
scalars: dict[str, str] = {} |
|
for line in body: |
|
match = SCALAR_RE.match(line) |
|
if ( |
|
match and match.group(1) not in scalars |
|
): # first wins; nested keys reuse names |
|
scalars[match.group(1)] = match.group(2).strip().strip('"') |
|
return { |
|
"type": res_type, |
|
"label": label, |
|
"file": file_name, |
|
"scalars": scalars, |
|
"gating_tags": _tag_shards_tags(body), |
|
} |
|
|
|
|
|
def _tag_shards_tags(body: list[str]) -> list[str]: |
|
"""Tag labels referenced inside the `tag_shards = [ ... ]` array only (not excluded_tag_ids). |
|
Tracks bracket depth so it stops at the array's close.""" |
|
tags: list[str] = [] |
|
capturing, depth = False, 0 |
|
for line in body: |
|
if not capturing and TAG_SHARDS_RE.match(line): |
|
capturing = True |
|
if capturing: |
|
tags.extend(TAG_REF_RE.findall(line)) |
|
depth += line.count("[") - line.count("]") |
|
if depth <= 0 and "]" in line: |
|
break |
|
return tags |
|
|
|
|
|
def _int_scalar(scalars: dict[str, str], key: str, default: int) -> int: |
|
"""Scalar as int, or default if absent/unparseable (provider omits == its own default).""" |
|
try: |
|
return int(scalars[key]) |
|
except (KeyError, ValueError): |
|
return default |
|
|
|
|
|
# severity buckets, worst (most actionable) first |
|
SEVERITY_ORDER: dict[str, int] = {"track_gated": 0, "partial": 1, "disabled": 2} |
|
|
|
|
|
def classify_shard( |
|
default_shard: int, shard_modulo: int, gating_tags: list[str] |
|
) -> tuple[str, int]: |
|
"""Return (severity, pct-of-prod). 'prod' means fully promoted (not a finding).""" |
|
if default_shard >= shard_modulo: |
|
return ("prod", 100) |
|
pct = round(default_shard / shard_modulo * 100) if shard_modulo else 0 |
|
if default_shard == 0: |
|
if not gating_tags: |
|
return ("disabled", 0) |
|
if set(gating_tags) <= TRACK_TAGS: |
|
return ("track_gated", 0) |
|
return ("audience_gated", 0) |
|
return ("partial", pct) |
|
|
|
|
|
def _finding( |
|
domain: str, kind: str, name: str, block: dict[str, Any], pct: int, severity: str |
|
) -> dict[str, Any]: |
|
"""Uniform finding record for the renderers.""" |
|
return { |
|
"domain": domain, |
|
"kind": kind, |
|
"name": name, |
|
"pct": pct, |
|
"severity": severity, |
|
"gating": ", ".join(block["gating_tags"]) or "—", |
|
"file": block["file"], |
|
} |
|
|
|
|
|
def classify_munki(blocks: list[dict[str, Any]]) -> list[dict[str, Any]]: |
|
"""sub_manifest_pkg_info entries whose prod availability is below 100%.""" |
|
findings: list[dict[str, Any]] = [] |
|
for block in blocks: |
|
if block["type"] != MUNKI_TYPE: |
|
continue |
|
scalars = block["scalars"] |
|
severity, pct = classify_shard( |
|
_int_scalar(scalars, "default_shard", 100), |
|
_int_scalar(scalars, "shard_modulo", 100), |
|
block["gating_tags"], |
|
) |
|
if severity == "prod": |
|
continue |
|
name = scalars.get("pkg_info_name", block["label"]) |
|
findings.append(_finding("munki", "pkg_info", name, block, pct, severity)) |
|
return findings |
|
|
|
|
|
def classify_mdm(blocks: list[dict[str, Any]]) -> list[dict[str, Any]]: |
|
"""Newest version per artifact + each blueprint_artifact assignment, where availability <100%.""" |
|
names: dict[str, str] = { |
|
block["label"]: block["scalars"].get("name", block["label"]) |
|
for block in blocks |
|
if block["type"] == MDM_ARTIFACT_TYPE |
|
} |
|
# newest-version child per artifact (older versions being gated is expected, not toil) |
|
newest: dict[str, dict[str, Any]] = {} |
|
for block in blocks: |
|
kind = MDM_VERSION_TYPES.get(block["type"]) |
|
if not kind: |
|
continue |
|
ref = ARTIFACT_REF_RE.search(block["scalars"].get("artifact_id", "")) |
|
if not ref: |
|
continue |
|
artifact = ref.group(1) |
|
version = _int_scalar(block["scalars"], "version", 0) |
|
if artifact not in newest or version > newest[artifact]["_version"]: |
|
newest[artifact] = {**block, "_version": version, "_kind": kind} |
|
findings: list[dict[str, Any]] = [] |
|
for artifact, block in newest.items(): |
|
severity, pct = classify_shard( |
|
_int_scalar(block["scalars"], "default_shard", 100), |
|
_int_scalar(block["scalars"], "shard_modulo", 100), |
|
block["gating_tags"], |
|
) |
|
if severity == "prod": |
|
continue |
|
label = ( |
|
f"{names.get(artifact, artifact)} (v{block['_version']} {block['_kind']})" |
|
) |
|
findings.append(_finding("mdm", "version", label, block, pct, severity)) |
|
for ( |
|
block |
|
) in blocks: # blueprint-assignment-level gating (separate axis from the version) |
|
if block["type"] != MDM_BLUEPRINT_TYPE: |
|
continue |
|
severity, pct = classify_shard( |
|
_int_scalar(block["scalars"], "default_shard", 100), |
|
_int_scalar(block["scalars"], "shard_modulo", 100), |
|
block["gating_tags"], |
|
) |
|
if severity == "prod": |
|
continue |
|
ref = ARTIFACT_REF_RE.search(block["scalars"].get("artifact_id", "")) |
|
artifact = ref.group(1) if ref else block["label"] |
|
label = f"{names.get(artifact, artifact)} (blueprint assignment)" |
|
findings.append(_finding("mdm", "blueprint", label, block, pct, severity)) |
|
return findings |
|
|
|
|
|
def summarize_santa(blocks: list[dict[str, Any]]) -> dict[str, Any]: |
|
"""Santa inventory plus Default-vs-Lockdown+ rule drift.""" |
|
configs = [ |
|
f"{block['scalars'].get('name', block['label'])} ({block['scalars'].get('client_mode', '?')})" |
|
for block in blocks |
|
if block["type"] == "zentral_santa_configuration" |
|
] |
|
rules = [block for block in blocks if block["type"] == "zentral_santa_rule"] |
|
by_config: dict[str, int] = {} |
|
by_policy: dict[str, int] = {} |
|
for rule in rules: |
|
config_ref = rule["scalars"].get("configuration_id", "?") |
|
config = config_ref.split(".")[1] if "." in config_ref else config_ref |
|
by_config[config] = by_config.get(config, 0) + 1 |
|
policy = rule["scalars"].get("policy", "?") |
|
by_policy[policy] = by_policy.get(policy, 0) + 1 |
|
return { |
|
"configs": configs, |
|
"enrollments": sum( |
|
1 for b in blocks if b["type"] == "zentral_santa_enrollment" |
|
), |
|
"rule_total": len(rules), |
|
"by_config": by_config, |
|
"by_policy": by_policy, |
|
"drift": _santa_rule_drift(rules), |
|
} |
|
|
|
|
|
def _santa_rule_drift(rules: list[dict[str, Any]]) -> list[dict[str, str]]: |
|
"""Compare Santa rules in Default and Lockdown+ by target, policy, and CEL expression.""" |
|
default_rules = _santa_rules_by_target(rules, SANTA_DEFAULT_CONFIG) |
|
lockdown_rules = _santa_rules_by_target(rules, SANTA_LOCKDOWN_CONFIG) |
|
drift: list[dict[str, str]] = [] |
|
for target in sorted(set(default_rules) | set(lockdown_rules)): |
|
default_group = default_rules.get(target, []) |
|
lockdown_group = lockdown_rules.get(target, []) |
|
if _santa_rule_group_fingerprints( |
|
default_group |
|
) == _santa_rule_group_fingerprints(lockdown_group): |
|
continue |
|
if not default_group: |
|
discrepancy = "missing from default" |
|
elif not lockdown_group: |
|
discrepancy = "missing from lockdown" |
|
else: |
|
discrepancy = "rule differs" |
|
files = sorted({rule["file"] for rule in default_group + lockdown_group}) |
|
drift.append( |
|
{ |
|
"target_type": target[0], |
|
"target_identifier": target[1], |
|
"default": _santa_rule_group_summary(default_group), |
|
"lockdown": _santa_rule_group_summary(lockdown_group), |
|
"discrepancy": discrepancy, |
|
"file": ", ".join(f"`{file_name}`" for file_name in files), |
|
"sort_label": _santa_rule_group_sort_label( |
|
default_group + lockdown_group |
|
), |
|
} |
|
) |
|
return sorted( |
|
drift, |
|
key=lambda item: ( |
|
item["target_type"], |
|
item["sort_label"], |
|
item["target_identifier"], |
|
), |
|
) |
|
|
|
|
|
def _santa_rules_by_target( |
|
rules: list[dict[str, Any]], config_label: str |
|
) -> dict[tuple[str, str], list[dict[str, Any]]]: |
|
"""Group rules for one Santa config by the target Santa evaluates.""" |
|
grouped: dict[tuple[str, str], list[dict[str, Any]]] = {} |
|
for rule in rules: |
|
if _santa_config_label(rule) != config_label: |
|
continue |
|
scalars = rule["scalars"] |
|
target = ( |
|
scalars.get("target_type", "?"), |
|
scalars.get("target_identifier", "?"), |
|
) |
|
grouped.setdefault(target, []).append(rule) |
|
return grouped |
|
|
|
|
|
def _santa_config_label(rule: dict[str, Any]) -> str: |
|
"""`zentral_santa_configuration.default.id` -> `default`.""" |
|
config_ref = rule["scalars"].get("configuration_id", "?") |
|
return config_ref.split(".")[1] if "." in config_ref else config_ref |
|
|
|
|
|
def _santa_rule_group_fingerprints( |
|
rules: list[dict[str, Any]], |
|
) -> list[tuple[str, str]]: |
|
"""Behavioral comparison key for rules on the same target.""" |
|
return sorted( |
|
( |
|
rule["scalars"].get("policy", "?"), |
|
rule["scalars"].get("cel_expr", ""), |
|
) |
|
for rule in rules |
|
) |
|
|
|
|
|
def _santa_rule_group_summary(rules: list[dict[str, Any]]) -> str: |
|
"""Compact table cell for all rules on one target in one config.""" |
|
if not rules: |
|
return "—" |
|
return "; ".join( |
|
sorted( |
|
f"{rule['scalars'].get('policy', '?')} ({rule['label']}{_santa_cel_suffix(rule)})" |
|
for rule in rules |
|
) |
|
) |
|
|
|
|
|
def _santa_rule_group_sort_label(rules: list[dict[str, Any]]) -> str: |
|
"""First rule label for human-centered ordering within one target type.""" |
|
if not rules: |
|
return "" |
|
return sorted(rule["label"] for rule in rules)[0] |
|
|
|
|
|
def _santa_cel_suffix(rule: dict[str, Any]) -> str: |
|
"""Flag CEL policy rows without putting the full expression into the drift table.""" |
|
return ", CEL" if rule["scalars"].get("cel_expr") else "" |
|
|
|
|
|
def summarize_osquery(blocks: list[dict[str, Any]]) -> dict[str, Any]: |
|
"""Cursory osquery inventory: counts of queries/packs/configuration_packs/ATC/enrollments.""" |
|
counts: dict[str, int] = {} |
|
for block in blocks: |
|
if block["type"].startswith("zentral_osquery_"): |
|
counts[block["type"]] = counts.get(block["type"], 0) + 1 |
|
packs = [b["label"] for b in blocks if b["type"] == "zentral_osquery_pack"] |
|
return {"counts": counts, "packs": packs} |
|
|
|
|
|
SEVERITY_LABEL: dict[str, str] = { |
|
"track_gated": "🔴 track-gated, never prod", |
|
"partial": "🟡 partial rollout", |
|
"disabled": "⚪ reaches nobody", |
|
"audience_gated": "🟣 audience-targeted", |
|
} |
|
|
|
|
|
def write_toil( |
|
path: Path, |
|
now: datetime.datetime, |
|
hit_list: list[dict[str, Any]], |
|
munki: list[dict[str, Any]], |
|
mdm: list[dict[str, Any]], |
|
santa: dict[str, Any], |
|
osquery: dict[str, Any], |
|
audience: list[dict[str, Any]], |
|
) -> None: |
|
"""Render the dashboard markdown and overwrite TOIL.md.""" |
|
out: list[str] = [ |
|
"# zentral_config promotion toil", |
|
"", |
|
f"_Generated {now.strftime('%Y-%m-%dT%H:%M:%SZ')} by `toil.py`. Overwritten each run; " |
|
"see git history for diffs._", |
|
"", |
|
"Availability % of prod is to an **untagged** (" |
|
"prod / default) machine = `default_shard / shard_modulo`. " |
|
"Track tags (`testing_track`, `temp`) are the testing " |
|
"pipeline — gating to _only_ those means it never reached prod.", |
|
"", |
|
"## Hit list — promotions to finish", |
|
"", |
|
] |
|
if hit_list: |
|
out += [ |
|
"| sev | domain | artifact | % of prod | gated to | file |", |
|
"| --- | --- | --- | --- | --- | --- |", |
|
] |
|
out += [ |
|
f"| {SEVERITY_LABEL[item['severity']]} | {item['domain']} | {item['name']} " |
|
f"| {item['pct']}% | {item['gating']} | `{item['file']}` |" |
|
for item in hit_list |
|
] |
|
else: |
|
out.append("_Nothing gated below prod. Everything's promoted._") |
|
out += ["", "## A. Munki — sub-manifest pkg_info", ""] |
|
out += _domain_table(munki) |
|
out += ["", "## B. MDM — artifacts & blueprint assignments", ""] |
|
out += _domain_table(mdm) |
|
out += [ |
|
"", |
|
"## C. Santa — configuration inventory & rule drift", |
|
"", |
|
f"- Configurations: {', '.join(santa['configs']) or '—'}", |
|
f"- Enrollments: {santa['enrollments']}", |
|
f"- Rules: {santa['rule_total']} total — by config " |
|
f"{_fmt_counts(santa['by_config'])}; by policy {_fmt_counts(santa['by_policy'])}", |
|
f"- Rule drift: {_santa_drift_summary(santa['drift'])}", |
|
"", |
|
"### Default vs Lockdown+ Rule Drift", |
|
"", |
|
*_santa_drift_table(santa["drift"]), |
|
"", |
|
"## D. Osquery _(cursory inventory — promotion-state analysis TBD)_", |
|
"", |
|
f"- Resource counts: {_fmt_counts(osquery['counts'])}", |
|
f"- Packs: {', '.join(osquery['packs']) or '—'}", |
|
"- _Refine: extension dropper / agent registration rollout lives in the munki section; " |
|
"tie query/pack enrollment coverage here._", |
|
"", |
|
"## E. Audience-targeted _(not toil — verify intent)_", |
|
"", |
|
] |
|
if audience: |
|
out += ["| domain | artifact | gated to | file |", "| --- | --- | --- | --- |"] |
|
out += [ |
|
f"| {item['domain']} | {item['name']} | {item['gating']} | `{item['file']}` |" |
|
for item in audience |
|
] |
|
else: |
|
out.append("_None._") |
|
out += [ |
|
"", |
|
"## Map", |
|
"", |
|
"- Source of truth: this repo's `*.tf`, block-parsed (no provider/API calls).", |
|
"- Exit codes — `1`: no `*.tf` resources found, unreadable/undecodable `*.tf` file, " |
|
"or `TOIL.md` write failure.", |
|
"", |
|
] |
|
# collapse leading/consecutive blank lines (markdownlint MD012) and end with one newline (MD047) |
|
cleaned: list[str] = [] |
|
for line in out: |
|
if line == "" and (not cleaned or cleaned[-1] == ""): |
|
continue |
|
cleaned.append(line) |
|
try: |
|
path.write_text("\n".join(cleaned).rstrip("\n") + "\n", encoding="utf-8") |
|
except OSError as err: |
|
raise ToilError(f"could not write {path}: {err}") from err |
|
|
|
|
|
def _domain_table(findings: list[dict[str, Any]]) -> list[str]: |
|
"""Markdown rows for a per-domain section, or an all-clear line.""" |
|
if not findings: |
|
return ["_All prod._"] |
|
rows = [ |
|
"| sev | artifact | % of prod | gated to | file |", |
|
"| --- | --- | --- | --- | --- |", |
|
] |
|
rows += [ |
|
f"| {SEVERITY_LABEL[item['severity']]} | {item['name']} | {item['pct']}% " |
|
f"| {item['gating']} | `{item['file']}` |" |
|
for item in sorted( |
|
findings, key=lambda i: (SEVERITY_ORDER.get(i["severity"], 9), i["name"]) |
|
) |
|
] |
|
return rows |
|
|
|
|
|
def _santa_drift_table(drift: list[dict[str, str]]) -> list[str]: |
|
"""Markdown rows for Santa Default-vs-Lockdown+ rule differences.""" |
|
if not drift: |
|
return ["_No Default-vs-Lockdown+ rule drift._"] |
|
rows = [ |
|
"| Rule target kind | Santa target identifier | Default config rule(s) | " |
|
"Lockdown+ config rule(s) | Difference | Source file(s) |", |
|
"| --- | --- | --- | --- | --- | --- |", |
|
] |
|
rows += [ |
|
f"| {item['target_type']} | {item['target_identifier']} | {item['default']} " |
|
f"| {item['lockdown']} | {item['discrepancy']} | {item['file']} |" |
|
for item in drift |
|
] |
|
return rows |
|
|
|
|
|
def _santa_drift_summary(drift: list[dict[str, str]]) -> str: |
|
"""Human summary of missing and differing Santa rules.""" |
|
missing_default = sum( |
|
1 for item in drift if item["discrepancy"] == "missing from default" |
|
) |
|
missing_lockdown = sum( |
|
1 for item in drift if item["discrepancy"] == "missing from lockdown" |
|
) |
|
differs = sum(1 for item in drift if item["discrepancy"] == "rule differs") |
|
return ( |
|
f"missing from Default {missing_default}; missing from Lockdown+ {missing_lockdown}; " |
|
f"present in both but different {differs}" |
|
) |
|
|
|
|
|
def _fmt_counts(counts: dict[str, Any]) -> str: |
|
"""`{a: 1, b: 2}` -> `a 1, b 2`, sorted high-to-low.""" |
|
items = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0])) |
|
return ", ".join(f"{key} {value}" for key, value in items) or "—" |
|
|
|
|
|
if __name__ == "__main__": |
|
sys.exit(main()) |
