"""
Sentinel Audit Results Analyser
=================================
Reads the CSV produced by sentinel_audit.py and produces:
1. A human-readable summary report (console + .txt file)
2. A per-rule scored CSV with PASS/WARN/FAIL ratings
3. A remediation backlog CSV sorted by priority
4. A column dictionary reference (embedded in summary report)
Usage:
python sentinel_analyse.py --input sentinel_audit_results.csv
--input Path to the audit CSV (default: sentinel_audit_results.csv)
--output Output folder for reports (default: ./audit_analysis)
"""
import os
import csv
import argparse
from collections import defaultdict
from datetime import datetime
# ── Scoring weights ────────────────────────────────────────────────────────────
# Each check contributes a penalty score. Higher = more critical.
# Total score per rule: 0 = perfect, higher = more issues.
CHECKS = [
# (column, condition_fn, penalty, severity, remediation_hint)
(
"enabled",
lambda v: v == "False",
5,
"WARN",
"Rule is disabled — confirm intentional and document reason",
),
(
"description_present",
lambda v: v == "False",
3,
"WARN",
"Add a meaningful description explaining what the rule detects and why",
),
(
"description_length",
lambda v: v.isdigit() and int(v) < 30,
2,
"WARN",
"Description is too short (< 30 chars) — likely a placeholder, expand it",
),
(
"query_present",
lambda v: v == "False",
10,
"FAIL",
"Rule has no KQL query — it will never produce alerts",
),
(
"frequency_period_gap_flag",
lambda v: v == "True",
8,
"FAIL",
"Query frequency exceeds lookback period — coverage gaps exist between runs. "
"Set queryFrequency <= queryPeriod",
),
(
"suppression_vs_frequency_flag",
lambda v: v == "True",
9,
"FAIL",
"Suppression duration >= frequency — rule may be permanently silenced after first alert. "
"Reduce suppressionDuration or increase queryFrequency",
),
(
"severity",
lambda v: v == "MISSING",
6,
"FAIL",
"Severity is not set — assign High/Medium/Low/Informational",
),
(
"severity_valid",
lambda v: v == "False",
4,
"WARN",
"Severity value is non-standard — use High, Medium, Low, or Informational",
),
(
"tactics_count",
lambda v: v.isdigit() and int(v) == 0,
4,
"WARN",
"No MITRE ATT&CK tactics assigned — add at least one tactic for coverage visibility",
),
(
"techniques_count",
lambda v: v.isdigit() and int(v) == 0,
2,
"WARN",
"No MITRE techniques assigned — add technique IDs for precise coverage mapping",
),
(
"tactics_valid_flag",
lambda v: v == "False",
3,
"WARN",
"One or more tactic names are not valid MITRE ATT&CK names — check spelling",
),
(
"incident_creation_enabled",
lambda v: v == "False",
7,
"FAIL",
"Incident creation is disabled — alerts fire silently with no SOC incident created. "
"Enable incidentConfiguration.createIncident",
),
(
"entity_mapping_present_flag",
lambda v: v == "False",
5,
"WARN",
"No entity mappings defined — add Account, Host, IP, or other entities to "
"enable investigation pivoting in Sentinel",
),
(
"alert_name_is_dynamic",
lambda v: v == "False",
2,
"WARN",
"Alert name is static — use dynamic fields like {{AccountName}} so each "
"alert is uniquely identifiable at triage",
),
(
"query_line_count",
lambda v: v.isdigit() and 0 < int(v) <= 2,
3,
"WARN",
"Query is very short (1-2 lines) — verify it is not a placeholder or overly broad",
),
]
MAX_POSSIBLE_SCORE = sum(c[2] for c in CHECKS)
SEVERITY_ORDER = {"FAIL": 0, "WARN": 1, "PASS": 2}
# ── Column dictionary ──────────────────────────────────────────────────────────
# Each entry: (column_name, description, good_bad_guidance)
COLUMN_DICTIONARY = [
# --- Identity ---
("--- IDENTITY ---", "", ""),
("file_name", "Source JSON filename on disk",
"Used as fallback identifier if rule name is missing"),
("rule_guid", "Unique GUID extracted from the ARM name field",
"UNKNOWN = GUID could not be parsed from the name field"),
("display_name", "Human-readable rule name shown in Sentinel",
"MISSING = rule has no display name set"),
("description_present", "Whether a description field exists and is non-empty",
"True = good | False = no description"),
("description_length", "Character count of the description",
"0 = no description | <30 chars usually means a placeholder"),
("rule_kind", "Rule type: Scheduled, NRT, Fusion, MicrosoftSecurityIncidentCreation, etc.",
"Determines which other columns are applicable"),
("enabled", "Whether the rule is active in Sentinel",
"True = active | False = disabled, will not fire"),
# --- Rule logic ---
("--- RULE LOGIC ---", "", ""),
("query_present", "Whether a non-empty KQL query exists",
"False = rule has no query and will never detect anything"),
("query_line_count", "Number of lines in the KQL query",
"1-2 lines may indicate an overly broad or placeholder query"),
("query_tables_referenced", "Pipe-separated list of data tables the KQL appears to query",
"NONE_DETECTED = no recognisable table names found, check query manually"),
("query_period", "How far back the query looks for data (e.g. PT1H=1hr, P1D=1day)",
"MISSING = rule has no lookback window defined"),
("query_frequency", "How often the rule runs (e.g. PT5M=every 5min, PT1H=every hour)",
"NRT rules do not have this field — shows N/A"),
("frequency_period_ratio", "frequency / period — relationship between run interval and lookback window",
"<=0.5 EXCELLENT | <=1.0 GOOD (no gaps) | >1.0 WARN/CRITICAL (coverage gaps exist)"),
("frequency_period_gap_flag", "True if frequency exceeds period — the rule has blind spots between runs",
"False = no gaps (good) | True = coverage gaps exist (bad)"),
("suppression_enabled", "Whether alert suppression is turned on",
"True = suppression active, verify duration vs frequency"),
("suppression_duration", "How long alerts are suppressed after the rule fires",
"Should always be shorter than the query frequency"),
("suppression_vs_frequency_flag", "True if suppression duration >= frequency — alerts may be permanently silenced",
"True = critical misconfiguration, rule may never alert twice"),
# --- Severity & MITRE ---
("--- SEVERITY & MITRE ---", "", ""),
("severity", "Alert severity: High, Medium, Low, Informational",
"MISSING = not set | All rules being Informational is a tuning red flag"),
("severity_valid", "Whether severity is one of the four accepted values",
"False = non-standard severity value"),
("tactics", "Pipe-separated MITRE ATT&CK tactic names assigned to the rule",
"NONE = no tactics mapped, reduces coverage visibility"),
("tactics_count", "Number of MITRE tactics assigned",
"0 = no MITRE coverage | 1-3 typical | >5 may indicate over-tagging"),
("tactics_valid_flag", "Whether all assigned tactics are recognised MITRE ATT&CK names",
"False = typo or non-standard tactic name present"),
("unknown_tactics", "Any tactic names not in the known MITRE ATT&CK list",
"None = all tactics are valid"),
("techniques", "Pipe-separated MITRE technique/sub-technique IDs (e.g. T1078, T1078.001)",
"NONE = no technique-level mapping"),
("techniques_count", "Number of MITRE techniques assigned",
"0 = no technique-level mapping"),
# --- Incident configuration ---
("--- INCIDENT CONFIGURATION ---", "", ""),
("incident_creation_enabled", "Whether the rule creates a Sentinel incident when it fires",
"False = alerts fire silently with no SOC incident created"),
("grouping_enabled", "Whether related alerts are grouped into a single incident",
"True = reduces alert noise (generally preferred)"),
("grouping_lookback", "Time window used to group related alerts together",
"Should align with the rule's query period"),
("grouping_reopen_closed", "Whether a new alert reopens a previously closed incident",
"Depends on SOC process — neither value is universally correct"),
("grouping_match_only", "Grouping method: AllEntities, AnyAlert, Selected, etc.",
"AllEntities = strictest | AnyAlert = broadest grouping"),
# --- Entity mapping & alert details ---
("--- ENTITY MAPPING & ALERT DETAILS ---", "", ""),
("entity_mapping_count", "Number of entity mappings defined on the rule",
"0 = no entities mapped, limits investigation pivoting in Sentinel"),
("entity_types_mapped", "Pipe-separated entity types (Account, Host, IP, URL, etc.)",
"NONE = no entity context attached to alerts"),
("entity_mapping_present_flag", "Whether at least one entity is mapped",
"False = no entity mapping at all (should be remediated)"),
("alert_name_format", "The alert name template — static string or dynamic formula",
"STATIC = every alert has the same name, harder to triage at volume"),
("alert_name_is_dynamic", "Whether the alert name uses dynamic fields like {{AccountName}}",
"True = alert names include contextual data (preferred)"),
("alert_description_format", "The alert description template",
"STATIC = generic description for every alert instance"),
("event_grouping_strategy", "How events within a single rule run are grouped",
"AlertPerResult = one alert per matching row (can be noisy at scale)"),
# --- Metadata ---
("--- METADATA ---", "", ""),
("kind_version", "Version of the rule template",
"MISSING = no version tracked, harder to manage changes"),
("last_modified_utc", "Timestamp of last modification (from export — may not reflect live state)",
"MISSING = modification history not captured in the export"),
("created_by", "Author name from the rule metadata",
"MISSING = no ownership information"),
("source_name", "Origin of the rule: Content Hub solution name or custom",
"Helps distinguish vendor templates from custom rules"),
("template_version", "Version string of the originating Content Hub template",
"MISSING = custom rule or version not tracked"),
# --- Operational health ---
("--- OPERATIONAL HEALTH (requires live access) ---", "", ""),
("query_syntax_valid", "Whether the KQL query executes without errors",
"N/A - No Live Access"),
("tables_exist_in_workspace", "Whether the tables referenced in the query are actively ingesting",
"N/A - No Live Access"),
("connector_status", "Whether the required data connectors are enabled and connected",
"N/A - No Live Access"),
("rule_last_run", "Timestamp of the most recent rule execution",
"N/A - No Live Access"),
("rule_last_fired", "Timestamp of the last time the rule produced an alert",
"N/A - No Live Access"),
# --- Audit metadata ---
("--- AUDIT METADATA ---", "", ""),
("audit_flags", "Pipe-separated list of all issues found for this rule",
"NONE = rule passed all assessable checks"),
("audit_timestamp", "When this audit row was generated (UTC)",
"Used to version the audit run"),
# --- Analysis columns (added by this script) ---
("--- ANALYSIS COLUMNS (added by sentinel_analyse.py) ---", "", ""),
("audit_score", "Penalty score: 0 = perfect, higher = more/worse issues",
f"Range: 0 to {MAX_POSSIBLE_SCORE} | 0 = PASS | 1-{int(MAX_POSSIBLE_SCORE*0.2)} = PASS WITH NOTES | "
f"{int(MAX_POSSIBLE_SCORE*0.2)+1}-{int(MAX_POSSIBLE_SCORE*0.5)} = WARN | "
f"{int(MAX_POSSIBLE_SCORE*0.5)+1}+ = FAIL"),
("audit_rating", "Overall rule rating derived from audit_score",
"PASS | PASS WITH NOTES | WARN | FAIL"),
("findings_count", "Total number of individual issues found on this rule",
"0 = no issues"),
("fail_count", "Number of FAIL-severity issues on this rule",
"0 = no critical issues"),
("warn_count", "Number of WARN-severity issues on this rule",
"0 = no warnings"),
("freq_ratio_rating", "Human-readable interpretation of frequency_period_ratio",
"EXCELLENT / GOOD / WARNING / CRITICAL"),
("remediation_actions", "Pipe-separated list of specific fix instructions for every issue found",
"Empty = no issues to fix"),
]
# ── Helpers ────────────────────────────────────────────────────────────────────
def rate_frequency_ratio(value: str) -> str:
"""Return a human-readable rating for frequency_period_ratio."""
try:
ratio = float(value)
except (ValueError, TypeError):
return "N/A"
if ratio <= 0:
return "N/A"
if ratio <= 0.5:
return "EXCELLENT (large overlap, no gaps)"
if ratio <= 1.0:
return "GOOD (some overlap, no gaps)"
if ratio <= 2.0:
return "WARNING (frequency > period, small gaps)"
return "CRITICAL (frequency >> period, large coverage gaps)"
def overall_rating(score: int, max_score: int) -> str:
pct = score / max_score if max_score else 0
if pct == 0:
return "PASS"
if pct <= 0.2:
return "PASS WITH NOTES"
if pct <= 0.5:
return "WARN"
return "FAIL"
def severity_badge(s: str) -> str:
badges = {"FAIL": "[FAIL]", "WARN": "[WARN]", "PASS": "[PASS]", "PASS WITH NOTES": "[NOTE]"}
return badges.get(s, s)
# ── Core analysis ──────────────────────────────────────────────────────────────
def analyse_rule(row: dict) -> dict:
"""Score a single rule row and return enriched result."""
findings = []
score = 0
kind = row.get("rule_kind", "")
for col, condition, penalty, sev, hint in CHECKS:
val = row.get(col, "")
# Skip checks that don't apply to this rule kind
logic_only = {"query_present", "frequency_period_gap_flag",
"suppression_vs_frequency_flag", "query_line_count",
"alert_name_is_dynamic"}
incident_only = {"incident_creation_enabled"}
if col in logic_only and kind not in ("Scheduled", "NRT"):
continue
if col in incident_only and kind in ("Fusion", "MLBehaviorAnalytics"):
continue
if condition(str(val)):
findings.append({
"column": col,
"severity": sev,
"penalty": penalty,
"hint": hint,
})
score += penalty
rating = overall_rating(score, MAX_POSSIBLE_SCORE)
# Enrich frequency ratio
freq_rating = rate_frequency_ratio(row.get("frequency_period_ratio", ""))
return {
"score": score,
"rating": rating,
"findings": findings,
"freq_ratio_rating": freq_rating,
}
def analyse_all(rows: list) -> list:
results = []
for row in rows:
analysis = analyse_rule(row)
results.append({**row, **{
"audit_score": analysis["score"],
"audit_rating": analysis["rating"],
"findings_count": len(analysis["findings"]),
"fail_count": sum(1 for f in analysis["findings"] if f["severity"] == "FAIL"),
"warn_count": sum(1 for f in analysis["findings"] if f["severity"] == "WARN"),
"freq_ratio_rating": analysis["freq_ratio_rating"],
"remediation_actions": " | ".join(f["hint"] for f in analysis["findings"]),
}})
return results
# ── Report generation ──────────────────────────────────────────────────────────
def format_column_dictionary() -> list:
"""Format the column dictionary as text lines for the report."""
lines = []
lines.append("=" * 70)
lines.append(" COLUMN DICTIONARY")
lines.append(" Reference guide for every column in the audit CSV output")
lines.append("=" * 70)
col_w = 38
desc_w = 35
lines.append(f" {'COLUMN':<{col_w}} {'DESCRIPTION':<{desc_w}} GOOD / BAD GUIDANCE")
lines.append(" " + "-" * 66)
for entry in COLUMN_DICTIONARY:
col, desc, guidance = entry
# Section headers
if col.startswith("---"):
lines.append("")
else:
# Wrap long descriptions
col_str = col[:col_w]
desc_str = desc[:desc_w]
lines.append(f" {col_str:<{col_w}} {desc_str:<{desc_w}} {guidance}")
lines.append("")
return lines
def print_summary(results: list, output_file=None):
lines = []
total = len(results)
pass_count = sum(1 for r in results if r["audit_rating"] in ("PASS", "PASS WITH NOTES"))
warn_count = sum(1 for r in results if r["audit_rating"] == "WARN")
fail_count = sum(1 for r in results if r["audit_rating"] == "FAIL")
disabled = sum(1 for r in results if str(r.get("enabled", "")) == "False")
no_entity = sum(1 for r in results if str(r.get("entity_mapping_present_flag", "")) == "False")
no_tactics = sum(1 for r in results if str(r.get("tactics_count", "0")).isdigit()
and int(r.get("tactics_count", 0)) == 0)
no_incident = sum(1 for r in results if str(r.get("incident_creation_enabled", "")) == "False")
gap_flag = sum(1 for r in results if str(r.get("frequency_period_gap_flag", "")) == "True")
suppression_flag = sum(1 for r in results if str(r.get("suppression_vs_frequency_flag", "")) == "True")
lines.append("=" * 70)
lines.append(" SENTINEL ANALYTICS RULE AUDIT — ANALYSIS REPORT")
lines.append(f" Generated: {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}")
lines.append("=" * 70)
lines.append("")
lines.append("OVERVIEW")
lines.append("-" * 40)
lines.append(f" Total rules assessed : {total}")
lines.append(f" PASS / PASS WITH NOTES : {pass_count} ({pct(pass_count, total)}%)")
lines.append(f" WARN : {warn_count} ({pct(warn_count, total)}%)")
lines.append(f" FAIL : {fail_count} ({pct(fail_count, total)}%)")
lines.append("")
lines.append("TOP FINDINGS")
lines.append("-" * 40)
lines.append(f" Rules disabled : {disabled} ({pct(disabled, total)}%)")
lines.append(f" No entity mapping : {no_entity} ({pct(no_entity, total)}%)")
lines.append(f" No MITRE tactics : {no_tactics} ({pct(no_tactics, total)}%)")
lines.append(f" Incident creation disabled : {no_incident} ({pct(no_incident, total)}%)")
lines.append(f" Coverage gaps (freq > period) : {gap_flag} ({pct(gap_flag, total)}%)")
lines.append(f" Suppression silencing alerts : {suppression_flag} ({pct(suppression_flag, total)}%)")
lines.append("")
# Severity distribution
sev_dist = defaultdict(int)
for r in results:
sev_dist[r.get("severity", "MISSING")] += 1
lines.append("SEVERITY DISTRIBUTION")
lines.append("-" * 40)
for sev, count in sorted(sev_dist.items(), key=lambda x: -x[1]):
lines.append(f" {sev:<20} : {count} ({pct(count, total)}%)")
lines.append("")
# Rule kind distribution
kind_dist = defaultdict(int)
for r in results:
kind_dist[r.get("rule_kind", "UNKNOWN")] += 1
lines.append("RULE KIND DISTRIBUTION")
lines.append("-" * 40)
for kind, count in sorted(kind_dist.items(), key=lambda x: -x[1]):
lines.append(f" {kind:<40} : {count}")
lines.append("")
# MITRE tactic coverage
tactic_counts = defaultdict(int)
for r in results:
tactics = r.get("tactics", "")
if tactics and tactics not in ("NONE", "N/A", ""):
for t in tactics.split("|"):
tactic_counts[t.strip()] += 1
if tactic_counts:
lines.append("MITRE TACTIC COVERAGE (rules per tactic)")
lines.append("-" * 40)
for tactic, count in sorted(tactic_counts.items(), key=lambda x: -x[1]):
bar = "#" * min(count, 40)
lines.append(f" {tactic:<35} : {count:>3} {bar}")
lines.append("")
# Top 10 worst rules
sorted_results = sorted(results, key=lambda x: -x["audit_score"])
lines.append("TOP 10 RULES NEEDING ATTENTION")
lines.append("-" * 40)
for r in sorted_results[:10]:
lines.append(
f" {severity_badge(r['audit_rating'])} "
f"Score:{r['audit_score']:>3} "
f"FAIL:{r['fail_count']} WARN:{r['warn_count']} "
f"{r.get('display_name', r.get('file_name', 'UNKNOWN'))[:50]}"
)
lines.append("")
lines.append("=" * 70)
lines.append(" OUTPUT FILES")
lines.append(" - scored_rules.csv : All rules with scores and ratings")
lines.append(" - remediation_backlog.csv: Rules with issues, sorted by priority")
lines.append(" - summary_report.txt : This report (includes column dictionary)")
lines.append("=" * 70)
lines.append("")
lines.extend(format_column_dictionary())
report = "\n".join(lines)
print(report)
if output_file:
with open(output_file, "w", encoding="utf-8") as f:
f.write(report)
return report
def pct(n, total):
if total == 0:
return 0
return round(n / total * 100)
def write_scored_csv(results: list, filepath: str):
if not results:
return
extra_cols = ["audit_score", "audit_rating", "findings_count",
"fail_count", "warn_count", "freq_ratio_rating", "remediation_actions"]
# Put extra cols right after audit_flags
base_cols = list(results[0].keys())
ordered = [c for c in base_cols if c not in extra_cols]
# Insert extra cols before audit_timestamp
insert_at = ordered.index("audit_timestamp") if "audit_timestamp" in ordered else len(ordered)
for i, col in enumerate(extra_cols):
ordered.insert(insert_at + i, col)
with open(filepath, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=ordered, extrasaction="ignore")
writer.writeheader()
writer.writerows(results)
def write_remediation_backlog(results: list, filepath: str):
"""Write only rules with issues, sorted by score descending, with clear action items."""
issues = [r for r in results if r["audit_rating"] not in ("PASS",)]
issues_sorted = sorted(issues, key=lambda x: (
SEVERITY_ORDER.get(x["audit_rating"], 99), -x["audit_score"]
))
cols = [
"priority_rank", "audit_rating", "audit_score", "fail_count", "warn_count",
"display_name", "rule_kind", "enabled", "severity", "file_name", "rule_guid",
"freq_ratio_rating", "remediation_actions", "audit_flags",
]
with open(filepath, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=cols, extrasaction="ignore")
writer.writeheader()
for i, row in enumerate(issues_sorted, 1):
row["priority_rank"] = i
writer.writerow(row)
return len(issues_sorted)
# ── Entry point ────────────────────────────────────────────────────────────────
def run(input_csv: str, output_folder: str):
if not os.path.isfile(input_csv):
print(f"Error: input file '{input_csv}' not found.")
return
os.makedirs(output_folder, exist_ok=True)
with open(input_csv, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
rows = list(reader)
if not rows:
print("No rows found in the audit CSV.")
return
print(f"Analysing {len(rows)} rule(s)...\n")
results = analyse_all(rows)
# Write outputs
scored_path = os.path.join(output_folder, "scored_rules.csv")
backlog_path = os.path.join(output_folder, "remediation_backlog.csv")
report_path = os.path.join(output_folder, "summary_report.txt")
write_scored_csv(results, scored_path)
backlog_count = write_remediation_backlog(results, backlog_path)
print_summary(results, report_path)
print(f"\nFiles written to '{output_folder}':")
print(f" scored_rules.csv ({len(results)} rules)")
print(f" remediation_backlog.csv ({backlog_count} rules needing attention)")
print(f" summary_report.txt")
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Analyse Sentinel audit CSV and produce scored reports."
)
parser.add_argument(
"--input",
default="sentinel_audit_results.csv",
help="Path to audit CSV from sentinel_audit.py (default: sentinel_audit_results.csv)",
)
parser.add_argument(
"--output",
default="./audit_analysis",
help="Output folder for analysis reports (default: ./audit_analysis)",
)
args = parser.parse_args()
run(args.input, args.output)
