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
2026-06-23 12:25:10 +02:00
parent be5771f7d0
commit b20c47a0a5
2 changed files with 173 additions and 13 deletions
@@ -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"],
@@ -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"