diff --git a/backend/app/main.py b/backend/app/main.py index e92fad8..7f7d9ec 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -35,6 +35,7 @@ from app.services.diagnostics import build_system_diagnostics, test_export_dependencies from app.services.executive_risk import build_executive_risk from app.services.nmap_audit import audit_service_versions +from app.services.dnsdumpster_like_inventory import build_dnsdumpster_like_inventory from app.services.legal_terms import legal_payload, create_acceptance, validate_acceptance from app.services.attack_graph import build_attack_graph @@ -155,6 +156,14 @@ async def create_audit(payload: AuditRequest, request: Request, db=Depends(get_d passive_cves = detect_passive_cves(web_result) cti_result = audit_cti(ip_inventory, domain) service_scan = await asyncio.to_thread(audit_service_versions, domain, ip_inventory) + dnsdumpster_like = await build_dnsdumpster_like_inventory( + domain, + dns_result, + mail_result, + subdomains_result, + ip_inventory, + service_scan, + ) raw_findings = collect_findings( dns_result, @@ -193,6 +202,7 @@ async def create_audit(payload: AuditRequest, request: Request, db=Depends(get_d "ip_inventory": ip_inventory, "passive_cves": passive_cves, "service_scan": service_scan, + "dnsdumpster_like": dnsdumpster_like, "cti": cti_result, "patching_sla": patching_sla, "executive_risk": executive_risk, @@ -275,6 +285,17 @@ async def create_audit(payload: AuditRequest, request: Request, db=Depends(get_d "items": passive_cves.get("items", [])[:30], "note": passive_cves.get("note"), }, + "dnsdumpster_like": { + "system_locations": dnsdumpster_like.get("system_locations", {}), + "hosting_networks": dnsdumpster_like.get("hosting_networks", [])[:50], + "services_banners": dnsdumpster_like.get("services_banners", {}), + "a_records": dnsdumpster_like.get("a_records", [])[:80], + "mx_records": dnsdumpster_like.get("mx_records", [])[:50], + "ns_records": dnsdumpster_like.get("ns_records", [])[:50], + "txt_records": dnsdumpster_like.get("txt_records", [])[:80], + "hosts": dnsdumpster_like.get("hosts", [])[:80], + "policy": dnsdumpster_like.get("policy", {}), + }, "service_scan": { "enabled": service_scan.get("enabled"), "mode": service_scan.get("mode"), diff --git a/backend/app/reports/excel_report.py b/backend/app/reports/excel_report.py index e9f6032..6526423 100644 --- a/backend/app/reports/excel_report.py +++ b/backend/app/reports/excel_report.py @@ -42,6 +42,7 @@ def generate_excel_report(audit: dict) -> str: ws.title = "Synthese Direction" _sheet_summary(ws, audit) + _sheet_dnsdumpster_like(wb, audit) _sheet_action_plan(wb, audit) _sheet_findings(wb, audit) _sheet_exposure(wb, audit) @@ -136,6 +137,100 @@ def _sheet_summary(ws, audit: dict): _format(ws, freeze=None) + +def _sheet_dnsdumpster_like(wb, audit): + dd = audit.get("dnsdumpster_like", {}) or {} + summary = wb.create_sheet("DNSDumpster-like Summary") + summary_rows = [ + ["Domain", dd.get("domain") or audit.get("domain")], + ["System Locations", len(dd.get("system_locations", {}) or {})], + ["Hosting / Networks", len(dd.get("hosting_networks", []) or [])], + ["Services / Banners", len(dd.get("services_banners", {}) or {})], + ["A Records enrichis", len(dd.get("a_records", []) or [])], + ["MX Records enrichis", len(dd.get("mx_records", []) or [])], + ["NS Records enrichis", len(dd.get("ns_records", []) or [])], + ["TXT Records", len(dd.get("txt_records", []) or [])], + ["Note", "Nmap est une source complémentaire ; HTTP/HTTPS/TLS sont intégrés même si Nmap est vide."], + ] + _table_sheet(summary, "DNSDUMPSTER-LIKE SUMMARY", ["Indicateur", "Valeur"], summary_rows) + + ws = wb.create_sheet("System Locations") + rows = [[country, count] for country, count in (dd.get("system_locations", {}) or {}).items()] + _table_sheet(ws, "SYSTEM LOCATIONS", ["Country", "Count"], rows or [["Unknown", 0]]) + + ws = wb.create_sheet("Hosting Networks") + rows = [[h.get("asn"), h.get("network"), h.get("asn_name"), h.get("country"), h.get("count")] for h in dd.get("hosting_networks", [])] + _table_sheet(ws, "HOSTING / NETWORKS", ["ASN", "Network", "ASN Name", "Country", "Hosts"], rows) + + ws = wb.create_sheet("Services Banners") + rows = [[banner, count] for banner, count in (dd.get("services_banners", {}) or {}).items()] + _table_sheet(ws, "SERVICES / BANNERS", ["Banner", "Count"], rows) + + for title, key, headers, fn in [ + ("A Records", "a_records", _host_headers(), _host_row), + ("MX Records", "mx_records", ["Priority"] + _host_headers(), lambda h: [h.get("priority")] + _host_row(h)), + ("NS Records", "ns_records", _host_headers(), _host_row), + ("Host Inventory", "hosts", _host_headers(), _host_row), + ]: + ws = wb.create_sheet(title) + _table_sheet(ws, title.upper(), headers, [fn(h) for h in dd.get(key, [])]) + + ws = wb.create_sheet("TXT Records") + rows = [] + for item in dd.get("txt_records", []) or []: + spf = item.get("spf") or {} + rows.append([item.get("type"), item.get("value"), "\n".join(spf.get("includes", [])), "\n".join(spf.get("ip4", [])), "\n".join(spf.get("ip6", [])), "\n".join(spf.get("providers", [])), ", ".join(item.get("sources", []))]) + _table_sheet(ws, "TXT RECORDS", ["Type", "Value", "SPF Includes", "SPF ip4", "SPF ip6", "Providers", "Sources"], rows) + + +def _host_headers(): + return ["Host", "IP", "ASN", "Network", "ASN Name", "Country", "Provider", "Open Services", "HTTP Status", "HTTPS Status", "Final URL", "Title", "Server Banner", "TLS CN", "TLS SAN", "TLS Issuer", "TLS Expiration", "Technologies", "Nmap Services", "RevIP", "Sources"] + + +def _host_row(h): + http = h.get("http") or {} + https = h.get("https") or {} + tls = h.get("tls") or {} + services = _services_text(h.get("open_services", [])) or "Non détecté" + server_banner = "; ".join(dict.fromkeys([s.get("banner") for s in h.get("open_services", []) if s.get("banner")])) + title = https.get("title") or http.get("title") or next((s.get("title") for s in h.get("open_services", []) if s.get("title") and s.get("title") != "Non détecté"), "Non détecté") + return [ + h.get("host"), h.get("ip"), h.get("asn"), h.get("network"), h.get("asn_name"), h.get("country"), h.get("provider"), + services, http.get("status_code") or "Non détecté", https.get("status_code") or "Non détecté", https.get("final_url") or http.get("final_url") or "Non détecté", title, + server_banner or "Non détecté", tls.get("cn") or "Non détecté", "\n".join(tls.get("san", []) or []), tls.get("issuer") or "Non détecté", tls.get("expires_at") or "Non détecté", + _tech_text(h.get("technologies", [])) or "Non détecté", _nmap_text(h.get("nmap_services", [])) or "Non détecté par Nmap", h.get("revip_count", 0), ", ".join(h.get("sources", [])), + ] + + +def _services_text(services): + parts = [] + for s in services or []: + seg = f"{s.get('scheme')}: {s.get('banner') or s.get('service') or 'unknown server'}" + if s.get("title") and s.get("title") != "Non détecté": + seg += f"; title: {s.get('title')}" + parts.append(seg) + return "\n".join(parts) + + +def _tech_text(technologies): + out = [] + for t in technologies or []: + name = t.get("name") or "" + if t.get("version"): + name += f":{t.get('version')}" + if t.get("note"): + name += f" ({t.get('note')})" + out.append(name) + return "\n".join(dict.fromkeys(out)) + + +def _nmap_text(services): + out = [] + for s in services or []: + version = s.get("version") or "Version non exposée" + out.append(f"{s.get('port')}/{s.get('protocol', 'tcp')} {s.get('name') or s.get('service')} {s.get('product') or ''} {version}".strip()) + return "\n".join(out) + def _sheet_action_plan(wb, audit): ws = wb.create_sheet("Plan Action") rows = [] @@ -259,7 +354,7 @@ def _sheet_limits(wb, audit): def _sheet_raw(wb, audit): - ws = wb.create_sheet("Resume JSON") + ws = wb.create_sheet("Raw JSON Summary") rows = [ ["id", audit.get("id")], ["domain", audit.get("domain")], @@ -267,6 +362,7 @@ def _sheet_raw(wb, audit): ["mode", audit.get("mode")], ["score", json.dumps(audit.get("score", {}), ensure_ascii=False, default=str)], ["executive_risk", json.dumps(audit.get("executive_risk", {}), ensure_ascii=False, default=str)[:32000]], + ["dnsdumpster_like", json.dumps(audit.get("dnsdumpster_like", {}), ensure_ascii=False, default=str)[:32000]], ] _table_sheet(ws, "RÉSUMÉ JSON", ["Clé", "Valeur"], rows) diff --git a/backend/app/reports/html_report.py b/backend/app/reports/html_report.py index 351507a..f273db0 100644 --- a/backend/app/reports/html_report.py +++ b/backend/app/reports/html_report.py @@ -370,9 +370,142 @@ def _render_graph(audit: dict) -> str: ) + +def _render_dnsdumpster_like(audit: dict) -> str: + dd = audit.get("dnsdumpster_like") or {} + note = '

Nmap est une source complémentaire. Les services HTTP/HTTPS peuvent être détectés par fingerprint web même lorsque Nmap ne remonte pas de service.

' + metrics = ('
' + _metric("System Locations", len(dd.get("system_locations", {}) or {})) + _metric("Hosting / Networks", len(dd.get("hosting_networks", []) or [])) + _metric("Services / Banners", len(dd.get("services_banners", {}) or {})) + _metric("A Records enrichis", len(dd.get("a_records", []) or [])) + _metric("MX Records enrichis", len(dd.get("mx_records", []) or [])) + _metric("NS Records enrichis", len(dd.get("ns_records", []) or [])) + '
') + return note + metrics + + +def _render_system_locations(audit: dict) -> str: + rows = [[country, count] for country, count in (safe_get(audit, "dnsdumpster_like.system_locations", {}) or {}).items()] + return render_table(["Country", "Count"], rows or [["Unknown", 0]]) + + +def _render_hosting_networks(audit: dict) -> str: + rows = [[h.get("asn"), h.get("network"), h.get("asn_name"), h.get("country"), h.get("count")] for h in safe_get(audit, "dnsdumpster_like.hosting_networks", []) or []] + return render_table(["ASN", "Network", "ASN Name", "Country", "Hosts"], rows) + + +def _render_services_banners(audit: dict) -> str: + rows = [[banner, count] for banner, count in (safe_get(audit, "dnsdumpster_like.services_banners", {}) or {}).items()] + return render_table(["Banner", "Count"], rows) + + +def _render_a_records_filterable(audit: dict) -> str: + records = safe_get(audit, "dnsdumpster_like.a_records", []) or [] + controls = """
+ + + + + + + +
""" + headers = ["Host", "IP", "ASN", "Network", "ASN Name", "Country", "Open Services observed", "RevIP", "Sources"] + rows = [] + for h in records: + tech = _tech_text(h) + services = _open_services_html(h) + badges = _host_badges(h) + detail = _host_detail_html(h) + attrs = {"host": h.get("host", ""), "asn": h.get("asn", ""), "country": h.get("country", ""), "tech": tech, "service": _plain_services(h), "webnonmap": str(bool(h.get("open_services") and not h.get("nmap_services"))).lower(), "guard": str(bool(h.get("guardrail"))).lower()} + attr_str = " ".join(f'data-{k}="{esc(v).lower()}"' for k, v in attrs.items()) + host_cell = SafeHtml(f'
{esc(h.get("host"))} {badges}{detail}
') + rows.append(SafeHtml(f'{host_cell}{esc(h.get("ip"))}{esc(h.get("asn"))}{esc(h.get("network"))}{esc(h.get("asn_name"))}{esc(h.get("country"))}{services}{esc(h.get("revip_count", 0))}{esc(", ".join(h.get("sources", []) or []))}')) + if not rows: + return controls + '

Aucun A record enrichi disponible.

' + head = "".join(f"{esc(h)}" for h in headers) + return controls + f'
{head}{"".join(rows)}
' + + +def _render_mx_records(audit: dict) -> str: + rows = [] + for h in safe_get(audit, "dnsdumpster_like.mx_records", []) or []: + rows.append([h.get("priority"), h.get("host"), h.get("ip"), h.get("asn"), h.get("network"), h.get("asn_name"), h.get("country"), SafeHtml(_open_services_html(h)), _tech_text(h), _join(h.get("sources", []))]) + return render_table(["Priority", "Host", "IP", "ASN", "Network", "ASN Name", "Country", "Open Services observed", "Technologies", "Sources"], rows) + + +def _render_ns_records(audit: dict) -> str: + rows = [] + for h in safe_get(audit, "dnsdumpster_like.ns_records", []) or []: + rows.append([h.get("host"), h.get("ip"), h.get("asn"), h.get("network"), h.get("asn_name"), h.get("country"), SafeHtml(_open_services_html(h)), _join(h.get("sources", []))]) + return render_table(["Host", "IP", "ASN", "Network", "ASN Name", "Country", "Open Services observed", "Sources"], rows) + + +def _render_txt_records(audit: dict) -> str: + rows = [] + for item in safe_get(audit, "dnsdumpster_like.txt_records", []) or []: + spf = item.get("spf") or {} + rows.append([item.get("type"), SafeHtml(f'
{esc(item.get("value"))}
'), _join(spf.get("providers", []), 10), _join(spf.get("includes", []), 15), _join((spf.get("ip4", []) or []) + (spf.get("ip6", []) or []), 20)]) + return render_table(["Type", "Value", "SPF providers", "SPF includes", "SPF IPs"], rows) + + +def _render_host_inventory_detail(audit: dict) -> str: + cards = [] + for h in safe_get(audit, "dnsdumpster_like.hosts", []) or []: + cards.append(f"
{esc(h.get('host'))} - {esc(h.get('ip'))} {_host_badges(h)}{_host_detail_html(h)}
") + return "".join(cards) or '

Aucun host inventory détaillé disponible.

' + + +def _host_badges(h: dict) -> str: + badges = [] + if any(s.get("scheme") == "http" for s in h.get("open_services", []) or []): badges.append("HTTP") + if any(s.get("scheme") == "https" for s in h.get("open_services", []) or []): badges.append("HTTPS") + if (h.get("tls") or {}).get("cn") and (h.get("tls") or {}).get("cn") != "Non détecté": badges.append("TLS") + if h.get("nmap_services"): badges.append("Nmap") + if h.get("asn") and h.get("asn") != "Non détecté": badges.append("ASN") + if h.get("technologies"): badges.append("Tech") + if h.get("guardrail"): badges.append("Bloqué par garde-fou") + if not h.get("open_services"): badges.append("Non détecté") + return " ".join(f'{esc(b)}' for b in badges) + + +def _open_services_html(h: dict) -> str: + services = h.get("open_services", []) or [] + if not services: + return 'Bloqué par garde-fou' if h.get("guardrail") else 'Non détecté' + parts = [] + for svc in services: + text = f"{svc.get('scheme')}: {svc.get('banner') or svc.get('service') or 'unknown server'}" + if svc.get("title") and svc.get("title") != "Non détecté": text += f"; title: {svc.get('title')}" + parts.append(esc(text)) + tls = h.get("tls") or {} + if tls.get("cn") and tls.get("cn") != "Non détecté": parts.append(f"cn: {esc(tls.get('cn'))}") + tech = _tech_text(h) + if tech: parts.append(f"tech: {esc(tech)}") + return "
".join(parts) + + +def _plain_services(h: dict) -> str: + return " ".join([str(s.get("banner") or s.get("service") or "") for s in h.get("open_services", []) or []]) + + +def _tech_text(h: dict) -> str: + out = [] + for t in h.get("technologies", []) or []: + name = t.get("name") or "" + if t.get("version"): name += f":{t.get('version')}" + if t.get("note"): name += f" ({t.get('note')})" + out.append(name) + return ", ".join(dict.fromkeys(out)) + + +def _host_detail_html(h: dict) -> str: + tls = h.get("tls") or {} + guard = h.get("guardrail") or {} + data = [["DNS", h.get("host")], ["IP", h.get("ip")], ["ASN", h.get("asn")], ["Network", h.get("network")], ["ASN Name", h.get("asn_name")], ["Country", h.get("country")], ["Provider", h.get("provider")], ["HTTP", h.get("http", {}).get("status_code") or "Non détecté"], ["HTTPS", h.get("https", {}).get("status_code") or "Non détecté"], ["TLS", f"CN={tls.get('cn', 'Non détecté')} SAN={', '.join(tls.get('san', []) or [])} issuer={tls.get('issuer', 'Non détecté')} exp={tls.get('expires_at', 'Non détecté')}"], ["Technologies", _tech_text(h) or "Non détecté"], ["Nmap", _nmap_services_text(h) or "Non détecté par Nmap"], ["Sources", ", ".join(h.get("sources", []) or [])], ["Erreurs / garde-fous", guard.get("reason") or _join(h.get("resolution_errors", []))]] + return render_table(["Champ", "Valeur"], data, max_rows=80) + + +def _nmap_services_text(h: dict) -> str: + return "; ".join(f"{s.get('port')}/{s.get('protocol', 'tcp')} {s.get('name') or s.get('service') or ''} {s.get('product') or ''} {s.get('version') or 'Version non exposée'}".strip() for s in h.get("nmap_services", []) or []) + def _css() -> str: return """ -:root{--bg:#07070a;--bg2:#09090d;--card:#101018;--card2:#151521;--red:#ff3045;--red2:#dc143c;--darkred:#65000b;--text:#f5f5f5;--muted:#a1a1aa;--border:rgba(255,48,69,.25);--green:#34d399;--orange:#fb923c}*{box-sizing:border-box}html{scroll-behavior:smooth}body{margin:0;background:radial-gradient(circle at 70% 0,rgba(220,20,60,.22),transparent 32%),linear-gradient(135deg,var(--bg),var(--bg2));color:var(--text);font-family:Inter,ui-sans-serif,system-ui,-apple-system,Segoe UI,Arial,sans-serif;line-height:1.55}.watermark{position:fixed;inset:auto -4vw 8vh auto;font-size:13vw;font-weight:900;letter-spacing:.08em;color:rgba(255,255,255,.025);transform:rotate(-10deg);pointer-events:none;z-index:0}.hero{position:relative;z-index:1;min-height:72vh;padding:72px 7vw 52px;border-bottom:1px solid var(--border);background:linear-gradient(120deg,rgba(255,48,69,.12),transparent 55%)}.hero:before{content:"";position:absolute;inset:24px;border:1px solid rgba(255,48,69,.18);border-radius:32px;pointer-events:none}.eyebrow{color:var(--red);text-transform:uppercase;letter-spacing:.28em;font-size:.78rem;font-weight:800}.hero h1{margin:.1em 0;font-size:clamp(3.6rem,9vw,8.8rem);line-height:.88;letter-spacing:-.08em}.hero h2{margin:0;color:#fff;font-size:clamp(1.15rem,2vw,2rem);font-weight:500}.hero-meta{display:flex;flex-wrap:wrap;gap:12px;margin:28px 0}.badge{display:inline-flex;align-items:center;border:1px solid var(--border);border-radius:999px;padding:5px 10px;background:rgba(255,48,69,.09);color:#fff;font-weight:800;font-size:.76rem;text-transform:uppercase;letter-spacing:.06em}.layout{position:relative;z-index:1;display:grid;grid-template-columns:280px minmax(0,1fr);gap:28px;padding:36px 5vw}.toc{position:sticky;top:20px;align-self:start;background:rgba(16,16,24,.86);backdrop-filter:blur(10px);border:1px solid var(--border);border-radius:20px;padding:18px;max-height:calc(100vh - 40px);overflow:auto}.toc strong{display:block;margin-bottom:12px;color:#fff}.toc a{display:block;color:var(--muted);text-decoration:none;padding:8px 10px;border-radius:10px;font-size:.92rem}.toc a:hover{background:rgba(255,48,69,.12);color:#fff}.content{min-width:0}.metric-grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(180px,1fr));gap:14px;margin-top:28px}.metric-card,.cardlet{background:linear-gradient(180deg,var(--card),rgba(16,16,24,.76));border:1px solid var(--border);border-radius:18px;padding:18px;box-shadow:0 18px 50px rgba(0,0,0,.24)}.metric-card span{display:block;color:var(--muted);font-size:.82rem;text-transform:uppercase;letter-spacing:.08em}.metric-card strong{display:block;margin-top:7px;font-size:1.7rem;color:#fff}.metric-card small{color:var(--muted)}.section{margin:0 0 28px;padding:26px;background:rgba(16,16,24,.78);border:1px solid var(--border);border-radius:24px;box-shadow:0 20px 70px rgba(0,0,0,.26)}.section-title{display:flex;align-items:center;gap:14px;border-bottom:1px solid rgba(255,48,69,.18);padding-bottom:14px;margin-bottom:18px}.section-title span{color:var(--red);font-weight:900}.section h2{margin:0;color:#fff;font-size:1.55rem}.section-subtitle,.notice{color:var(--muted)}.lead{font-size:1.06rem;color:#e7e7e9}.two-col{display:grid;grid-template-columns:repeat(auto-fit,minmax(240px,1fr));gap:16px}.cardlet h3{margin-top:0;color:var(--red)}.table-scroll{overflow:auto;border:1px solid rgba(255,48,69,.16);border-radius:16px}.table{width:100%;border-collapse:collapse;min-width:760px;background:rgba(7,7,10,.55)}.table th,.table td{padding:11px 12px;border-bottom:1px solid rgba(255,255,255,.07);text-align:left;vertical-align:top}.table th{color:#fff;background:rgba(101,0,11,.5);font-size:.78rem;text-transform:uppercase;letter-spacing:.08em}.table td{color:#d8d8dc;font-size:.9rem}.finding{border:1px solid rgba(255,48,69,.22);border-radius:18px;padding:18px;margin:14px 0;background:linear-gradient(135deg,rgba(255,48,69,.05),rgba(255,255,255,.02))}.finding-head{display:flex;justify-content:space-between;gap:12px;align-items:start}.finding h3{margin:0;color:#fff}.finding-meta{display:grid;grid-template-columns:140px 1fr;gap:8px 12px}.finding-meta dt{color:var(--red);font-weight:800}.finding-meta dd{margin:0;color:#dcdce1}.severity-critical{background:rgba(101,0,11,.88);border-color:rgba(255,48,69,.75)}.severity-high{background:rgba(220,20,60,.55)}.severity-medium{background:rgba(251,146,60,.28);border-color:rgba(251,146,60,.45)}.severity-low{background:rgba(161,161,170,.18)}.severity-info{background:rgba(59,130,246,.16);border-color:rgba(59,130,246,.35)}.progress-wrap{margin:14px 0}.progress-label{display:flex;justify-content:space-between;color:#fff;font-weight:800}.progress{height:12px;border:1px solid var(--border);border-radius:999px;background:#09090d;overflow:hidden}.progress span{display:block;height:100%;background:linear-gradient(90deg,var(--darkred),var(--red));border-radius:999px}.code{font-family:ui-monospace,SFMono-Regular,Menlo,Consolas,monospace;background:#08080c;border:1px solid rgba(255,255,255,.1);border-radius:10px;padding:8px;color:#f1f1f3;white-space:pre-wrap;word-break:break-word}@media(max-width:880px){.layout{grid-template-columns:1fr}.toc{position:relative;top:0}.hero{padding:48px 6vw}.section{padding:20px}}@media print{body{background:#09090d}.toc{position:relative}.layout{display:block}.section{break-inside:avoid}} +:root{--bg:#07070a;--bg2:#09090d;--card:#101018;--card2:#151521;--red:#ff3045;--red2:#dc143c;--darkred:#65000b;--text:#f5f5f5;--muted:#a1a1aa;--border:rgba(255,48,69,.25);--green:#34d399;--orange:#fb923c}*{box-sizing:border-box}html{scroll-behavior:smooth}body{margin:0;background:radial-gradient(circle at 70% 0,rgba(220,20,60,.22),transparent 32%),linear-gradient(135deg,var(--bg),var(--bg2));color:var(--text);font-family:Inter,ui-sans-serif,system-ui,-apple-system,Segoe UI,Arial,sans-serif;line-height:1.55}.watermark{position:fixed;inset:auto -4vw 8vh auto;font-size:13vw;font-weight:900;letter-spacing:.08em;color:rgba(255,255,255,.025);transform:rotate(-10deg);pointer-events:none;z-index:0}.hero{position:relative;z-index:1;min-height:72vh;padding:72px 7vw 52px;border-bottom:1px solid var(--border);background:linear-gradient(120deg,rgba(255,48,69,.12),transparent 55%)}.hero:before{content:"";position:absolute;inset:24px;border:1px solid rgba(255,48,69,.18);border-radius:32px;pointer-events:none}.eyebrow{color:var(--red);text-transform:uppercase;letter-spacing:.28em;font-size:.78rem;font-weight:800}.hero h1{margin:.1em 0;font-size:clamp(3.6rem,9vw,8.8rem);line-height:.88;letter-spacing:-.08em}.hero h2{margin:0;color:#fff;font-size:clamp(1.15rem,2vw,2rem);font-weight:500}.hero-meta{display:flex;flex-wrap:wrap;gap:12px;margin:28px 0}.badge{display:inline-flex;align-items:center;border:1px solid var(--border);border-radius:999px;padding:5px 10px;background:rgba(255,48,69,.09);color:#fff;font-weight:800;font-size:.76rem;text-transform:uppercase;letter-spacing:.06em}.layout{position:relative;z-index:1;display:grid;grid-template-columns:280px minmax(0,1fr);gap:28px;padding:36px 5vw}.toc{position:sticky;top:20px;align-self:start;background:rgba(16,16,24,.86);backdrop-filter:blur(10px);border:1px solid var(--border);border-radius:20px;padding:18px;max-height:calc(100vh - 40px);overflow:auto}.toc strong{display:block;margin-bottom:12px;color:#fff}.toc a{display:block;color:var(--muted);text-decoration:none;padding:8px 10px;border-radius:10px;font-size:.92rem}.toc a:hover{background:rgba(255,48,69,.12);color:#fff}.content{min-width:0}.metric-grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(180px,1fr));gap:14px;margin-top:28px}.metric-card,.cardlet{background:linear-gradient(180deg,var(--card),rgba(16,16,24,.76));border:1px solid var(--border);border-radius:18px;padding:18px;box-shadow:0 18px 50px rgba(0,0,0,.24)}.metric-card span{display:block;color:var(--muted);font-size:.82rem;text-transform:uppercase;letter-spacing:.08em}.metric-card strong{display:block;margin-top:7px;font-size:1.7rem;color:#fff}.metric-card small{color:var(--muted)}.section{margin:0 0 28px;padding:26px;background:rgba(16,16,24,.78);border:1px solid var(--border);border-radius:24px;box-shadow:0 20px 70px rgba(0,0,0,.26)}.section-title{display:flex;align-items:center;gap:14px;border-bottom:1px solid rgba(255,48,69,.18);padding-bottom:14px;margin-bottom:18px}.section-title span{color:var(--red);font-weight:900}.section h2{margin:0;color:#fff;font-size:1.55rem}.section-subtitle,.notice{color:var(--muted)}.lead{font-size:1.06rem;color:#e7e7e9}.two-col{display:grid;grid-template-columns:repeat(auto-fit,minmax(240px,1fr));gap:16px}.cardlet h3{margin-top:0;color:var(--red)}.table-scroll{overflow:auto;border:1px solid rgba(255,48,69,.16);border-radius:16px}.table{width:100%;border-collapse:collapse;min-width:760px;background:rgba(7,7,10,.55)}.table th,.table td{padding:11px 12px;border-bottom:1px solid rgba(255,255,255,.07);text-align:left;vertical-align:top}.table th{color:#fff;background:rgba(101,0,11,.5);font-size:.78rem;text-transform:uppercase;letter-spacing:.08em}.table td{color:#d8d8dc;font-size:.9rem}.finding{border:1px solid rgba(255,48,69,.22);border-radius:18px;padding:18px;margin:14px 0;background:linear-gradient(135deg,rgba(255,48,69,.05),rgba(255,255,255,.02))}.finding-head{display:flex;justify-content:space-between;gap:12px;align-items:start}.finding h3{margin:0;color:#fff}.finding-meta{display:grid;grid-template-columns:140px 1fr;gap:8px 12px}.finding-meta dt{color:var(--red);font-weight:800}.finding-meta dd{margin:0;color:#dcdce1}.severity-critical{background:rgba(101,0,11,.88);border-color:rgba(255,48,69,.75)}.severity-high{background:rgba(220,20,60,.55)}.severity-medium{background:rgba(251,146,60,.28);border-color:rgba(251,146,60,.45)}.severity-low{background:rgba(161,161,170,.18)}.severity-info{background:rgba(59,130,246,.16);border-color:rgba(59,130,246,.35)}.progress-wrap{margin:14px 0}.progress-label{display:flex;justify-content:space-between;color:#fff;font-weight:800}.progress{height:12px;border:1px solid var(--border);border-radius:999px;background:#09090d;overflow:hidden}.progress span{display:block;height:100%;background:linear-gradient(90deg,var(--darkred),var(--red));border-radius:999px}.code{font-family:ui-monospace,SFMono-Regular,Menlo,Consolas,monospace;background:#08080c;border:1px solid rgba(255,255,255,.1);border-radius:10px;padding:8px;color:#f1f1f3;white-space:pre-wrap;word-break:break-word}.filters{display:grid;grid-template-columns:repeat(auto-fit,minmax(190px,1fr));gap:10px;margin:14px 0}.filters input{background:#08080c;color:#fff;border:1px solid var(--border);border-radius:12px;padding:10px}.filters label{color:#d8d8dc;font-size:.9rem}.badge.mini{font-size:.64rem;padding:3px 7px;margin:2px}.host-card{border:1px solid rgba(255,48,69,.2);border-radius:16px;margin:12px 0;padding:12px;background:rgba(7,7,10,.35)}details summary{cursor:pointer;color:#fff;font-weight:800}@media(max-width:880px){.layout{grid-template-columns:1fr}.toc{position:relative;top:0}.hero{padding:48px 6vw}.section{padding:20px}}@media print{body{background:#09090d}.toc{position:relative}.layout{display:block}.section{break-inside:avoid}} """ @@ -386,11 +519,11 @@ def generate_html_report(audit: dict) -> str: subdomains = safe_get(audit, "subdomains.subdomains", []) cve_count = (safe_get(audit, "passive_cves.count", 0) or 0) + (safe_get(audit, "service_scan.count_cves", 0) or 0) toc = [ - ("synthese", "Synthèse exécutive"), ("scoring", "Scoring"), ("plan", "Plan d’action priorisé"), - ("findings", "Findings"), ("dns", "DNS"), ("messagerie", "Messagerie"), ("web", "Web / Headers"), - ("tls", "TLS / SSL"), ("subdomains", "Sous-domaines"), ("ips", "IP publiques"), - ("nmap", "Nmap service/version/port"), ("cve", "CVE"), ("cti", "CTI / réputation"), - ("graph", "Graph Explorer"), ("limites", "Limites et responsabilité"), + ("synthese", "Synthèse"), ("system_locations", "System Locations"), ("hosting_networks", "Hosting / Networks"), + ("services_banners", "Services / Banners"), ("a_records", "A Records / Subdomains"), + ("mx_records", "MX Records"), ("ns_records", "NS Records"), ("txt_records", "TXT Records"), + ("host_inventory", "Host Inventory détaillé"), ("nmap", "Nmap en annexe"), ("cve", "CVE"), + ("limites", "Limites et sources"), ] hero_metrics = ( '
' @@ -416,21 +549,36 @@ def generate_html_report(audit: dict) -> str:

OpenEASM Beta

RAPPORT OPENEASM

External Attack Surface Management · Rapport d’exposition externe

Domaine: {esc(domain)}Généré: {esc(now)}NON EXPLOITANTService/version/CVE défensif public
{hero_metrics}
-{section('synthese','Synthèse exécutive',_build_executive_summary(audit))} -{section('scoring','Scoring',_render_scoring(audit),'Scores consolidés et jauges visuelles CSS.')} -{section('plan','Plan d’action priorisé',_render_action_plan(audit))} -{section('findings','Findings détaillés',_render_findings(audit))} -{section('dns','DNS',_render_dns(audit))} -{section('messagerie','Messagerie',_render_mail(audit))} -{section('web','Web / Headers',_render_web(audit))} -{section('tls','TLS / SSL',_render_tls(audit))} -{section('subdomains','Sous-domaines',_render_subdomains(audit))} -{section('ips','IP publiques',_render_ips(audit))} -{section('nmap','Nmap service/version/port',_render_nmap(audit),'Rappel : collecte non exploitante, sans scripts intrusifs.')} +{section('synthese','Synthèse',_build_executive_summary(audit) + _render_dnsdumpster_like(audit))} +{section('system_locations','System Locations',_render_system_locations(audit))} +{section('hosting_networks','Hosting / Networks',_render_hosting_networks(audit))} +{section('services_banners','Services / Banners',_render_services_banners(audit))} +{section('a_records','A Records / Subdomains from dataset',_render_a_records_filterable(audit),'Table filtrable : host, ASN, pays, technologie, service, HTTP/HTTPS sans Nmap, garde-fou.')} +{section('mx_records','MX Records',_render_mx_records(audit))} +{section('ns_records','NS Records',_render_ns_records(audit))} +{section('txt_records','TXT Records',_render_txt_records(audit))} +{section('host_inventory','Host Inventory détaillé',_render_host_inventory_detail(audit))} +{section('nmap','Nmap en annexe',_render_nmap(audit),'Rappel : collecte non exploitante, sans scripts intrusifs.')} {section('cve','CVE',_render_cves(audit))} -{section('cti','CTI / réputation',_render_cti(audit))} -{section('graph','Graph Explorer',_render_graph(audit))} -{section('limites','Limites et responsabilité',limits)} -
""" +{section('limites','Limites et sources',limits)} +
""" path.write_text(html, encoding="utf-8") return filename diff --git a/backend/app/reports/pdf_report.py b/backend/app/reports/pdf_report.py index bc2580d..a8603fb 100644 --- a/backend/app/reports/pdf_report.py +++ b/backend/app/reports/pdf_report.py @@ -96,12 +96,7 @@ def draw(self): def generate_pdf_report(audit: dict) -> str: - """Generate the professional OpenEASM PDF report. - - Beta fix: large subsections such as subdomains and IP inventory are not nested - inside another table anymore. ReportLab cannot split a large nested table across - pages, which caused the PDF export to fail when many subdomains were present. - """ + """Generate a DNSDumpster-like PDF where Nmap is an appendix.""" filename = f"open_easm_beta_{audit['domain'].replace('.', '_')}_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.pdf" path = REPORT_DIR / filename @@ -114,82 +109,66 @@ def generate_pdf_report(audit: dict) -> str: bottomMargin=1.05 * cm, title=f"OpenEASM Beta - {audit['domain']}", author="OpenEASM", - subject="External Attack Surface Management defensive audit", + subject="DNSDumpster-like External Attack Surface Management defensive audit", ) styles = _styles() story = [] + dd = audit.get("dnsdumpster_like", {}) or {} risk = audit.get("executive_risk", {}) or {} - score = audit.get("score", {}) or {} scan = audit.get("service_scan", {}) or {} - graph = audit.get("attack_graph", {}) or {} story.append(_cover_block(audit, styles)) - story.append(Spacer(1, 0.30 * cm)) + story.append(Spacer(1, 0.20 * cm)) story.append(_kpi_cards(audit, styles)) story.append(Spacer(1, 0.18 * cm)) - story.append(_score_panel(audit, styles)) - story.append(Spacer(1, 0.25 * cm)) - - story.append(Paragraph("Synthèse exécutive", styles["Section"])) + story.append(Paragraph("Page de synthèse", styles["Section"])) story.append(_callout(str(risk.get("board_summary") or _conclusion(audit)), styles, tone="neutral")) - story.append(Spacer(1, 0.20 * cm)) - story.append(Paragraph("Plan d'action priorisé", styles["Section"])) - story.append(_actions_table(audit, styles, limit=12)) + story.append(Spacer(1, 0.14 * cm)) + story.append(_callout("Nmap est une source complémentaire. Les services HTTP/HTTPS peuvent être détectés par fingerprint web même lorsque Nmap ne remonte pas de service.", styles, tone="safe")) + story.append(Spacer(1, 0.18 * cm)) + story.append(_dnsdumpster_summary_table(dd, styles)) story.append(PageBreak()) - story.append(Paragraph("Vue de risque par pilier", styles["Section"])) - story.append(_risk_overview(risk, styles)) + story.append(Paragraph("System Locations", styles["Section"])) + story.append(_system_locations_table(dd, styles)) + story.append(Spacer(1, 0.22 * cm)) + story.append(Paragraph("Hosting / Networks", styles["Section"])) + story.append(_hosting_networks_table(dd, styles)) story.append(Spacer(1, 0.22 * cm)) - story.append(SeverityLegend(score.get("by_severity", {}))) - story.append(Spacer(1, 0.28 * cm)) - story.append(Paragraph("Constats prioritaires localisés", styles["Section"])) - story.append(_findings_table(audit, styles, limit=26)) + story.append(Paragraph("Services / Banners", styles["Section"])) + story.append(_services_banners_table(dd, styles)) story.append(PageBreak()) - story.append(Paragraph("Cartographie de l'exposition", styles["Section"])) - graph_rows = [ - ["Indicateur", "Valeur", "Lecture"], - ["Noeuds Graph Explorer", str((graph.get("metrics") or {}).get("nodes", 0)), "Domaines, sous-domaines, IP, services, CVE et constats"], - ["Relations", str((graph.get("metrics") or {}).get("edges", 0)), "Liens DNS, exposition web, ports, CVE et constats"], - ["IP publiques", str(audit.get("ip_inventory", {}).get("public_ip_count", 0)), "Surface réseau publique observée"], - ["Sous-domaines", str(audit.get("subdomains", {}).get("count", 0)), "Découverte passive"], - ["Ports Nmap", str(scan.get("count_open_ports", 0)), "Service/version/port, non exploitant"], - ] - story.append(_table(graph_rows, [5.2 * cm, 3.7 * cm, 16.6 * cm], styles=styles)) - story.append(Spacer(1, 0.25 * cm)) - - # Important: do not nest the subdomain and IP tables in a single parent table. - # Nested tables are treated as one unbreakable flowable by ReportLab. If the - # list is long, the PDF generation fails with "Flowable ... too large". - story.append(Paragraph("Sous-domaines publics", styles["SectionSmall"])) - story.append(_subdomains_table(audit, styles, limit=80)) - story.append(Spacer(1, 0.25 * cm)) - story.append(Paragraph("Inventaire IP", styles["SectionSmall"])) - story.append(_ip_table(audit, styles, limit=80)) + story.append(Paragraph("A Records / Subdomains from dataset", styles["Section"])) + story.append(_dnsdumpster_hosts_table(dd.get("a_records", []), styles, limit=70)) + + story.append(PageBreak()) + story.append(Paragraph("MX Records", styles["Section"])) + story.append(_dnsdumpster_hosts_table(dd.get("mx_records", []), styles, limit=60, mx=True)) + story.append(Spacer(1, 0.22 * cm)) + story.append(Paragraph("NS Records", styles["Section"])) + story.append(_dnsdumpster_hosts_table(dd.get("ns_records", []), styles, limit=60)) + story.append(Spacer(1, 0.22 * cm)) + story.append(Paragraph("TXT Records", styles["Section"])) + story.append(_txt_records_table(dd, styles)) + + story.append(PageBreak()) + story.append(Paragraph("Host Inventory détaillé", styles["Section"])) + story.append(_host_inventory_detail(dd, styles, limit=40)) story.append(PageBreak()) - story.append(Paragraph("Nmap service / version / CVE", styles["Section"])) - story.append(_callout( - "Contrôle non exploitant : identification des ports ouverts, services et versions. La corrélation CVE est effectuée côté OpenEASM à partir des versions détectées. Aucun exploit, bruteforce, DoS ou script intrusif n'est exécuté.", - styles, - tone="safe", - )) + story.append(Paragraph("Annexe — Nmap service/version", styles["Section"])) + story.append(_callout(str(scan.get("note") or "Nmap service/version/port uniquement, sans exploitation."), styles, tone="safe")) story.append(Spacer(1, 0.16 * cm)) story.append(_nmap_table(audit, styles, limit=80)) story.append(PageBreak()) - story.append(Paragraph("Annexes : portée, limites et responsabilité", styles["Section"])) - limits = [ - ["Point", "Détail"], - ["Nature de l'audit", "Audit public défensif d'exposition externe. Les résultats sont issus d'informations visibles publiquement."], - ["Nmap", str(scan.get("note", "Service/version/port uniquement, sans exploitation."))], - ["CVE", "Une CVE n'est affichée que si la version détectée permet une corrélation raisonnable. Une version masquée ne doit pas générer de faux positif."], - ["Versions masquées", "OpenEASM indique 'version non exposée' et recommande une vérification interne via inventaire serveur, EDR, gestion de parc ou paquet système."], - ["Backports", "Les distributions Linux peuvent intégrer des correctifs de sécurité sans changer le numéro de version amont."], - ["Responsabilité", "L'utilisateur doit disposer d'un droit, d'une autorisation explicite ou d'un motif légitime de sécurité informatique."], - ] - story.append(_table(limits, [5.2 * cm, 20.3 * cm], styles=styles)) + story.append(Paragraph("Annexe — CVE", styles["Section"])) + story.append(_cve_table(audit, styles, limit=80)) + story.append(Spacer(1, 0.22 * cm)) + story.append(Paragraph("Limites et sources", styles["Section"])) + story.append(_limits_sources_table(audit, dd, styles)) doc.build(story, onFirstPage=_decorate_page, onLaterPages=_decorate_page) return filename @@ -359,6 +338,149 @@ def _nmap_table(audit: dict, styles, limit: int = 80): return _table(rows, [4.4 * cm, 2.2 * cm, 2.9 * cm, 4.2 * cm, 3.5 * cm, 3.4 * cm, 2.5 * cm], small=True, styles=styles) + +def _dnsdumpster_summary_table(dd: dict, styles): + rows = [["Section", "Volume"]] + rows.extend([ + ["System Locations", str(len(dd.get("system_locations", {}) or {}))], + ["Hosting / Networks", str(len(dd.get("hosting_networks", []) or []))], + ["Services / Banners", str(len(dd.get("services_banners", {}) or {}))], + ["A Records enrichis", str(len(dd.get("a_records", []) or []))], + ["MX Records enrichis", str(len(dd.get("mx_records", []) or []))], + ["NS Records enrichis", str(len(dd.get("ns_records", []) or []))], + ["TXT Records", str(len(dd.get("txt_records", []) or []))], + ]) + return _table(rows, [15.0 * cm, 5.0 * cm], styles=styles) + + +def _system_locations_table(dd: dict, styles): + rows = [["Country", "Count"]] + for country, count in (dd.get("system_locations", {}) or {}).items(): + rows.append([country, str(count)]) + if len(rows) == 1: + rows.append(["Unknown", "0"]) + return _table(rows, [16.0 * cm, 4.0 * cm], styles=styles) + + +def _hosting_networks_table(dd: dict, styles): + rows = [["ASN", "Network", "ASN Name", "Country", "Hosts"]] + for h in dd.get("hosting_networks", []) or []: + rows.append([h.get("asn"), h.get("network"), h.get("asn_name"), h.get("country"), h.get("count")]) + if len(rows) == 1: + rows.append(["Non détecté", "Non détecté", "Non détecté", "Unknown", "0"]) + return _table(rows, [3.0 * cm, 4.2 * cm, 11.8 * cm, 3.0 * cm, 2.0 * cm], small=True, styles=styles) + + +def _services_banners_table(dd: dict, styles): + rows = [["Banner", "Count"]] + for banner, count in (dd.get("services_banners", {}) or {}).items(): + rows.append([banner, str(count)]) + if len(rows) == 1: + rows.append(["Non détecté", "0"]) + return _table(rows, [17.0 * cm, 3.0 * cm], styles=styles) + + +def _dnsdumpster_hosts_table(records: list, styles, limit: int = 70, mx: bool = False): + headers = (["Priority"] if mx else []) + ["Host", "IP", "ASN", "Network", "ASN Name", "Country", "Open Services observed", "RevIP", "Sources"] + rows = [headers] + for h in (records or [])[:limit]: + base = [h.get("host"), h.get("ip"), h.get("asn"), h.get("network"), h.get("asn_name"), h.get("country"), _open_services_line(h), h.get("revip_count", 0), ", ".join(h.get("sources", []) or [])] + rows.append(([h.get("priority")] if mx else []) + base) + if len(records or []) > limit: + rows.append((["..."] if mx else []) + [f"{len(records)-limit} lignes supplémentaires", "Voir JSON/Excel", "", "", "", "", "", "", ""]) + if len(rows) == 1: + rows.append(([""] if mx else []) + ["Aucun", "Not found", "Non détecté", "Non détecté", "Non détecté", "Unknown", "Non détecté", "0", ""]) + widths = ([1.8 * cm] if mx else []) + [4.2 * cm, 2.8 * cm, 2.4 * cm, 3.4 * cm, 5.2 * cm, 2.5 * cm, 7.4 * cm, 1.7 * cm, 2.7 * cm] + scale = 25.5 * cm / sum(widths) + widths = [w * scale for w in widths] + return _table(rows, widths, small=True, styles=styles) + + +def _txt_records_table(dd: dict, styles): + rows = [["Type", "Value", "SPF providers", "SPF includes/ip"]] + for item in dd.get("txt_records", []) or []: + spf = item.get("spf") or {} + rows.append([item.get("type"), item.get("value"), ", ".join(spf.get("providers", []) or []), "; ".join((spf.get("includes", []) or []) + (spf.get("ip4", []) or []) + (spf.get("ip6", []) or []))]) + if len(rows) == 1: + rows.append(["TXT", "Non détecté", "", ""]) + return _table(rows, [2.1 * cm, 14.4 * cm, 4.0 * cm, 5.0 * cm], small=True, styles=styles) + + +def _host_inventory_detail(dd: dict, styles, limit: int = 40): + rows = [["Host", "IP / ASN", "HTTP/HTTPS/TLS", "Technologies", "Nmap"]] + for h in (dd.get("hosts", []) or [])[:limit]: + tls = h.get("tls") or {} + tls_txt = f"cn: {tls.get('cn', 'Non détecté')}; issuer: {tls.get('issuer', 'Non détecté')}; exp: {tls.get('expires_at', 'Non détecté')}" + rows.append([ + h.get("host"), + f"{h.get('ip')} | {h.get('asn')} | {h.get('network')} | {h.get('country')}", + f"{_open_services_line(h)}; {tls_txt}", + _tech_line(h), + _nmap_line(h) or "Non détecté par Nmap", + ]) + if len(rows) == 1: + rows.append(["Aucun", "", "", "", ""]) + return _table(rows, [4.3 * cm, 5.3 * cm, 7.4 * cm, 5.0 * cm, 3.5 * cm], small=True, styles=styles) + + +def _cve_table(audit: dict, styles, limit: int = 80): + cves = (audit.get("service_scan", {}) or {}).get("cves", []) or [] + rows = [["CVE", "Host", "Service", "Version", "Sévérité", "Preuve"]] + for c in cves[:limit]: + rows.append([c.get("cve"), c.get("hostname") or c.get("host"), c.get("service") or c.get("product"), c.get("version") or "Version non exposée", c.get("severity"), c.get("evidence")]) + if len(rows) == 1: + rows.append(["Aucune", "", "", "", "", "Aucune CVE corrélée par version exposée."]) + return _table(rows, [3.0 * cm, 4.0 * cm, 4.0 * cm, 3.5 * cm, 2.5 * cm, 8.5 * cm], small=True, styles=styles) + + +def _limits_sources_table(audit: dict, dd: dict, styles): + policy = dd.get("policy", {}) or {} + rows = [["Point", "Détail"]] + rows.extend([ + ["Sources", "DNS, sous-domaines passifs/fallback, ASN Team Cymru, HTTP/HTTPS fingerprint, TLS certificate, Nmap en annexe."], + ["HTTP", f"User-Agent: {policy.get('user_agent', 'OpenEASM-Beta-26.6 Defensive Audit')} ; timeout {policy.get('http_timeout_seconds', 8)} s ; concurrence {policy.get('concurrency', 10)}."], + ["Garde-fous", "Aucun exploit, aucun bruteforce, aucun DoS, aucun NSE intrusif ; cibles privées/réservées ou mixtes bloquées."], + ["Nmap", "Source complémentaire seulement. Une absence Nmap ne masque pas un service HTTP/HTTPS observé."], + ["Données passives", "Optionnelles via clés API ; aucune donnée inventée si non configuré."], + ["Non détecté", "Les informations absentes restent Non détecté / Version non exposée / Non détecté par Nmap."], + ]) + return _table(rows, [5.0 * cm, 20.5 * cm], styles=styles) + + +def _open_services_line(h: dict) -> str: + parts = [] + for svc in h.get("open_services", []) or []: + seg = f"{svc.get('scheme')}: {svc.get('banner') or svc.get('service') or 'unknown server'}" + if svc.get("title") and svc.get("title") != "Non détecté": + seg += f"; title: {svc.get('title')}" + parts.append(seg) + tls = h.get("tls") or {} + if tls.get("cn") and tls.get("cn") != "Non détecté": + parts.append(f"cn: {tls.get('cn')}") + tech = _tech_line(h) + if tech and tech != "Non détecté": + parts.append(f"tech: {tech}") + return "; ".join(parts) or ("Bloqué par garde-fou" if h.get("guardrail") else "Non détecté") + + +def _tech_line(h: dict) -> str: + out = [] + for t in h.get("technologies", []) or []: + name = t.get("name") or "" + if t.get("version"): + name += f":{t.get('version')}" + if t.get("note"): + name += f" ({t.get('note')})" + out.append(name) + return ", ".join(dict.fromkeys(out)) or "Non détecté" + + +def _nmap_line(h: dict) -> str: + parts = [] + for s in h.get("nmap_services", []) or []: + parts.append(f"{s.get('port')}/{s.get('protocol', 'tcp')} {s.get('name') or s.get('service') or ''} {s.get('product') or ''} {s.get('version') or 'Version non exposée'}".strip()) + return "; ".join(parts) + def _callout(text: str, styles, tone: str = "neutral"): bg = colors.HexColor("#ECFFF5") if tone == "safe" else CARD_ALT bar = GREEN if tone == "safe" else GOLD diff --git a/backend/app/services/dnsdumpster_like_inventory.py b/backend/app/services/dnsdumpster_like_inventory.py new file mode 100644 index 0000000..e848d8c --- /dev/null +++ b/backend/app/services/dnsdumpster_like_inventory.py @@ -0,0 +1,635 @@ +from __future__ import annotations + +import asyncio +import os +import re +import socket +import ssl +from collections import Counter, defaultdict +from datetime import datetime, timezone +from typing import Any + +import dns.resolver +import httpx + +from app.validators import is_public_ip + +UNKNOWN = "Non détecté" +UNKNOWN_COUNTRY = "Unknown" +USER_AGENT = "OpenEASM-Beta-26.6 Defensive Audit" +HTTP_TIMEOUT = float(os.getenv("OPENEASM_HTTP_TIMEOUT", "8") or 8) +TLS_TIMEOUT = float(os.getenv("OPENEASM_TLS_TIMEOUT", "8") or 8) +MAX_HOSTS = int(os.getenv("OPENEASM_HOST_INVENTORY_MAX_HOSTS", "200") or 200) +CONCURRENCY = int(os.getenv("OPENEASM_HOST_INVENTORY_CONCURRENCY", "10") or 10) +SCAN_ALL_SUBDOMAINS = os.getenv("OPENEASM_SCAN_ALL_SUBDOMAINS", "true").lower() in {"1", "true", "yes", "on"} +PASSIVE_DB_ENABLED = os.getenv("OPENEASM_PASSIVE_DB_ENABLED", "false").lower() in {"1", "true", "yes", "on"} + +COUNTRY_NAMES = { + "FR": "France", + "US": "United States", + "GB": "United Kingdom", + "DE": "Germany", + "ES": "Spain", + "IT": "Italy", + "NL": "Netherlands", + "BE": "Belgium", + "CH": "Switzerland", + "CA": "Canada", + "IE": "Ireland", + "LU": "Luxembourg", + "PT": "Portugal", +} + + +def env_enabled() -> bool: + return os.getenv("OPENEASM_DNSDUMPSTER_LIKE_ENABLED", "true").lower() in {"1", "true", "yes", "on"} + + +async def build_dnsdumpster_like_inventory( + domain: str, + dns_result: dict, + mail_result: dict, + subdomains_result: dict, + ip_inventory: dict, + service_scan: dict | None = None, +) -> dict: + """Build a DNSDumpster-like view from observed OpenEASM data. + + HTTP/HTTPS/TLS fingerprints are first-class sources. Nmap is merged as a + complementary source only; missing Nmap output never suppresses web evidence. + """ + if not env_enabled(): + return _empty(domain, enabled=False) + + a_hosts = _a_record_hosts(domain, dns_result, subdomains_result, ip_inventory) + mx_hosts = _mx_records(mail_result) + ns_hosts = _ns_records(dns_result) + txt_records = _txt_records(dns_result) + + all_for_revip = defaultdict(set) + for item in a_hosts: + for ip in item.get("ips", []): + if is_public_ip(ip): + all_for_revip[ip].add(item["host"]) + + sem = asyncio.Semaphore(max(1, CONCURRENCY)) + asn_cache: dict[str, dict] = {} + + async def enrich_host(item: dict, record_type: str = "a") -> dict: + async with sem: + return await _enrich_host(item, record_type, all_for_revip, asn_cache, service_scan or {}) + + if os.getenv("OPENEASM_HOST_INVENTORY_ENABLED", "true").lower() not in {"1", "true", "yes", "on"}: + hosts = [_host_skeleton(x, "a") for x in a_hosts[:MAX_HOSTS]] + mx_records = [_record_skeleton(x, "mx") for x in mx_hosts] + ns_records = [_record_skeleton(x, "ns") for x in ns_hosts] + else: + hosts = await asyncio.gather(*(enrich_host(item, "a") for item in a_hosts[:MAX_HOSTS])) + mx_records = await asyncio.gather(*(enrich_host(item, "mx") for item in mx_hosts[:MAX_HOSTS])) + ns_records = await asyncio.gather(*(enrich_host(item, "ns") for item in ns_hosts[:MAX_HOSTS])) + + system_locations = Counter(h.get("country") or UNKNOWN_COUNTRY for h in hosts if _public_ip(h.get("ip"))) + system_locations.update(h.get("country") or UNKNOWN_COUNTRY for h in mx_records + ns_records if _public_ip(h.get("ip"))) + + hosting_counter: dict[tuple[str, str, str, str], set[str]] = defaultdict(set) + for h in hosts + mx_records + ns_records: + if not _public_ip(h.get("ip")): + continue + key = ( + h.get("asn") or UNKNOWN, + h.get("network") or UNKNOWN, + h.get("asn_name") or UNKNOWN, + h.get("country") or UNKNOWN_COUNTRY, + ) + hosting_counter[key].add(h.get("host") or h.get("ip")) + hosting_networks = [ + {"asn": k[0], "network": k[1], "asn_name": k[2], "country": k[3], "count": len(v)} + for k, v in sorted(hosting_counter.items(), key=lambda kv: (-len(kv[1]), kv[0])) + ] + + services_banners = Counter() + for h in hosts + mx_records + ns_records: + for svc in h.get("open_services", []) or []: + banner = svc.get("banner") or svc.get("service") or "unknown server" + services_banners[banner or "unknown server"] += 1 + for nmap in h.get("nmap_services", []) or []: + banner = _nmap_banner(nmap) + if banner: + services_banners[banner] += 1 + + return { + "enabled": True, + "domain": domain, + "generated_at": datetime.now(timezone.utc).isoformat(), + "policy": { + "web_fingerprint_primary": True, + "nmap_role": "complementary_appendix", + "user_agent": USER_AGENT, + "http_timeout_seconds": HTTP_TIMEOUT, + "tls_timeout_seconds": TLS_TIMEOUT, + "concurrency": CONCURRENCY, + "passive_db_enabled": PASSIVE_DB_ENABLED, + "passive_db_configured": _passive_configured(), + "open_services_label": "Open Services from DB" if PASSIVE_DB_ENABLED and _passive_configured() else "Open Services observed", + }, + "system_locations": dict(system_locations) or {UNKNOWN_COUNTRY: 0}, + "hosting_networks": hosting_networks, + "services_banners": dict(services_banners), + "a_records": [_a_record_row(h) for h in hosts], + "mx_records": [_mx_record_row(h) for h in mx_records], + "ns_records": [_ns_record_row(h) for h in ns_records], + "txt_records": txt_records, + "hosts": hosts, + "guardrail_logs": [h for h in hosts + mx_records + ns_records if h.get("guardrail")], + "notes": [ + "Nmap est une source complémentaire. Les services HTTP/HTTPS peuvent être détectés par fingerprint web même lorsque Nmap ne remonte pas de service.", + "Les valeurs non observées sont marquées Non détecté ou Version non exposée, sans invention de technologie.", + ], + } + + +def _empty(domain: str, enabled: bool = True) -> dict: + return { + "enabled": enabled, + "domain": domain, + "system_locations": {}, + "hosting_networks": [], + "services_banners": {}, + "a_records": [], + "mx_records": [], + "ns_records": [], + "txt_records": [], + "hosts": [], + } + + +def _resolver(timeout: float = 4.0) -> dns.resolver.Resolver: + r = dns.resolver.Resolver() + r.timeout = timeout + r.lifetime = timeout + return r + + +def _resolve_ips(hostname: str) -> dict: + ips: list[str] = [] + errors: list[str] = [] + for rtype in ("A", "AAAA"): + try: + ips.extend(str(a).strip().strip(".") for a in _resolver().resolve(hostname, rtype)) + except dns.resolver.NoAnswer: + continue + except dns.resolver.NXDOMAIN: + errors.append("NXDOMAIN") + break + except Exception as exc: + errors.append(str(exc)) + ips = sorted(set(ips)) + return { + "ips": ips, + "public_ips": [ip for ip in ips if is_public_ip(ip)], + "blocked_ips": [ip for ip in ips if not is_public_ip(ip)], + "errors": sorted(set(errors)), + } + + +def _a_record_hosts(domain: str, dns_result: dict, subdomains_result: dict, ip_inventory: dict) -> list[dict]: + hosts: dict[str, dict] = {} + + def add(host: str, source: str): + host = (host or "").strip().lower().strip(".") + if not host: + return + hosts.setdefault(host, {"host": host, "sources": set(), "priority": None})["sources"].add(source) + + add(domain, "dns") + add(f"www.{domain}", "dns") + if SCAN_ALL_SUBDOMAINS: + for sub in subdomains_result.get("subdomains", []) or []: + if isinstance(sub, dict): + add(sub.get("host") or sub.get("subdomain") or sub.get("name"), "subdomain_dataset") + else: + add(str(sub), "subdomain_dataset") + for entry in ip_inventory.get("entries", []) or []: + if entry.get("source") in {"subdomain", "root", "www", "web_target"}: + add(entry.get("hostname"), entry.get("source") or "ip_inventory") + + out = [] + by_entry = {e.get("hostname"): e for e in ip_inventory.get("entries", []) or []} + for host, data in sorted(hosts.items()): + inv = by_entry.get(host) or {} + resolved = {"ips": inv.get("ips") or [], "public_ips": inv.get("public_ips") or [], "blocked_ips": inv.get("blocked_ips") or [], "errors": inv.get("errors") or []} + if not resolved["ips"]: + resolved = _resolve_ips(host) + out.append({"host": host, "ips": resolved["ips"], "public_ips": resolved["public_ips"], "blocked_ips": resolved["blocked_ips"], "errors": resolved["errors"], "sources": sorted(data["sources"] | {"dns"})}) + return out + + +def _mx_records(mail_result: dict) -> list[dict]: + rows = [] + for raw in mail_result.get("mx", {}).get("values", []) or []: + parts = str(raw).strip().split() + if len(parts) >= 2: + priority = _to_int(parts[0]) + host = parts[-1].strip(".").lower() + else: + priority = None + host = str(raw).strip(".").lower() + resolved = _resolve_ips(host) if host else {"ips": [], "public_ips": [], "blocked_ips": [], "errors": []} + rows.append({"host": host, "priority": priority, **resolved, "sources": ["mx", "dns"]}) + return rows + + +def _ns_records(dns_result: dict) -> list[dict]: + rows = [] + for host in dns_result.get("records", {}).get("NS", {}).get("values", []) or []: + host = str(host).strip(".").lower() + resolved = _resolve_ips(host) + rows.append({"host": host, **resolved, "sources": ["ns", "dns"]}) + return rows + + +def _txt_records(dns_result: dict) -> list[dict]: + rows = [] + for raw in dns_result.get("records", {}).get("TXT", {}).get("values", []) or []: + value = str(raw).replace('" "', "").replace('"', "").strip() + item = {"value": value, "type": "TXT", "sources": ["dns"]} + if value.lower().startswith("v=spf1"): + item["type"] = "SPF" + item["spf"] = { + "includes": re.findall(r"\binclude:([^\s]+)", value, flags=re.I)[:15], + "ip4": re.findall(r"\bip4:([^\s]+)", value, flags=re.I)[:20], + "ip6": re.findall(r"\bip6:([^\s]+)", value, flags=re.I)[:20], + "providers": _spf_providers(value), + "resolution_policy": "includes parsed only; no aggressive recursive resolution", + } + rows.append(item) + return rows + + +def _spf_providers(value: str) -> list[str]: + providers = [] + for inc in re.findall(r"\binclude:([^\s]+)", value, flags=re.I): + host = inc.lower().strip(".") + if "mailjet" in host: + providers.append("Mailjet") + elif "outlook" in host or "protection.outlook" in host: + providers.append("Microsoft 365 / Exchange Online Protection") + elif "mailinblack" in host: + providers.append("Mailinblack") + else: + providers.append(host) + return sorted(dict.fromkeys(providers)) + + +async def _enrich_host(item: dict, record_type: str, revip: dict[str, set[str]], asn_cache: dict[str, dict], service_scan: dict) -> dict: + host = item.get("host") or UNKNOWN + public_ips = item.get("public_ips") or [] + blocked_ips = item.get("blocked_ips") or [] + ip = public_ips[0] if public_ips else (item.get("ips") or ["Not found"])[0] if item.get("ips") else "Not found" + guardrail = None + sources = set(item.get("sources") or []) + + if blocked_ips or (item.get("ips") and not public_ips): + guardrail = { + "status": "blocked_by_guard", + "reason": "IP privée, réservée ou résolution mixte ; fingerprint HTTP/TLS bloqué.", + "blocked_ips": blocked_ips or item.get("ips") or [], + } + sources.add("guardrail") + + asn = await _asn_for_ip(ip, asn_cache) if _public_ip(ip) else _asn_unknown() + http_results: dict[str, dict] = {} + open_services: list[dict] = [] + technologies: list[dict] = [] + tls = {} + + if _public_ip(ip) and not guardrail: + http_results = await _fingerprint_http(host) + for scheme, result in http_results.items(): + if result.get("reachable"): + sources.add(f"{scheme}_fingerprint") + banner = _server_banner(result.get("headers") or {}) + svc = { + "scheme": scheme, + "port": 443 if scheme == "https" else 80, + "service": scheme, + "banner": banner or "unknown server", + "title": result.get("title") or UNKNOWN, + "status_code": result.get("status_code"), + "final_url": result.get("final_url"), + "source": f"{scheme}_fingerprint", + } + open_services.append(svc) + technologies.extend(_detect_technologies(result.get("headers") or {}, result.get("body_sample") or "")) + if http_results.get("https", {}).get("reachable"): + tls = await asyncio.to_thread(_fingerprint_tls, host) + if tls and not tls.get("error"): + sources.add("tls_certificate") + + nmap_services = _nmap_for_host(host, service_scan) + if nmap_services: + sources.add("nmap") + technologies = _dedup_technologies(technologies) + + return { + "host": host, + "ip": ip, + "ips": item.get("ips") or [], + "asn": asn.get("asn") or UNKNOWN, + "network": asn.get("network") or UNKNOWN, + "asn_name": asn.get("asn_name") or UNKNOWN, + "country": asn.get("country") or UNKNOWN_COUNTRY, + "city": UNKNOWN, + "provider": asn.get("provider") or UNKNOWN, + "record_type": record_type, + "priority": item.get("priority"), + "open_services": open_services, + "http": http_results.get("http") or {}, + "https": http_results.get("https") or {}, + "tls": tls or {}, + "technologies": technologies, + "nmap_services": nmap_services, + "revip_count": len(revip.get(ip, set())) if _public_ip(ip) else 0, + "guardrail": guardrail, + "resolution_errors": item.get("errors") or [], + "sources": sorted(sources), + } + + +def _host_skeleton(item: dict, record_type: str) -> dict: + return {"host": item.get("host"), "ip": (item.get("public_ips") or item.get("ips") or ["Not found"])[0], "record_type": record_type, "open_services": [], "technologies": [], "nmap_services": [], "revip_count": 0, "sources": item.get("sources") or []} + + +def _record_skeleton(item: dict, record_type: str) -> dict: + h = _host_skeleton(item, record_type) + h["priority"] = item.get("priority") + return h + + +async def _fingerprint_http(host: str) -> dict[str, dict]: + headers = {"User-Agent": USER_AGENT, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"} + limits = httpx.Limits(max_connections=CONCURRENCY, max_keepalive_connections=CONCURRENCY) + async with httpx.AsyncClient(timeout=HTTP_TIMEOUT, follow_redirects=True, verify=False, headers=headers, limits=limits) as client: + http, https = await asyncio.gather(_fetch(client, f"http://{host}"), _fetch(client, f"https://{host}")) + return {"http": http, "https": https} + + +async def _fetch(client: httpx.AsyncClient, url: str) -> dict: + try: + response = await client.get(url) + content_type = response.headers.get("content-type", "") + body = response.text[:180000] if "text" in content_type or "html" in content_type or not content_type else "" + return { + "url": url, + "reachable": True, + "status_code": response.status_code, + "final_url": str(response.url), + "headers": dict(response.headers), + "title": _html_title(body), + "body_sample": body[:50000], + "error": None, + } + except Exception as exc: + return {"url": url, "reachable": False, "status_code": None, "final_url": None, "headers": {}, "title": None, "body_sample": "", "error": str(exc)} + + +def _fingerprint_tls(host: str) -> dict: + try: + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + with socket.create_connection((host, 443), timeout=TLS_TIMEOUT) as sock: + with ctx.wrap_socket(sock, server_hostname=host) as ssock: + der = ssock.getpeercert(binary_form=True) + parsed = _parse_tls_der(der) if der else {} + parsed.update({ + "tls_version": ssock.version() or UNKNOWN, + "cipher": ssock.cipher()[0] if ssock.cipher() else UNKNOWN, + }) + return parsed + except Exception as exc: + return {"cn": UNKNOWN, "san": [], "issuer": UNKNOWN, "issuer_organization": UNKNOWN, "expires_at": UNKNOWN, "days_remaining": 0, "tls_version": UNKNOWN, "error": str(exc)} + + +def _parse_tls_der(der: bytes) -> dict: + try: + from cryptography import x509 + from cryptography.hazmat.backends import default_backend + cert = x509.load_der_x509_certificate(der, default_backend()) + cn = UNKNOWN + try: + cn = cert.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)[0].value + except Exception: + pass + issuer_cn = UNKNOWN + issuer_org = UNKNOWN + try: + issuer_cn = cert.issuer.get_attributes_for_oid(x509.NameOID.COMMON_NAME)[0].value + except Exception: + pass + try: + issuer_org = cert.issuer.get_attributes_for_oid(x509.NameOID.ORGANIZATION_NAME)[0].value + except Exception: + pass + san = [] + try: + san_ext = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName).value + san = san_ext.get_values_for_type(x509.DNSName) + except Exception: + pass + expires = cert.not_valid_after_utc if hasattr(cert, "not_valid_after_utc") else cert.not_valid_after.replace(tzinfo=timezone.utc) + return { + "cn": cn or UNKNOWN, + "san": san, + "issuer": issuer_cn or UNKNOWN, + "issuer_organization": issuer_org or UNKNOWN, + "expires_at": expires.astimezone(timezone.utc).isoformat(), + "days_remaining": max(0, (expires.astimezone(timezone.utc) - datetime.now(timezone.utc)).days), + } + except Exception as exc: + return {"cn": UNKNOWN, "san": [], "issuer": UNKNOWN, "issuer_organization": UNKNOWN, "expires_at": UNKNOWN, "days_remaining": 0, "parse_error": str(exc)} + + +def _html_title(body: str) -> str | None: + match = re.search(r"]*>(.*?)", body or "", flags=re.I | re.S) + if not match: + return None + title = re.sub(r"\s+", " ", re.sub(r"<[^>]+>", "", match.group(1))).strip() + return title[:220] or None + + +def _server_banner(headers: dict) -> str | None: + for k, v in headers.items(): + if k.lower() == "server" and v: + return str(v).strip() + return None + + +def _detect_technologies(headers: dict, body: str) -> list[dict]: + tech: list[dict] = [] + header_l = {str(k).lower(): str(v) for k, v in (headers or {}).items()} + server = header_l.get("server", "") + powered = header_l.get("x-powered-by", "") + + if server: + if "microsoft-iis" in server.lower(): + version = _first_version(server) + tech.append(_tech("Microsoft-IIS", version, "http_header", "high")) + tech.append(_tech("IIS", version, "http_header", "high")) + tech.append(_tech("Windows Server", None, "http_header", "medium", note="probable")) + if "apache" in server.lower(): + tech.append(_tech("Apache HTTP Server", _version_after(server, "Apache/"), "http_header", "high")) + if "win64" in server.lower() or "win32" in server.lower(): + tech.append(_tech("Windows Server", None, "http_header", "medium", note="probable")) + openssl = re.search(r"OpenSSL/([0-9][^\s;)]+)", server, flags=re.I) + if openssl: + tech.append(_tech("OpenSSL", openssl.group(1), "http_header", "high")) + if "nginx" in server.lower(): + tech.append(_tech("Nginx", _version_after(server, "nginx/"), "http_header", "high")) + if powered: + if "asp.net" in powered.lower(): + tech.append(_tech("ASP.NET", _first_version(powered), "http_header", "high")) + php = re.search(r"PHP/?([0-9][^\s;]*)?", powered, flags=re.I) + if php: + tech.append(_tech("PHP", php.group(1), "http_header", "high")) + + body_l = (body or "").lower() + patterns = [ + (r"jquery(?:\.min)?[-.]([0-9][0-9a-zA-Z.\-]*)\.js|jquery-([0-9][0-9a-zA-Z.\-]*)", "jQuery"), + (r"jquery-ui(?:\.min)?[-.]([0-9][0-9a-zA-Z.\-]*)\.js|jquery-ui-([0-9][0-9a-zA-Z.\-]*)", "jQuery UI"), + (r"bootstrap(?:\.bundle|\.min)?[-.]([0-9][0-9a-zA-Z.\-]*)\.(?:js|css)|bootstrap[/_-]([0-9][0-9a-zA-Z.\-]*)", "Bootstrap"), + ] + for regex, name in patterns: + m = re.search(regex, body_l, flags=re.I) + if m: + tech.append(_tech(name, next((g for g in m.groups() if g), None), "html_body", "medium")) + elif name.lower().replace(" ", "-") in body_l or name.lower().replace(" ", "") in body_l: + tech.append(_tech(name, None, "html_body", "medium")) + simple = [ + ("modernizr", "Modernizr"), ("onetrust", "OneTrust"), ("prettyphoto", "prettyPhoto"), + ("platform.twitter.com", "Twitter"), ("widgets.js", "Twitter"), ("wp-content", "WordPress"), + ("glpi", "GLPI"), ("zabbix", "Zabbix"), ("fortinet", "Fortinet"), ("fortigate", "Fortinet"), + ("fortipam", "Fortinet"), ("fortiguard", "Fortinet"), + ] + for needle, name in simple: + if needle in body_l: + tech.append(_tech(name, None, "html_body", "medium")) + return tech + + +def _tech(name: str, version: str | None, source: str, confidence: str, note: str | None = None) -> dict: + item = {"name": name, "source": source, "confidence": confidence} + if version: + item["version"] = version.strip().strip("/;,)") + if note: + item["note"] = note + return item + + +def _dedup_technologies(items: list[dict]) -> list[dict]: + dedup = {} + for t in items: + key = (t.get("name"), t.get("version") or "") + dedup[key] = t + return list(dedup.values()) + + +def _version_after(text: str, token: str) -> str | None: + idx = text.lower().find(token.lower()) + if idx < 0: + return None + rest = text[idx + len(token):] + m = re.match(r"([0-9][^\s;)]*)", rest) + return m.group(1) if m else None + + +def _first_version(text: str) -> str | None: + m = re.search(r"([0-9]+(?:\.[0-9A-Za-z_-]+)+)", text or "") + return m.group(1) if m else None + + +async def _asn_for_ip(ip: str, cache: dict[str, dict]) -> dict: + if ip in cache: + return cache[ip] + if not _public_ip(ip): + cache[ip] = _asn_unknown() + return cache[ip] + try: + import ipaddress + obj = ipaddress.ip_address(ip) + if obj.version == 4: + qname = ".".join(reversed(ip.split("."))) + ".origin.asn.cymru.com" + else: + qname = ".".join(reversed(obj.exploded.replace(":", ""))) + ".origin6.asn.cymru.com" + answers = await asyncio.to_thread(lambda: list(_resolver(3.0).resolve(qname, "TXT"))) + raw = str(answers[0]).replace('"', "") if answers else "" + parts = [p.strip() for p in raw.split("|")] + asn_num = parts[0] if parts else "" + network = parts[1] if len(parts) > 1 else UNKNOWN + cc = parts[2] if len(parts) > 2 else "" + as_name = parts[4] if len(parts) > 4 else UNKNOWN + asn_name = f"{as_name}" if as_name.startswith("AS") else f"AS{asn_num} - {as_name}" if asn_num and as_name != UNKNOWN else UNKNOWN + country = COUNTRY_NAMES.get(cc.upper(), cc.upper() or UNKNOWN_COUNTRY) + provider = _provider_from_asn_name(asn_name) + cache[ip] = {"asn": f"AS{asn_num}" if asn_num and not str(asn_num).startswith("AS") else asn_num or UNKNOWN, "network": network, "asn_name": asn_name, "country": country, "provider": provider} + except Exception: + cache[ip] = _asn_unknown() + return cache[ip] + + +def _asn_unknown() -> dict: + return {"asn": UNKNOWN, "network": UNKNOWN, "asn_name": UNKNOWN, "country": UNKNOWN_COUNTRY, "provider": UNKNOWN} + + +def _provider_from_asn_name(name: str) -> str: + if not name or name == UNKNOWN: + return UNKNOWN + if " - " in name: + tail = name.split(" - ", 1)[1] + else: + tail = name + return tail.split(",")[0].strip() or UNKNOWN + + +def _nmap_for_host(host: str, service_scan: dict) -> list[dict]: + rows = [] + for p in service_scan.get("open_ports", []) or []: + if (p.get("hostname") or p.get("host") or "").lower() == host.lower(): + rows.append(p) + return rows + + +def _nmap_banner(p: dict) -> str | None: + product = p.get("product") or "" + version = p.get("version") or "" + name = p.get("name") or p.get("service") or "" + return " ".join(x for x in [product or name, version] if x).strip() or None + + +def _a_record_row(h: dict) -> dict: + return {k: h.get(k) for k in ["host", "ip", "asn", "network", "asn_name", "country", "open_services", "technologies", "revip_count", "sources", "guardrail"]} + + +def _mx_record_row(h: dict) -> dict: + row = _a_record_row(h) + row["priority"] = h.get("priority") + return row + + +def _ns_record_row(h: dict) -> dict: + return _a_record_row(h) + + +def _to_int(value: Any) -> int | None: + try: + return int(str(value)) + except Exception: + return None + + +def _public_ip(ip: str | None) -> bool: + return bool(ip and ip != "Not found" and is_public_ip(ip)) + + +def _passive_configured() -> bool: + return bool(os.getenv("SHODAN_API_KEY") or (os.getenv("CENSYS_API_ID") and os.getenv("CENSYS_API_SECRET")) or os.getenv("SECURITYTRAILS_API_KEY")) diff --git a/backend/app/services/nmap_audit.py b/backend/app/services/nmap_audit.py index 3238501..11eee60 100644 --- a/backend/app/services/nmap_audit.py +++ b/backend/app/services/nmap_audit.py @@ -11,7 +11,7 @@ from app.services.network_guard import resolve_ips -TOP_PORTS = os.getenv("OPENEASM_NMAP_TOP_PORTS", "") +TOP_PORTS = os.getenv("OPENEASM_NMAP_TOP_PORTS", "1000") HOST_TIMEOUT = os.getenv("OPENEASM_NMAP_HOST_TIMEOUT", "120s") PROCESS_TIMEOUT = int(os.getenv("OPENEASM_NMAP_PROCESS_TIMEOUT", "150"))