#!/usr/bin/env python3 """PeerCortex Daily Audit Mailer — detailed report via localhost:25""" import os, json, smtplib from datetime import datetime, timezone, timedelta from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.utils import make_msgid, formatdate SMTP_HOST = "localhost" SMTP_PORT = 25 MAIL_FROM = "ctx-report@fichtmueller.org" MAIL_TO = "rf@flexoptix.net" def td(content, color=None, align="left", mono=False, bold=False): ff = "monospace" if mono else "sans-serif" fw = "font-weight:700;" if bold else "" cl = "color:%s;" % color if color else "" return '%s' % (ff, align, cl, fw, content) def load_report(): for delta in [0, 1]: d = (datetime.now(timezone.utc) - timedelta(days=delta)).strftime("%Y-%m-%d") try: with open("/var/log/peercortex/audit-%s.json" % d) as f: return json.load(f), d except FileNotFoundError: continue return None, datetime.now(timezone.utc).strftime("%Y-%m-%d") def build(report, date_str): if not report: return ("

No report for %s

" % date_str, "PeerCortex Audit %s — no report" % date_str) counts = report.get("counts", {}) results = report.get("results", []) elapsed = report.get("elapsed_s", 0) pool = report.get("pool_size", 0) ok = counts.get("ok", 0) warn = counts.get("warning", 0) crit = counts.get("critical", 0) skip = counts.get("skip", 0) total = len(results) mins, secs = elapsed // 60, elapsed % 60 if crit == 0 and warn == 0: sc, emoji, text, spfx = "#22c55e", "✅", "All Clear — no issues detected", "✅ All Clear" elif crit == 0: sc, emoji, text, spfx = "#f97316", "⚠️", "%d warning(s)" % warn, "⚠️ %d Warning(s)" % warn else: sc, emoji, text, spfx = "#ef4444", "🚨", "%d CRITICAL issue(s) — action required" % crit, "🚨 %d CRITICAL" % crit subject = "PeerCortex Daily Audit %s — %s" % (date_str, spfx) # Issues issue_results = [r for r in results if r.get("status") in ("critical", "warning")] issues_html = "" if issue_results: rows = "" for r in issue_results: asn = str(r["asn"]) name = (r.get("data", {}).get("name") or "?")[:28] for sev, field, msg in r.get("issues", []): c = "#ef4444" if sev == "CRITICAL" else "#f97316" rows += "%s%s%s%s%s" % ( td("[%s]" % sev[:4], c, mono=True), td("AS" + asn, "#94a3b8", mono=True), td(name, "#e2e8f0"), td(field, "#64748b", mono=True), td(msg[:60], "#94a3b8")) issues_html = ( "

⚠ Issues (%d ASNs)

" % len(issue_results) + "%s
" % rows ) # Skipped skip_results = [r for r in results if r.get("status") == "skip"] skipped_html = "" if skip_results: rows = "" for r in skip_results: asn = str(r["asn"]) reason = r.get("issues", [["SKIP","","no response from PeerCortex"]])[0][2] if r.get("issues") else "no response from PeerCortex" url = "https://peercortex.org/?asn=" + asn rows += ("%s%s" "→ lookup") % ( td("AS" + asn, "#94a3b8", mono=True), td(reason, "#64748b"), url) skipped_html = ( "

⏭ Skipped ASNs (%d) — no PeerCortex response

" % skip + "

Possible causes: ASN not in any routing table, very large prefix count, or temporary API timeout.

" + "%s
" % rows ) # Low RPKI rpki_bad = sorted( [r for r in results if r.get("status") != "skip" and r.get("data", {}).get("rpki_pct", 100) < 60], key=lambda r: r.get("data", {}).get("rpki_pct", 100) )[:8] rpki_html = "" if rpki_bad: rows = "" for r in rpki_bad: d = r.get("data", {}) pct = d.get("rpki_pct", 0) bar = ("
" "
") % max(2, pct) rows += "%s%s%s%s%s" % ( td("AS" + str(r["asn"]), "#94a3b8", mono=True), td((d.get("name") or "?")[:28], "#e2e8f0"), td(d.get("rir", "?"), "#64748b", "center"), bar, td(str(pct) + "%", "#ef4444", "right", bold=True)) rpki_html = ( "

📉 Low RPKI Coverage (<60%%)

" + "%s
" % rows ) # All ASNs table sorted_results = sorted([r for r in results if r.get("status") != "skip"], key=lambda x: x["asn"]) header = ("%s%s%s%s%s%s%s%s%s%s%s" % ( td("ASN", "#64748b", "right", mono=True), td("Name", "#64748b"), td("RIR", "#64748b", "center"), td("CC", "#64748b", "center"), td("Pfx", "#64748b", "right"), td("RIPE Stat", "#64748b", "right"), td("RPKI%", "#64748b", "right"), td("IX", "#64748b", "right"), td("PDB IX", "#64748b", "right"), td("Fac", "#64748b", "right"), td("Status", "#64748b", "center"))) rows = "" for i, r in enumerate(sorted_results): d = r.get("data", {}) bg = "#0f172a" if i % 2 == 0 else "#111827" st = r.get("status", "ok") if st == "critical": st_html = '✗ CRIT' elif st == "warning": st_html = '⚠ WARN' else: st_html = '✓' pct = d.get("rpki_pct", 100) rpki_c = "#ef4444" if pct < 50 else "#f97316" if pct < 80 else "#94a3b8" ripe = str(d.get("ripe_stat_pfx", "—")) pdb = str(d.get("pdb_ix_cnt", "—")) rows += "%s%s%s%s%s%s%s%s%s%s%s" % ( bg, td("AS" + str(r["asn"]), "#64748b", "right", mono=True), td((d.get("name") or "?")[:28], "#e2e8f0"), td(d.get("rir", "?"), "#94a3b8", "center"), td(d.get("cc", "?"), "#94a3b8", "center"), td(str(d.get("pfx_total", 0)), "#94a3b8", "right", mono=True), td(ripe, "#64748b", "right", mono=True), td(str(pct) + "%", rpki_c, "right", mono=True), td(str(d.get("ix_cnt", 0)), "#94a3b8", "right", mono=True), td(pdb, "#64748b", "right", mono=True), td(str(d.get("fac_cnt", 0)), "#94a3b8", "right", mono=True), st_html) all_table = ( "

All Tested ASNs (%d)

" % len(sorted_results) + "
%s%s
" % (header, rows) ) html = """
PEERCORTEX Daily Audit Report · %(date)s
%(emoji)s %(text)s
%(total)d ASNs tested from pool of %(pool)d · Runtime: %(mins)dm %(secs)ds · Cross-validated: RIPE Stat + PeeringDB
%(ok)d
PERFECT
%(warn)d
WARNINGS
%(crit)d
CRITICAL
%(skip)d
SKIPPED
%(issues)s %(skipped)s %(rpki)s %(alltable)s
PeerCortex · peercortex.org · 100 ASNs rotate daily from %(pool)d ASN pool · Audit at 02:00 UTC, report at 06:00 UTC
""" % dict( date=date_str, sc=sc, emoji=emoji, text=text, total=total, pool=pool, mins=mins, secs=secs, ok=ok, warn=warn, crit=crit, skip=skip, issues=issues_html, skipped=skipped_html, rpki=rpki_html, alltable=all_table) return html, subject def send_email(subject, html_body): msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = MAIL_FROM msg["To"] = MAIL_TO msg["Date"] = formatdate(localtime=True) msg["Message-ID"] = make_msgid(domain="fichtmueller.org") msg.attach(MIMEText("PeerCortex Daily Audit — please view in an HTML-capable email client.\n\nhttps://peercortex.org", "plain", "utf-8")) msg.attach(MIMEText(html_body, "html", "utf-8")) with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=15) as s: s.sendmail(MAIL_FROM, [MAIL_TO], msg.as_bytes()) print("Sent '%s' -> %s" % (subject, MAIL_TO)) if __name__ == "__main__": report, date_str = load_report() html, subject = build(report, date_str) print("Email size: %d bytes" % len(html.encode())) send_email(subject, html)