From b20c47a0a5f3a4e0f0b13c564eba401abe39a68c Mon Sep 17 00:00:00 2001 From: Pouzor Date: Tue, 23 Jun 2026 12:25:10 +0200 Subject: [PATCH] feat: add tiered service matching with http_regex and port-agnostic signatures Adds match_service() priority walk for fingerprinting: 1. port + http_regex confirmed 2. port + banner_regex confirmed 3. port:null + http_regex confirmed (custom-port services) 4. port-only fallback http_regex is strict only once a probe has run; with no probe signals it degrades to port-only matching, preserving pre-probe behaviour. match_port kept as a probe-less alias. ha-relevant: yes --- backend/app/services/fingerprint.py | 102 ++++++++++++++++++++++++---- backend/tests/test_fingerprint.py | 84 +++++++++++++++++++++++ 2 files changed, 173 insertions(+), 13 deletions(-) diff --git a/backend/app/services/fingerprint.py b/backend/app/services/fingerprint.py index 460bdf3..4532644 100644 --- a/backend/app/services/fingerprint.py +++ b/backend/app/services/fingerprint.py @@ -50,25 +50,101 @@ def _load_oui() -> dict[str, str]: return _OUI_MAP -def match_port(port: int, protocol: str, banner: str | None = None) -> dict[str, Any] | None: - """Return the first signature matching port+protocol, optionally banner.""" - for sig in _load(): - if sig["port"] != port or sig["protocol"] != protocol: - continue - if sig.get("banner_regex") and (not banner or not re.search(sig["banner_regex"], banner, re.IGNORECASE)): - continue - return sig - return None +def _http_regex_hit(sig: dict[str, Any], http_signals: dict[str, Any] | None) -> bool: + """True when the signature's http_regex matches the probe's title/headers.""" + rx = sig.get("http_regex") + if not rx or not http_signals: + return False + headers = http_signals.get("headers") or {} + haystack = " ".join( + s for s in ( + http_signals.get("title"), + headers.get("Server"), + headers.get("X-Powered-By"), + ) if s + ) + return bool(haystack and re.search(rx, haystack, re.IGNORECASE)) -def fingerprint_ports(open_ports: list[dict[str, Any]]) -> list[dict[str, Any]]: +def _service_tier( + sig: dict[str, Any], + port: int, + protocol: str, + banner: str | None, + http_signals: dict[str, Any] | None, +) -> int | None: """ - Given a list of {port, protocol, banner?} dicts, return matched services. - Unknown ports are included as unknown_service. + Rank how well a signature matches (lower = stronger). None = not a match. + + Tier 1: port match + http_regex confirmed + Tier 2: port match + banner_regex confirmed + Tier 3: port-agnostic (port: null) + http_regex confirmed + Tier 4: port match only (no regex, or http_regex with probe disabled) + + When http_signals is None (probe not run) an http_regex entry degrades to + a port-only match — identical to pre-probe behaviour, no regression. + When http_signals is provided, http_regex is strict: a miss disqualifies. + """ + probe_ran = http_signals is not None + has_http = bool(sig.get("http_regex")) + + # Port-agnostic entries (port: null) match purely on HTTP signals. + if sig.get("port") is None: + if has_http and _http_regex_hit(sig, http_signals): + return 3 + return None + + if sig["port"] != port or sig["protocol"] != protocol: + return None + + # http_regex is authoritative once a probe has run. + if has_http and probe_ran: + return 1 if _http_regex_hit(sig, http_signals) else None + + if sig.get("banner_regex"): + if banner and re.search(sig["banner_regex"], banner, re.IGNORECASE): + return 2 + return None + + # No regex constraint (or http_regex but probe disabled) → port-only guess. + return 4 + + +def match_service( + port: int, + protocol: str, + banner: str | None = None, + http_signals: dict[str, Any] | None = None, +) -> dict[str, Any] | None: + """Return the best signature for a port, walking tiers most-specific first.""" + best: dict[str, Any] | None = None + best_tier = 99 + for sig in _load(): + tier = _service_tier(sig, port, protocol, banner, http_signals) + if tier is not None and tier < best_tier: + best, best_tier = sig, tier + if best_tier == 1: + break # strongest possible — stop early + return best + + +def match_port(port: int, protocol: str, banner: str | None = None) -> dict[str, Any] | None: + """Back-compat alias: match without HTTP-probe signals.""" + return match_service(port, protocol, banner) + + +def fingerprint_ports( + open_ports: list[dict[str, Any]], +) -> list[dict[str, Any]]: + """ + Given a list of {port, protocol, banner?, http_signals?} dicts, return + matched services. Unknown ports are included as unknown_service. """ results = [] for p in open_ports: - sig = match_port(p["port"], p.get("protocol", "tcp"), p.get("banner")) + sig = match_service( + p["port"], p.get("protocol", "tcp"), p.get("banner"), p.get("http_signals") + ) if sig: results.append({ "port": p["port"], diff --git a/backend/tests/test_fingerprint.py b/backend/tests/test_fingerprint.py index 168b437..bdf487e 100644 --- a/backend/tests/test_fingerprint.py +++ b/backend/tests/test_fingerprint.py @@ -5,6 +5,7 @@ import pytest from app.services.fingerprint import ( fingerprint_ports, match_port, + match_service, suggest_node_type, suggest_type_from_mac, ) @@ -249,3 +250,86 @@ def test_suggest_node_type_ubiquiti_mac_with_bgp_upgrades_to_router(): mac="24:a4:3c:11:22:33", ) assert result == "router" + + +# ── match_service: HTTP probe + port-agnostic ────────────────────────────────── + +HTTP_SIGNATURES = [ + # Generic web fallback on 8096 (port-only guess) + {"port": 8096, "protocol": "tcp", "banner_regex": None, "http_regex": None, + "service_name": "HTTP", "icon": "🌐", "category": "web", "suggested_node_type": "server"}, + # Same port, but confirmed by HTML title → should win when probe confirms + {"port": 8096, "protocol": "tcp", "banner_regex": None, "http_regex": "Jellyfin", + "service_name": "Jellyfin", "icon": "🎬", "category": "media", "suggested_node_type": "server"}, + # Port-agnostic: matches on HTTP content regardless of port + {"port": None, "protocol": "tcp", "banner_regex": None, "http_regex": "Portainer", + "service_name": "Portainer", "icon": "🐳", "category": "container", "suggested_node_type": "server"}, + # Banner-based entry, no http + {"port": 9090, "protocol": "tcp", "banner_regex": "prometheus", "http_regex": None, + "service_name": "Prometheus", "icon": "🔥", "category": "monitoring", "suggested_node_type": "server"}, +] + + +@pytest.fixture +def http_signatures(): + with patch("app.services.fingerprint._load", return_value=HTTP_SIGNATURES): + yield + + +def test_http_regex_confirmed_beats_port_only(http_signatures): + # Probe ran and title matches → Jellyfin (tier 1) beats generic HTTP (tier 4) + sig = match_service(8096, "tcp", banner=None, + http_signals={"title": "Jellyfin", "headers": {}}) + assert sig["service_name"] == "Jellyfin" + + +def test_http_regex_matches_on_header(http_signatures): + sig = match_service(8096, "tcp", banner=None, + http_signals={"title": None, "headers": {"Server": "Jellyfin"}}) + assert sig["service_name"] == "Jellyfin" + + +def test_http_regex_miss_falls_back_to_port_only(http_signatures): + # Probe ran but nothing matched the http_regex → generic port-only entry wins + sig = match_service(8096, "tcp", banner=None, + http_signals={"title": "Some Other App", "headers": {}}) + assert sig["service_name"] == "HTTP" + + +def test_probe_disabled_ignores_http_regex(http_signatures): + # http_signals=None (deep scan off) → http_regex entry degrades to port-only, + # generic entry (listed first) wins — identical to pre-probe behaviour. + sig = match_service(8096, "tcp", banner=None, http_signals=None) + assert sig["service_name"] == "HTTP" + + +def test_port_agnostic_match_on_custom_port(http_signatures): + # Portainer found on a non-standard port, recognised purely by HTTP content + sig = match_service(54321, "tcp", banner=None, + http_signals={"title": "Portainer", "headers": {}}) + assert sig["service_name"] == "Portainer" + + +def test_port_agnostic_requires_probe(http_signatures): + # Same custom port, probe off → no signal → no match + assert match_service(54321, "tcp", banner=None, http_signals=None) is None + + +def test_banner_match_still_works_with_probe(http_signatures): + sig = match_service(9090, "tcp", banner="prometheus 2.x", + http_signals={"title": "x", "headers": {}}) + assert sig["service_name"] == "Prometheus" + + +def test_match_port_alias_has_no_http(http_signatures): + # match_port() is the probe-less alias → http_regex entry degrades to port-only + sig = match_port(8096, "tcp") + assert sig["service_name"] == "HTTP" + + +def test_fingerprint_ports_uses_http_signals(http_signatures): + results = fingerprint_ports([ + {"port": 8096, "protocol": "tcp", "banner": None, + "http_signals": {"title": "Jellyfin", "headers": {}}}, + ]) + assert results[0]["service_name"] == "Jellyfin"