Reading ESTABLISHED sockets as evidence (943569f) exposed a dedup key that
was too narrow: it included the protocol, which was harmless while only
listeners counted, but a busy host has many established sockets on the same
service port, and tcp vs tcp6 already made a single listener look like two
services.
The tester's domain controller listed "RDP :3389" four times and LDAP four
times. Since each additional entry adds 40% of its weight, the exposure score
inflated to 100 for what is one RDP and one LDAP service. Deduplicating by
port alone fixes both: one service, one entry, one weight.
Also feeds Recent Critical from four narrow server-side queries (critical,
high, KEV, EUVD) instead of filtering the 300 newest rows in the browser. A
batch of low-severity CVEs — Chrome publishes dozens at once — fills that
window completely and empties the widget, which no amount of extra depth
fixes; only filtering before the limit does.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
76 lines
3.4 KiB
Python
76 lines
3.4 KiB
Python
"""Ports as evidence + comparable builds — run: python tests/test_port_and_build_evidence.py
|
|
|
|
Two independent findings from the same test estate:
|
|
|
|
1. Wazuh's syscollector does not always report a LISTENER. On a Windows Server
|
|
2025 DC, netstat showed 3389 ESTABLISHED and syscollector returned no
|
|
listening entry at all — so RDP scored zero exposure and the DC role went
|
|
undetected on a live domain controller.
|
|
2. MSRC states the Teams fix as 25060212043 while Teams ships
|
|
26183.1003.4002.4460. Compared element by element, 26183 < 25060212043 is
|
|
True, so a current Teams was reported vulnerable.
|
|
"""
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from app.services.exposure_service import analyze_ports
|
|
from app.services.risk_dimensions_service import detect_risk_dimensions
|
|
from app.services.msrc_scan_service import affected_cves
|
|
|
|
|
|
def demo():
|
|
# --- 1. an inbound ESTABLISHED socket proves the service runs ---
|
|
rdp_established = [{"protocol": "tcp", "state": "established",
|
|
"local_port": 3389, "local_ip": "10.0.0.1",
|
|
"process": "svchost.exe"}]
|
|
score, services = analyze_ports(rdp_established)
|
|
assert score > 0 and services[0]["service"] == "RDP"
|
|
|
|
# An OUTBOUND connection uses an ephemeral local port — never a service.
|
|
outbound = [{"protocol": "tcp", "state": "established",
|
|
"local_port": 52079, "local_ip": "10.0.0.1"}]
|
|
assert analyze_ports(outbound) == (0.0, [])
|
|
|
|
# Closing states still say nothing.
|
|
assert analyze_ports([{"protocol": "tcp", "state": "time_wait",
|
|
"local_port": 3389, "local_ip": "10.0.0.1"}]) == (0.0, [])
|
|
|
|
# A DC answering Kerberos over an established socket is a DC.
|
|
dc = detect_risk_dimensions(
|
|
[{"protocol": "tcp", "state": "established", "local_port": 88,
|
|
"local_ip": "10.0.0.1"}], [])
|
|
assert any(d["role"] == "domain_controller" for d in dc["dimensions"])
|
|
# Ephemeral local port on an established socket stays ignored.
|
|
assert detect_risk_dimensions(
|
|
[{"protocol": "tcp", "state": "established", "local_port": 49712,
|
|
"local_ip": "10.0.0.1"}], [])["dimensions"] == []
|
|
|
|
# --- 2. builds of different shapes are not comparable ---
|
|
teams_entry = [{"cve": "CVE-2025-49731", "build": "25060212043"}]
|
|
assert affected_cves(teams_entry, "26183.1003.4002.4460",
|
|
branch_match=False) == []
|
|
# Same shape still compares normally.
|
|
same_shape = [{"cve": "CVE-X", "build": "26183.1003.4002.5000"}]
|
|
assert len(affected_cves(same_shape, "26183.1003.4002.4460",
|
|
branch_match=False)) == 1
|
|
assert affected_cves(same_shape, "26183.1003.4002.5000",
|
|
branch_match=False) == []
|
|
|
|
# One service, many sockets. Every duplicate used to add 40% of its
|
|
# weight, so a busy DC inflated to 100 on what is one RDP service.
|
|
many = [{"protocol": "tcp", "state": "established", "local_port": 3389,
|
|
"local_ip": "10.0.0.1"} for _ in range(4)]
|
|
many.append({"protocol": "tcp6", "state": "listening", "local_port": 3389,
|
|
"local_ip": "::"})
|
|
score_many, services_many = analyze_ports(many)
|
|
assert len(services_many) == 1, services_many
|
|
assert score_many == 30.0, score_many # exactly one RDP, not four
|
|
|
|
print("port + build evidence OK")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
demo()
|