From 396943d69d2ccc9ee50f93ff9cb4bccd03f57133 Mon Sep 17 00:00:00 2001 From: FestiveKyle Date: Wed, 15 Jul 2026 15:52:59 -0300 Subject: [PATCH 1/2] Add script for scan-based summary backfill --- services/summaries/scan_summaries.py | 491 +++++++++++++++++++++++++++ 1 file changed, 491 insertions(+) create mode 100644 services/summaries/scan_summaries.py diff --git a/services/summaries/scan_summaries.py b/services/summaries/scan_summaries.py new file mode 100644 index 0000000000..dece4a0747 --- /dev/null +++ b/services/summaries/scan_summaries.py @@ -0,0 +1,491 @@ +import os +import sys +import logging +import argparse +from datetime import date, datetime, timedelta + +from arango import ArangoClient +from dotenv import load_dotenv + +from summaries import CHARTS, SCOPES, ignore_domain, domain_scopes + +load_dotenv() + +DB_USER = os.getenv("DB_USER") +DB_PASS = os.getenv("DB_PASS") +DB_NAME = os.getenv("DB_NAME") +DB_URL = os.getenv("DB_URL") + +CHART_TARGETS = {"shadow": "chartSummaries_rebuild", "prod": "chartSummaries"} +ORG_TARGETS = {"shadow": "organizationSummaries_rebuild", "prod": "organizationSummaries"} + +logging.basicConfig(stream=sys.stdout, level=logging.INFO) + + +def worst_status(statuses): + if "fail" in statuses: + return "fail" + if "pass" in statuses: + return "pass" + return "info" + + +def ensure_collections(db, target): + for name in (CHART_TARGETS[target], ORG_TARGETS[target]): + if not db.has_collection(name): + db.create_collection(name) + + +DATE_PREFIX = r"^\d{4}-\d{2}-\d{2}" + + +def profile_scans(db): + """Report raw-scan form; return the earliest day-bucketable scan date. + + Scans whose timestamp isn't a "YYYY-MM-DD..." string match no day range and + are skipped by reconstruction. Count and sample them so a systematic format + change is visible rather than silently shrinking the rebuilt history. + """ + earliest = next( + db.aql.execute( + """ + FOR d IN dns + FILTER REGEX_TEST(d.timestamp, @pattern) + SORT d.timestamp ASC + LIMIT 1 + RETURN d.timestamp + """, + bind_vars={"pattern": DATE_PREFIX}, + ), + None, + ) + if earliest is None: + return None + + for collection in ("dns", "web"): + skipped = next( + db.aql.execute( + """ + FOR d IN @@collection + FILTER !REGEX_TEST(d.timestamp, @pattern) + COLLECT WITH COUNT INTO total + RETURN total + """, + bind_vars={"@collection": collection, "pattern": DATE_PREFIX}, + ), + 0, + ) + if skipped: + samples = list( + db.aql.execute( + """ + FOR d IN @@collection + FILTER !REGEX_TEST(d.timestamp, @pattern) + LIMIT 3 + RETURN d.timestamp + """, + bind_vars={"@collection": collection, "pattern": DATE_PREFIX}, + ) + ) + logging.warning( + f"{collection}: {skipped} scans have non-bucketable timestamps and will be " + f"skipped. Samples: {samples}" + ) + else: + logging.info(f"{collection}: all scan timestamps are day-bucketable.") + + missing_dmarc = next( + db.aql.execute( + "FOR d IN dns FILTER d.dmarc.status == null COLLECT WITH COUNT INTO total RETURN total" + ), + 0, + ) + if missing_dmarc: + logging.warning( + f"dns: {missing_dmarc} scans have no dmarc.status; they reconstruct as 'info' " + "and drop out of pass/fail totals." + ) + + logging.info(f"earliest day-bucketable dns scan: {earliest!r}") + return datetime.strptime(earliest[:10], "%Y-%m-%d").date() + + +def reconstruct_day(db, day): + """Return {domain_name: entry} for domains scanned on `day` (latest scan wins).""" + start = day.isoformat() + end = (day + timedelta(days=1)).isoformat() + + dns_cursor = db.aql.execute( + """ + FOR d IN dns + FILTER d.timestamp >= @start AND d.timestamp < @end + COLLECT domain = d.domain INTO scans = d + LET latest = FIRST(FOR s IN scans SORT s.timestamp DESC LIMIT 1 RETURN s) + RETURN { + domain: domain, + rcode: latest.rcode, + dmarc: latest.dmarc.status, + spf: latest.spf.status, + dkim: latest.dkim.status, + phase: latest.dmarc.phase, + dmarcTags: latest.dmarc.negativeTags, + spfTags: latest.spf.negativeTags, + dkimTags: latest.dkim.negativeTags + } + """, + bind_vars={"start": start, "end": end}, + ) + + updates = {} + for row in dns_cursor: + tags = ( + (row.get("dmarcTags") or []) + + (row.get("spfTags") or []) + + (row.get("dkimTags") or []) + ) + updates[row["domain"]] = { + "rcode": row.get("rcode"), + "phase": row.get("phase"), + "status": { + "dmarc": row.get("dmarc") or "info", + "spf": row.get("spf") or "info", + "dkim": row.get("dkim") or "info", + "https": "info", + "hsts": "info", + "ssl": "info", + }, + "dns_tags": tags, + "web_tags": [], + } + + web_cursor = db.aql.execute( + """ + WITH webScan + FOR w IN web + FILTER w.timestamp >= @start AND w.timestamp < @end + COLLECT domain = w.domain INTO webs = w + LET latest = FIRST(FOR x IN webs SORT x.timestamp DESC LIMIT 1 RETURN x) + LET scans = ( + FOR s IN 1..1 ANY latest._id webToWebScans + FILTER s.status == "complete" + RETURN { + https: s.results.connectionResults.httpsStatus, + hsts: s.results.connectionResults.hstsStatus, + ssl: s.results.tlsResult.sslStatus, + tlsTags: s.results.tlsResult.negativeTags, + connTags: s.results.connectionResults.negativeTags + } + ) + RETURN { domain: domain, scans: scans } + """, + bind_vars={"start": start, "end": end}, + ) + + for row in web_cursor: + entry = updates.get(row["domain"]) + if entry is None: + continue + scans = row["scans"] + entry["status"]["https"] = worst_status([s["https"] for s in scans]) + entry["status"]["hsts"] = worst_status([s["hsts"] for s in scans]) + entry["status"]["ssl"] = worst_status([s["ssl"] for s in scans]) + web_tags = [] + for s in scans: + web_tags.extend(s.get("tlsTags") or []) + web_tags.extend(s.get("connTags") or []) + entry["web_tags"] = web_tags + + return updates + + +def build_precomputed(db): + domain_by_name = {} + scopes_by_domain_id = {} + cursor = db.aql.execute( + """ + FOR d IN domains + RETURN { _id: d._id, domain: d.domain, archived: d.archived, blocked: d.blocked } + """ + ) + for dom in cursor: + domain_by_name[dom["domain"]] = dom + scopes_by_domain_id[dom["_id"]] = domain_scopes(dom, db) + + id_to_name = {dom["_id"]: name for name, dom in domain_by_name.items()} + orgs_by_domain_name = {} + for claim in db.collection("claims"): + if claim.get("assetState") != "approved": + continue + name = id_to_name.get(claim["_to"]) + if name is None: + continue + orgs_by_domain_name.setdefault(name, []).append(claim["_from"]) + + return domain_by_name, scopes_by_domain_id, orgs_by_domain_name + + +def new_org_acc(): + return { + "https": {"pass": 0, "fail": 0}, + "dmarc": {"pass": 0, "fail": 0}, + "web_connections": {"pass": 0, "fail": 0}, + "ssl": {"pass": 0, "fail": 0}, + "spf": {"pass": 0, "fail": 0}, + "dkim": {"pass": 0, "fail": 0}, + "web": {"pass": 0, "fail": 0}, + "mail": {"pass": 0, "fail": 0}, + "dmarc_phase": {"assess": 0, "deploy": 0, "enforce": 0, "maintain": 0}, + "negative_tags": {}, + } + + +def accumulate_chart(chart_summaries, dmarc_phases, scopes, status, phase): + for chart_type, scan_types in CHARTS.items(): + category_status = [status.get(scan_type) for scan_type in scan_types] + if "fail" in category_status: + result = "fail" + elif ( + chart_type == "mail" + and status.get("dkim") == "info" + and "pass" in category_status + ) or "info" not in category_status: + result = "pass" + else: + continue + for scope in scopes: + chart = chart_summaries[scope][chart_type] + chart[result] += 1 + chart["total"] += 1 + + if phase is None or status.get("dmarc") == "info": + return + if phase in ("assess", "deploy", "enforce", "maintain"): + for scope in scopes: + dmarc_phases[scope][phase] += 1 + + +def accumulate_org(acc, status, phase, tags): + https = status.get("https") + dmarc = status.get("dmarc") + hsts = status.get("hsts") + ssl = status.get("ssl") + spf = status.get("spf") + dkim = status.get("dkim") + + if https == "pass": + acc["https"]["pass"] += 1 + elif https == "fail": + acc["https"]["fail"] += 1 + if dmarc == "pass": + acc["dmarc"]["pass"] += 1 + elif dmarc == "fail": + acc["dmarc"]["fail"] += 1 + + if https == "pass" and hsts == "pass": + acc["web_connections"]["pass"] += 1 + elif https == "fail" or hsts == "fail": + acc["web_connections"]["fail"] += 1 + if ssl == "pass": + acc["ssl"]["pass"] += 1 + elif ssl == "fail": + acc["ssl"]["fail"] += 1 + if spf == "pass": + acc["spf"]["pass"] += 1 + elif spf == "fail": + acc["spf"]["fail"] += 1 + if dkim == "pass": + acc["dkim"]["pass"] += 1 + elif dkim == "fail": + acc["dkim"]["fail"] += 1 + + if ssl == "pass" and https == "pass": + acc["web"]["pass"] += 1 + elif ssl == "fail" or https == "fail": + acc["web"]["fail"] += 1 + if dkim == "info": + if dmarc == "pass" and spf == "pass": + acc["mail"]["pass"] += 1 + elif dmarc == "fail" or spf == "fail": + acc["mail"]["fail"] += 1 + else: + if dmarc == "pass" and spf == "pass" and dkim == "pass": + acc["mail"]["pass"] += 1 + elif dmarc == "fail" or spf == "fail" or dkim == "fail": + acc["mail"]["fail"] += 1 + + if phase is None or dmarc == "info": + return + if phase in ("assess", "deploy", "enforce", "maintain"): + acc["dmarc_phase"][phase] += 1 + for tag in tags: + acc["negative_tags"][tag] = acc["negative_tags"].get(tag, 0) + 1 + + +def to_category(metric): + return { + "pass": metric["pass"], + "fail": metric["fail"], + "total": metric["pass"] + metric["fail"], + } + + +def build_org_doc(org_id, day_iso, acc): + phases = acc["dmarc_phase"] + return { + "organization": org_id, + "date": day_iso, + "dmarc": to_category(acc["dmarc"]), + "web": to_category(acc["web"]), + "mail": to_category(acc["mail"]), + "dmarc_phase": {**phases, "total": sum(phases.values())}, + "https": to_category(acc["https"]), + "ssl": to_category(acc["ssl"]), + "spf": to_category(acc["spf"]), + "dkim": to_category(acc["dkim"]), + "web_connections": to_category(acc["web_connections"]), + "negative_tags": acc["negative_tags"], + } + + +def upsert_chart(col, day_iso, scope, charts, phases): + doc = {**charts, "dmarc_phase": {**phases, "total": sum(phases.values())}} + existing = col.find({"date": day_iso, "scope": scope}) + if existing.empty(): + col.insert({"date": day_iso, "scope": scope, **doc}) + else: + col.update_match({"date": day_iso, "scope": scope}, doc) + + +def upsert_org(col, day_iso, org_id, acc): + doc = build_org_doc(org_id, day_iso, acc) + existing = col.find({"organization": org_id, "date": day_iso}) + if existing.empty(): + col.insert(doc) + else: + existing_doc = existing.next() + col.update({"_key": existing_doc["_key"], **doc}) + + +def write_day( + chart_col, + org_col, + day, + state, + domain_by_name, + scopes_by_domain_id, + orgs_by_domain_name, +): + day_iso = day.isoformat() + + chart_summaries = { + scope: { + chart_type: {"scan_types": scan_types, "pass": 0, "fail": 0, "total": 0} + for chart_type, scan_types in CHARTS.items() + } + for scope in SCOPES + } + dmarc_phases = { + scope: {"assess": 0, "deploy": 0, "enforce": 0, "maintain": 0} + for scope in SCOPES + } + org_accs = {} + + for name, entry in state.items(): + domain = domain_by_name.get(name) + if domain is None: + continue + if ignore_domain( + { + "archived": domain.get("archived"), + "blocked": domain.get("blocked"), + "rcode": entry["rcode"], + } + ): + continue + + status = entry["status"] + phase = entry["phase"] + + scopes = scopes_by_domain_id.get(domain["_id"], set()) + if scopes: + accumulate_chart(chart_summaries, dmarc_phases, scopes, status, phase) + + tags = entry["dns_tags"] + entry["web_tags"] + for org_id in orgs_by_domain_name.get(name, []): + acc = org_accs.get(org_id) + if acc is None: + acc = new_org_acc() + org_accs[org_id] = acc + accumulate_org(acc, status, phase, tags) + + for scope in SCOPES: + upsert_chart(chart_col, day_iso, scope, chart_summaries[scope], dmarc_phases[scope]) + for org_id, acc in org_accs.items(): + upsert_org(org_col, day_iso, org_id, acc) + + +def run_backfill( + host=DB_URL, + name=DB_NAME, + user=DB_USER, + password=DB_PASS, + target="shadow", + start=None, + end=None, +): + client = ArangoClient(hosts=host) + db = client.db(name, username=user, password=password) + + ensure_collections(db, target) + + earliest = profile_scans(db) + if earliest is None: + logging.info("No DNS scans found; nothing to backfill.") + return + + write_start = datetime.strptime(start, "%Y-%m-%d").date() if start else earliest + write_end = datetime.strptime(end, "%Y-%m-%d").date() if end else date.today() + + chart_col = db.collection(CHART_TARGETS[target]) + org_col = db.collection(ORG_TARGETS[target]) + domain_by_name, scopes_by_domain_id, orgs_by_domain_name = build_precomputed(db) + + logging.info( + f"Backfilling {CHART_TARGETS[target]}/{ORG_TARGETS[target]} " + f"from {write_start} to {write_end} (accumulating from {earliest})..." + ) + + state = {} + day = earliest + while day <= write_end: + state.update(reconstruct_day(db, day)) + if day >= write_start: + write_day( + chart_col, + org_col, + day, + state, + domain_by_name, + scopes_by_domain_id, + orgs_by_domain_name, + ) + logging.info(f"Wrote summaries for {day.isoformat()} ({len(state)} domains in state).") + day += timedelta(days=1) + + logging.info("Backfill completed.") + + +def main(): + parser = argparse.ArgumentParser(description="Reconstruct chart/org summaries from raw scans.") + parser.add_argument("--target", choices=["shadow", "prod"], default="shadow") + parser.add_argument("--start", help="First day to write (YYYY-MM-DD). Defaults to earliest scan.") + parser.add_argument("--end", help="Last day to write (YYYY-MM-DD). Defaults to today.") + args = parser.parse_args() + + logging.info("Scan-based summary backfill started") + run_backfill(target=args.target, start=args.start, end=args.end) + logging.info("Scan-based summary backfill shutting down...") + + +if __name__ == "__main__": + main() From d516720721f132548bd5e0f59daab6703b3c5465 Mon Sep 17 00:00:00 2001 From: FestiveKyle Date: Mon, 20 Jul 2026 14:31:25 -0300 Subject: [PATCH 2/2] Optimize scan summary backfill. Add tests --- services/summaries/scan_summaries.py | 164 ++++++----- .../summaries/tests/test_scan_summaries.py | 256 ++++++++++++++++++ 2 files changed, 335 insertions(+), 85 deletions(-) create mode 100644 services/summaries/tests/test_scan_summaries.py diff --git a/services/summaries/scan_summaries.py b/services/summaries/scan_summaries.py index dece4a0747..7d61455861 100644 --- a/services/summaries/scan_summaries.py +++ b/services/summaries/scan_summaries.py @@ -23,6 +23,7 @@ def worst_status(statuses): + """Return the worst status in the list: fail, else pass, else info.""" if "fail" in statuses: return "fail" if "pass" in statuses: @@ -36,21 +37,31 @@ def ensure_collections(db, target): db.create_collection(name) +def warn_missing_source_indexes(db): + """Warn if dns/web are missing the timestamp index the daily reads rely on.""" + for name in ("dns", "web"): + has_index = any( + ix["type"] == "persistent" and ix["fields"] == ["timestamp"] + for ix in db.collection(name).indexes() + ) + if not has_index: + logging.warning( + f"{name} has no persistent 'timestamp' index; each day's read will full-scan " + f"{name}. Create it once with: db.{name}.ensureIndex(" + "{type:'persistent', fields:['timestamp'], inBackground:true})" + ) + + DATE_PREFIX = r"^\d{4}-\d{2}-\d{2}" def profile_scans(db): - """Report raw-scan form; return the earliest day-bucketable scan date. - - Scans whose timestamp isn't a "YYYY-MM-DD..." string match no day range and - are skipped by reconstruction. Count and sample them so a systematic format - change is visible rather than silently shrinking the rebuilt history. - """ + """Find the earliest day worth backfilling and flag scans we'll skip.""" earliest = next( db.aql.execute( """ FOR d IN dns - FILTER REGEX_TEST(d.timestamp, @pattern) + FILTER d.domain != null AND REGEX_TEST(d.timestamp, @pattern) SORT d.timestamp ASC LIMIT 1 RETURN d.timestamp @@ -67,7 +78,7 @@ def profile_scans(db): db.aql.execute( """ FOR d IN @@collection - FILTER !REGEX_TEST(d.timestamp, @pattern) + FILTER d.domain == null OR NOT REGEX_TEST(d.timestamp, @pattern) COLLECT WITH COUNT INTO total RETURN total """, @@ -76,49 +87,39 @@ def profile_scans(db): 0, ) if skipped: - samples = list( - db.aql.execute( - """ - FOR d IN @@collection - FILTER !REGEX_TEST(d.timestamp, @pattern) - LIMIT 3 - RETURN d.timestamp - """, - bind_vars={"@collection": collection, "pattern": DATE_PREFIX}, - ) - ) - logging.warning( - f"{collection}: {skipped} scans have non-bucketable timestamps and will be " - f"skipped. Samples: {samples}" - ) - else: - logging.info(f"{collection}: all scan timestamps are day-bucketable.") - - missing_dmarc = next( - db.aql.execute( - "FOR d IN dns FILTER d.dmarc.status == null COLLECT WITH COUNT INTO total RETURN total" - ), - 0, - ) - if missing_dmarc: - logging.warning( - f"dns: {missing_dmarc} scans have no dmarc.status; they reconstruct as 'info' " - "and drop out of pass/fail totals." - ) + logging.warning(f"{collection}: {skipped} scans lack a usable domain/timestamp; skipping them.") - logging.info(f"earliest day-bucketable dns scan: {earliest!r}") + logging.info(f"Earliest usable dns scan: {earliest!r}") return datetime.strptime(earliest[:10], "%Y-%m-%d").date() -def reconstruct_day(db, day): - """Return {domain_name: entry} for domains scanned on `day` (latest scan wins).""" +def new_state_entry(): + """Blank per-domain state: every field unknown ('info') until a scan fills it in.""" + return { + "rcode": None, + "phase": None, + "status": { + "dmarc": "info", + "spf": "info", + "dkim": "info", + "https": "info", + "hsts": "info", + "ssl": "info", + }, + "dns_tags": [], + "web_tags": [], + } + + +def reconstruct_day(db, day, state): + """Generate a day's state from its scans. Uses the existing state to carry forward any domains that had no scans that day.""" start = day.isoformat() end = (day + timedelta(days=1)).isoformat() dns_cursor = db.aql.execute( """ FOR d IN dns - FILTER d.timestamp >= @start AND d.timestamp < @end + FILTER d.domain != null AND d.timestamp >= @start AND d.timestamp < @end COLLECT domain = d.domain INTO scans = d LET latest = FIRST(FOR s IN scans SORT s.timestamp DESC LIMIT 1 RETURN s) RETURN { @@ -136,33 +137,24 @@ def reconstruct_day(db, day): bind_vars={"start": start, "end": end}, ) - updates = {} for row in dns_cursor: tags = ( (row.get("dmarcTags") or []) + (row.get("spfTags") or []) + (row.get("dkimTags") or []) ) - updates[row["domain"]] = { - "rcode": row.get("rcode"), - "phase": row.get("phase"), - "status": { - "dmarc": row.get("dmarc") or "info", - "spf": row.get("spf") or "info", - "dkim": row.get("dkim") or "info", - "https": "info", - "hsts": "info", - "ssl": "info", - }, - "dns_tags": tags, - "web_tags": [], - } + entry = state.setdefault(row["domain"], new_state_entry()) + entry["rcode"] = row.get("rcode") + entry["phase"] = row.get("phase") + entry["status"]["dmarc"] = row.get("dmarc") or "info" + entry["status"]["spf"] = row.get("spf") or "info" + entry["status"]["dkim"] = row.get("dkim") or "info" + entry["dns_tags"] = tags web_cursor = db.aql.execute( """ - WITH webScan FOR w IN web - FILTER w.timestamp >= @start AND w.timestamp < @end + FILTER w.domain != null AND w.timestamp >= @start AND w.timestamp < @end COLLECT domain = w.domain INTO webs = w LET latest = FIRST(FOR x IN webs SORT x.timestamp DESC LIMIT 1 RETURN x) LET scans = ( @@ -182,10 +174,10 @@ def reconstruct_day(db, day): ) for row in web_cursor: - entry = updates.get(row["domain"]) - if entry is None: - continue scans = row["scans"] + if not scans: + continue + entry = state.setdefault(row["domain"], new_state_entry()) entry["status"]["https"] = worst_status([s["https"] for s in scans]) entry["status"]["hsts"] = worst_status([s["hsts"] for s in scans]) entry["status"]["ssl"] = worst_status([s["ssl"] for s in scans]) @@ -195,10 +187,9 @@ def reconstruct_day(db, day): web_tags.extend(s.get("connTags") or []) entry["web_tags"] = web_tags - return updates - def build_precomputed(db): + """Load the domain, scope, and org lookups reused for every day.""" domain_by_name = {} scopes_by_domain_id = {} cursor = db.aql.execute( @@ -240,6 +231,7 @@ def new_org_acc(): def accumulate_chart(chart_summaries, dmarc_phases, scopes, status, phase): + """Add one domain's result to the chart summaries for each of its scopes.""" for chart_type, scan_types in CHARTS.items(): category_status = [status.get(scan_type) for scan_type in scan_types] if "fail" in category_status: @@ -265,6 +257,7 @@ def accumulate_chart(chart_summaries, dmarc_phases, scopes, status, phase): def accumulate_org(acc, status, phase, tags): + """Add one domain's result to its organization's running counters.""" https = status.get("https") dmarc = status.get("dmarc") hsts = status.get("hsts") @@ -322,6 +315,7 @@ def accumulate_org(acc, status, phase, tags): def to_category(metric): + """Turn a pass/fail pair into a pass/fail/total block.""" return { "pass": metric["pass"], "fail": metric["fail"], @@ -330,6 +324,7 @@ def to_category(metric): def build_org_doc(org_id, day_iso, acc): + """Shape one org's counters into a summary document.""" phases = acc["dmarc_phase"] return { "organization": org_id, @@ -347,25 +342,6 @@ def build_org_doc(org_id, day_iso, acc): } -def upsert_chart(col, day_iso, scope, charts, phases): - doc = {**charts, "dmarc_phase": {**phases, "total": sum(phases.values())}} - existing = col.find({"date": day_iso, "scope": scope}) - if existing.empty(): - col.insert({"date": day_iso, "scope": scope, **doc}) - else: - col.update_match({"date": day_iso, "scope": scope}, doc) - - -def upsert_org(col, day_iso, org_id, acc): - doc = build_org_doc(org_id, day_iso, acc) - existing = col.find({"organization": org_id, "date": day_iso}) - if existing.empty(): - col.insert(doc) - else: - existing_doc = existing.next() - col.update({"_key": existing_doc["_key"], **doc}) - - def write_day( chart_col, org_col, @@ -375,6 +351,7 @@ def write_day( scopes_by_domain_id, orgs_by_domain_name, ): + """Roll the day's state into chart + org summaries and save them.""" day_iso = day.isoformat() chart_summaries = { @@ -418,10 +395,25 @@ def write_day( org_accs[org_id] = acc accumulate_org(acc, status, phase, tags) + chart_docs = [] for scope in SCOPES: - upsert_chart(chart_col, day_iso, scope, chart_summaries[scope], dmarc_phases[scope]) + phases = dmarc_phases[scope] + chart_docs.append({ + "_key": f"{day_iso}:{scope}", + "date": day_iso, + "scope": scope, + **chart_summaries[scope], + "dmarc_phase": {**phases, "total": sum(phases.values())}, + }) + org_docs = [] for org_id, acc in org_accs.items(): - upsert_org(org_col, day_iso, org_id, acc) + doc = build_org_doc(org_id, day_iso, acc) + doc["_key"] = f"{day_iso}:{org_id.split('/')[-1]}" + org_docs.append(doc) + + chart_col.insert_many(chart_docs, overwrite_mode="replace") + if org_docs: + org_col.insert_many(org_docs, overwrite_mode="replace") def run_backfill( @@ -433,10 +425,12 @@ def run_backfill( start=None, end=None, ): + """Rebuild summaries day by day, from the first scan up to the end date.""" client = ArangoClient(hosts=host) db = client.db(name, username=user, password=password) ensure_collections(db, target) + warn_missing_source_indexes(db) earliest = profile_scans(db) if earliest is None: @@ -458,7 +452,7 @@ def run_backfill( state = {} day = earliest while day <= write_end: - state.update(reconstruct_day(db, day)) + reconstruct_day(db, day, state) if day >= write_start: write_day( chart_col, diff --git a/services/summaries/tests/test_scan_summaries.py b/services/summaries/tests/test_scan_summaries.py new file mode 100644 index 0000000000..255d5b1bf6 --- /dev/null +++ b/services/summaries/tests/test_scan_summaries.py @@ -0,0 +1,256 @@ +import os + +import pytest +from datetime import date, timedelta +from arango import ArangoClient +from dotenv import load_dotenv + +from scan_summaries import run_backfill + +load_dotenv(os.path.join(os.path.dirname(__file__), "test.env")) + +DB_URL = os.getenv("DB_URL", "http://localhost:8530") +DB_USER = os.getenv("DB_USER", "root") +DB_PASS = os.getenv("DB_PASS", "test") + +TODAY = date.today().isoformat() +TIMESTAMP = f"{TODAY} 12:00:00.000000+00:00" + + +def dns_doc(domain, dmarc, spf, dkim, phase): + return { + "domain": domain, + "timestamp": TIMESTAMP, + "rcode": "NOERROR", + "dmarc": {"status": dmarc, "phase": phase, "negativeTags": []}, + "spf": {"status": spf, "negativeTags": []}, + "dkim": {"status": dkim, "negativeTags": []}, + } + + +def web_scan_results(https, hsts, ssl): + return { + "status": "complete", + "results": { + "connectionResults": {"httpsStatus": https, "hstsStatus": hsts, "negativeTags": []}, + "tlsResult": {"sslStatus": ssl, "negativeTags": []}, + }, + } + + +class TestScanSummaries: + @pytest.fixture + def arango_db(self): + arango_client = ArangoClient(hosts=DB_URL) + sys_db = arango_client.db("_system", username=DB_USER, password=DB_PASS) + + db_name = os.path.basename(__file__).split(".")[0] + if sys_db.has_database(db_name): + sys_db.delete_database(db_name) + sys_db.create_database(db_name) + + db = arango_client.db(db_name, username=DB_USER, password=DB_PASS) + dns = db.create_collection("dns") + web = db.create_collection("web") + web_scan = db.create_collection("webScan") + graph = db.create_graph("compliance") + domains = graph.create_vertex_collection("domains") + orgs = graph.create_vertex_collection("organizations") + claims = graph.create_edge_definition( + edge_collection="claims", + from_vertex_collections=["organizations"], + to_vertex_collections=["domains"], + ) + web_to_web_scans = graph.create_edge_definition( + edge_collection="webToWebScans", + from_vertex_collections=["web"], + to_vertex_collections=["webScan"], + ) + + org = orgs.insert( + { + "_key": "tbs", + "verified": True, + "policies": {"psd": True, "pgs": False}, + "orgDetails": {"en": {"name": "Treasury Board of Canada Secretariat"}}, + } + ) + + specs = [ + ("tbs1.gc.ca", "pass", "pass", "fail", "not implemented", "pass", "pass", "pass"), + ("tbs2.gc.ca", "pass", "pass", "pass", "maintain", "pass", "pass", "pass"), + ("tbs3.gc.ca", "fail", "fail", "fail", "maintain", "fail", "fail", "fail"), + ] + for name, dmarc, spf, dkim, phase, https, hsts, ssl in specs: + domain = domains.insert({"domain": name, "archived": False, "blocked": False}) + claims.insert({"_from": org["_id"], "_to": domain["_id"], "assetState": "approved"}) + dns.insert(dns_doc(name, dmarc, spf, dkim, phase)) + web_entry = web.insert({"domain": name, "timestamp": TIMESTAMP}) + scan = web_scan.insert(web_scan_results(https, hsts, ssl)) + web_to_web_scans.insert({"_from": web_entry["_id"], "_to": scan["_id"]}) + + yield db + + sys_db.delete_database(db_name) + + def test_chart_rebuild_matches_domain_based_output(self, arango_db): + db_name = os.path.basename(__file__).split(".")[0] + run_backfill(host=DB_URL, name=db_name, user=DB_USER, password=DB_PASS, target="shadow") + + chart_summaries = arango_db.collection("chartSummaries_rebuild") + + def get_scope(scope): + summary = chart_summaries.find({"date": TODAY, "scope": scope}).next() + for k in ("_id", "_key", "_rev"): + summary.pop(k, None) + return summary + + populated = { + "date": TODAY, + "https": {"scan_types": ["https"], "pass": 2, "fail": 1, "total": 3}, + "dmarc": {"scan_types": ["dmarc"], "pass": 2, "fail": 1, "total": 3}, + "web_connections": {"scan_types": ["https", "hsts"], "pass": 2, "fail": 1, "total": 3}, + "ssl": {"scan_types": ["ssl"], "pass": 2, "fail": 1, "total": 3}, + "spf": {"scan_types": ["spf"], "pass": 2, "fail": 1, "total": 3}, + "dkim": {"scan_types": ["dkim"], "pass": 1, "fail": 2, "total": 3}, + "mail": {"scan_types": ["dmarc", "spf", "dkim"], "pass": 1, "fail": 2, "total": 3}, + "web": {"scan_types": ["https", "hsts", "ssl"], "pass": 2, "fail": 1, "total": 3}, + "dmarc_phase": {"assess": 0, "deploy": 0, "enforce": 0, "maintain": 2, "total": 2}, + } + + for scope in ("all", "verified", "psd"): + assert get_scope(scope) == {**populated, "scope": scope} + + assert get_scope("pgs") == { + "date": TODAY, + "scope": "pgs", + "https": {"scan_types": ["https"], "pass": 0, "fail": 0, "total": 0}, + "dmarc": {"scan_types": ["dmarc"], "pass": 0, "fail": 0, "total": 0}, + "web_connections": {"scan_types": ["https", "hsts"], "pass": 0, "fail": 0, "total": 0}, + "ssl": {"scan_types": ["ssl"], "pass": 0, "fail": 0, "total": 0}, + "spf": {"scan_types": ["spf"], "pass": 0, "fail": 0, "total": 0}, + "dkim": {"scan_types": ["dkim"], "pass": 0, "fail": 0, "total": 0}, + "mail": {"scan_types": ["dmarc", "spf", "dkim"], "pass": 0, "fail": 0, "total": 0}, + "web": {"scan_types": ["https", "hsts", "ssl"], "pass": 0, "fail": 0, "total": 0}, + "dmarc_phase": {"assess": 0, "deploy": 0, "enforce": 0, "maintain": 0, "total": 0}, + } + + def test_org_rebuild_matches_domain_based_output(self, arango_db): + db_name = os.path.basename(__file__).split(".")[0] + run_backfill(host=DB_URL, name=db_name, user=DB_USER, password=DB_PASS, target="shadow") + + summary = ( + arango_db.collection("organizationSummaries_rebuild") + .find({"organization": "organizations/tbs", "date": TODAY}) + .next() + ) + for k in ("_id", "_key", "_rev"): + summary.pop(k, None) + + assert summary == { + "organization": "organizations/tbs", + "date": TODAY, + "dmarc": {"pass": 2, "fail": 1, "total": 3}, + "https": {"pass": 2, "fail": 1, "total": 3}, + "web": {"pass": 2, "fail": 1, "total": 3}, + "mail": {"pass": 1, "fail": 2, "total": 3}, + "web_connections": {"pass": 2, "fail": 1, "total": 3}, + "ssl": {"pass": 2, "fail": 1, "total": 3}, + "dkim": {"pass": 1, "fail": 2, "total": 3}, + "spf": {"pass": 2, "fail": 1, "total": 3}, + "dmarc_phase": {"assess": 0, "deploy": 0, "enforce": 0, "maintain": 2, "total": 2}, + "negative_tags": {}, + } + + +YESTERDAY = (date.today() - timedelta(days=1)).isoformat() + + +class TestCarryForwardWebStatus: + """A day with no completed web scan should carry the prior web status forward, + not reset https/hsts/ssl to 'info' and drop the domain from the totals.""" + + @pytest.fixture + def arango_db(self): + arango_client = ArangoClient(hosts=DB_URL) + sys_db = arango_client.db("_system", username=DB_USER, password=DB_PASS) + + db_name = "test_scan_summaries_carry" + if sys_db.has_database(db_name): + sys_db.delete_database(db_name) + sys_db.create_database(db_name) + + db = arango_client.db(db_name, username=DB_USER, password=DB_PASS) + dns = db.create_collection("dns") + web = db.create_collection("web") + web_scan = db.create_collection("webScan") + graph = db.create_graph("compliance") + domains = graph.create_vertex_collection("domains") + orgs = graph.create_vertex_collection("organizations") + claims = graph.create_edge_definition( + edge_collection="claims", + from_vertex_collections=["organizations"], + to_vertex_collections=["domains"], + ) + web_to_web_scans = graph.create_edge_definition( + edge_collection="webToWebScans", + from_vertex_collections=["web"], + to_vertex_collections=["webScan"], + ) + + org = orgs.insert( + { + "_key": "tbs", + "verified": True, + "policies": {"psd": False, "pgs": False}, + "orgDetails": {"en": {"name": "Treasury Board of Canada Secretariat"}}, + } + ) + domain = domains.insert({"domain": "keep.gc.ca", "archived": False, "blocked": False}) + claims.insert({"_from": org["_id"], "_to": domain["_id"], "assetState": "approved"}) + + def add_dns(day): + dns.insert(dns_doc("keep.gc.ca", "pass", "pass", "pass", "maintain") | {"timestamp": f"{day} 12:00:00.000000+00:00"}) + + def add_web(day, scan_status): + web_entry = web.insert({"domain": "keep.gc.ca", "timestamp": f"{day} 12:00:00.000000+00:00"}) + scan = web_scan.insert(web_scan_results("pass", "pass", "pass") | {"status": scan_status}) + web_to_web_scans.insert({"_from": web_entry["_id"], "_to": scan["_id"]}) + + # Yesterday: full dns + completed web scan (https/hsts/ssl = pass) + add_dns(YESTERDAY) + add_web(YESTERDAY, "complete") + # Today: dns scan again, but the web scan is still pending (not complete) + add_dns(TODAY) + add_web(TODAY, "pending") + + yield db + + sys_db.delete_database(db_name) + + def test_incomplete_web_scan_carries_previous_status_forward(self, arango_db): + run_backfill(host=DB_URL, name="test_scan_summaries_carry", user=DB_USER, password=DB_PASS, target="shadow") + + summary = ( + arango_db.collection("organizationSummaries_rebuild") + .find({"organization": "organizations/tbs", "date": TODAY}) + .next() + ) + for k in ("_id", "_key", "_rev"): + summary.pop(k, None) + + # Web metrics still show yesterday's pass, not a dropped/reset domain. + assert summary == { + "organization": "organizations/tbs", + "date": TODAY, + "dmarc": {"pass": 1, "fail": 0, "total": 1}, + "https": {"pass": 1, "fail": 0, "total": 1}, + "web": {"pass": 1, "fail": 0, "total": 1}, + "mail": {"pass": 1, "fail": 0, "total": 1}, + "web_connections": {"pass": 1, "fail": 0, "total": 1}, + "ssl": {"pass": 1, "fail": 0, "total": 1}, + "dkim": {"pass": 1, "fail": 0, "total": 1}, + "spf": {"pass": 1, "fail": 0, "total": 1}, + "dmarc_phase": {"assess": 0, "deploy": 0, "enforce": 0, "maintain": 1, "total": 1}, + "negative_tags": {}, + }