admin / Synapse-NetscanXi
publicNetwork Scanning, Vulnerability and Compliance Application
Synapse-NetscanXi / NetscanXiVersion13 / app / diff.py
6182 B · main
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | """ Change detection: compare a new scan's hosts against the previous scan. Produces a list of change records used for the dashboard + alerting. """ def _index(hosts): return {h["ip"]: h for h in hosts if h.get("ip")} def _ports_of(host): return {f'{p["port"]}/{p["protocol"]}': p for p in host.get("port_details", [])} def _vuln_ids(host): return {v["id"] for v in host.get("vulns", [])} def diff_scans(old_hosts, new_hosts): """Return a list of change dicts: {ip, kind, detail, severity}.""" changes = [] old = _index(old_hosts) new = _index(new_hosts) # New / gone hosts for ip in sorted(set(new) - set(old)): h = new[ip] name = h.get("hostname", "-") changes.append({ "ip": ip, "kind": "host_new", "detail": f"New device appeared ({name}, {h.get('device_type','?')})", "severity": "warn", }) for ip in sorted(set(old) - set(new)): changes.append({ "ip": ip, "kind": "host_gone", "detail": f"Device no longer responding ({old[ip].get('hostname','-')})", "severity": "info", }) # Per-host comparison for hosts present in both for ip in sorted(set(old) & set(new)): oh, nh = old[ip], new[ip] oports, nports = _ports_of(oh), _ports_of(nh) for p in sorted(set(nports) - set(oports)): svc = nports[p].get("service", "") changes.append({ "ip": ip, "kind": "port_new", "detail": f"Port {p} opened ({svc or 'unknown'})", "severity": "warn", }) for p in sorted(set(oports) - set(nports)): changes.append({ "ip": ip, "kind": "port_gone", "detail": f"Port {p} closed", "severity": "info", }) # Service/version changes on shared ports for p in sorted(set(oports) & set(nports)): o_v = (oports[p].get("product", ""), oports[p].get("version", "")) n_v = (nports[p].get("product", ""), nports[p].get("version", "")) if o_v != n_v and any(n_v): changes.append({ "ip": ip, "kind": "service_changed", "detail": f"Port {p} service changed: " f"{' '.join(filter(None,o_v)) or '?'} -> " f"{' '.join(filter(None,n_v)) or '?'}", "severity": "info", }) # New vulnerabilities new_vulns = _vuln_ids(nh) - _vuln_ids(oh) for v in nh.get("vulns", []): if v["id"] in new_vulns: sev = "critical" if (v.get("kev") or v.get("exploit")) else "warn" tag = " [KEV]" if v.get("kev") else (" [exploit]" if v.get("exploit") else "") changes.append({ "ip": ip, "kind": "vuln_new", "detail": f"New vulnerability: {v['id']}" f"{(' CVSS '+str(v['cvss'])) if v.get('cvss') else ''}{tag}", "severity": sev, }) # OS change if oh.get("os") and nh.get("os") and oh["os"] != nh["os"] \ and nh["os"] != "unknown" and oh["os"] != "unknown": changes.append({ "ip": ip, "kind": "os_changed", "detail": f"OS fingerprint changed: {oh['os']} -> {nh['os']}", "severity": "info", }) # v10r1: Docker container / image drift for c in _docker_changes(ip, oh, nh): changes.append(c) return changes def _container_names(host): d = host.get("docker") or {} return {c.get("name") for c in (d.get("containers") or []) if c.get("name")} def _image_tags(host): d = host.get("docker") or {} tags = set() for img in (d.get("images") or []): for t in (img.get("tags") or []): tags.add(t) return tags def _docker_changes(ip, oh, nh): """Detect container/image drift and new container-image CVEs between scans.""" out = [] od, nd = oh.get("docker") or {}, nh.get("docker") or {} if not (od.get("is_docker") or nd.get("is_docker")): return out # Containers added / removed oc, nc = _container_names(oh), _container_names(nh) for name in sorted(nc - oc): out.append({"ip": ip, "kind": "container_new", "detail": f"New container: {name}", "severity": "info"}) for name in sorted(oc - nc): out.append({"ip": ip, "kind": "container_gone", "detail": f"Container removed: {name}", "severity": "info"}) # Image tags added / removed ot, nt = _image_tags(oh), _image_tags(nh) for t in sorted(nt - ot): out.append({"ip": ip, "kind": "image_new", "detail": f"New image: {t}", "severity": "info"}) # Privileged / docker.sock exposure appearing def _risky(host): d = host.get("docker") or {} return {c.get("name"): (c.get("privileged"), c.get("docker_sock_mounted")) for c in (d.get("containers") or [])} orisk, nrisk = _risky(oh), _risky(nh) for name, (priv, sock) in nrisk.items(): oprev = orisk.get(name, (False, False)) if priv and not oprev[0]: out.append({"ip": ip, "kind": "container_privileged", "detail": f"Container '{name}' is now privileged", "severity": "warn"}) if sock and not oprev[1]: out.append({"ip": ip, "kind": "container_sock", "detail": f"Container '{name}' now mounts docker.sock", "severity": "critical"}) # New container-image CVEs (count delta of KEV findings) def _img_kev(host): d = host.get("docker") or {} return (d.get("image_scan") or {}).get("kev_count", 0) or 0 nk, ok_ = _img_kev(nh), _img_kev(oh) if nk > ok_: out.append({"ip": ip, "kind": "image_kev_new", "detail": f"{nk - ok_} new known-exploited (KEV) container CVE(s)", "severity": "critical"}) return out |