url/homelable
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add async HTTP prober for deep-scan service identification

Browse Source 
probe_port() GETs https:// then http:// for an open port and returns the page
<title> plus Server/X-Powered-By headers as identifying signals. Non-web ports
(SSH, DB, etc.) are skipped; body read is capped at 64KB; 3s timeout.
probe_open_ports() fans out over all open ports with a bounded semaphore.

ha-relevant: yes
This commit is contained in:
Pouzor
2026-06-23 13:41:32 +02:00
parent b20c47a0a5
commit 44e0029f2b
2 changed files with 203 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/http_probe.py
+85
View File
@@ -0,0 +1,85 @@
"""HTTP probe: GET a discovered port and extract identifying signals.
Used by the optional deep-scan mode to confirm what service sits behind an
open port, regardless of port number. Returns the page <title> plus a small
set of identifying response headers, which fingerprint.match_service() then
matches against signature http_regex fields.
"""
import asyncio
import logging
import re
from typing import Any
import httpx
logger = logging.getLogger(__name__)
# Headers that commonly carry the application name.
_SIGNAL_HEADERS = ("Server", "X-Powered-By")
# Cap how much body we read when hunting for <title> — avoids large downloads.
_MAX_BODY_BYTES = 64 * 1024
_TITLE_RE = re.compile(r"<title[^>]*>(.*?)</title>", re.IGNORECASE | re.DOTALL)
_PROBE_TIMEOUT = 3.0
# Ports we never bother probing over HTTP (not web services).
_NON_HTTP_PORTS = frozenset({22, 21, 23, 25, 53, 110, 143, 161, 162, 179, 445, 3306, 5432, 6379})
def _extract_title(body: str) -> str | None:
m = _TITLE_RE.search(body)
if not m:
return None
title = re.sub(r"\s+", " ", m.group(1)).strip()
return title or None
async def _probe_scheme(client: httpx.AsyncClient, url: str) -> dict[str, Any] | None:
try:
resp = await client.get(url, follow_redirects=True)
except (httpx.HTTPError, OSError):
return None
headers = {h: resp.headers[h] for h in _SIGNAL_HEADERS if h in resp.headers}
body = resp.text[:_MAX_BODY_BYTES] if resp.text else ""
title = _extract_title(body)
if not title and not headers:
return None
return {"title": title, "headers": headers}
async def probe_port(
ip: str, port: int, verify_tls: bool = False
) -> dict[str, Any] | None:
"""
GET https:// then http:// for a port and return {title, headers} or None.
None means the port did not answer HTTP or yielded no usable signal.
"""
if port in _NON_HTTP_PORTS:
return None
async with httpx.AsyncClient(verify=verify_tls, timeout=_PROBE_TIMEOUT) as client:
for scheme in ("https", "http"):
result = await _probe_scheme(client, f"{scheme}://{ip}:{port}/")
if result is not None:
return result
return None
async def probe_open_ports(
ip: str,
open_ports: list[dict[str, Any]],
verify_tls: bool = False,
concurrency: int = 50,
) -> list[dict[str, Any]]:
"""
Probe every open port for HTTP signals (option 2: probe all, match after).
Returns the same port dicts, each enriched with an http_signals key
(None when the port gave no HTTP signal).
"""
sem = asyncio.Semaphore(concurrency)
async def _one(p: dict[str, Any]) -> dict[str, Any]:
async with sem:
signals = await probe_port(ip, p["port"], verify_tls)
return {**p, "http_signals": signals}
return await asyncio.gather(*(_one(p) for p in open_ports))
backend/tests/test_http_probe.py
+118
View File
@@ -0,0 +1,118 @@
"""Tests for the HTTP probe used by deep-scan service identification."""
from unittest.mock import AsyncMock, patch
import httpx
import pytest
from app.services.http_probe import (
_extract_title,
probe_open_ports,
probe_port,
)
def _response(text: str = "", headers: dict | None = None, status: int = 200) -> httpx.Response:
return httpx.Response(status_code=status, text=text, headers=headers or {})
# ── _extract_title ──────────────────────────────────────────────────────────
def test_extract_title_basic():
assert _extract_title("<html><title>Jellyfin</title></html>") == "Jellyfin"
def test_extract_title_collapses_whitespace():
assert _extract_title("<title>\n My App\n</title>") == "My App"
def test_extract_title_missing():
assert _extract_title("<html><body>no title</body></html>") is None
def test_extract_title_case_insensitive():
assert _extract_title("<TITLE>Portainer</TITLE>") == "Portainer"
# ── probe_port ──────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_probe_port_reads_title():
with patch("httpx.AsyncClient.get", new=AsyncMock(return_value=_response("<title>Jellyfin</title>"))):
result = await probe_port("10.0.0.5", 8096)
assert result == {"title": "Jellyfin", "headers": {}}
@pytest.mark.asyncio
async def test_probe_port_reads_headers():
resp = _response("", headers={"Server": "nginx", "X-Powered-By": "Express"})
with patch("httpx.AsyncClient.get", new=AsyncMock(return_value=resp)):
result = await probe_port("10.0.0.5", 3000)
assert result["headers"] == {"Server": "nginx", "X-Powered-By": "Express"}
@pytest.mark.asyncio
async def test_probe_port_falls_back_to_http():
# https raises, http succeeds
calls = {"n": 0}
async def fake_get(self, url, **kw):
calls["n"] += 1
if url.startswith("https"):
raise httpx.ConnectError("tls fail")
return _response("<title>HTTP App</title>")
with patch("httpx.AsyncClient.get", new=fake_get):
result = await probe_port("10.0.0.5", 8080)
assert result["title"] == "HTTP App"
assert calls["n"] == 2 # tried https then http
@pytest.mark.asyncio
async def test_probe_port_no_signal_returns_none():
with patch("httpx.AsyncClient.get", new=AsyncMock(return_value=_response(""))):
result = await probe_port("10.0.0.5", 8080)
assert result is None
@pytest.mark.asyncio
async def test_probe_port_timeout_returns_none():
with patch("httpx.AsyncClient.get", new=AsyncMock(side_effect=httpx.TimeoutException("slow"))):
result = await probe_port("10.0.0.5", 8080)
assert result is None
@pytest.mark.asyncio
async def test_probe_port_skips_non_http_ports():
# SSH should never trigger an HTTP request
get = AsyncMock()
with patch("httpx.AsyncClient.get", new=get):
result = await probe_port("10.0.0.5", 22)
assert result is None
get.assert_not_called()
@pytest.mark.asyncio
async def test_probe_port_verify_tls_flag_passed():
with patch("app.services.http_probe.httpx.AsyncClient") as client_cls:
instance = client_cls.return_value.__aenter__.return_value
instance.get = AsyncMock(return_value=_response("<title>X</title>"))
await probe_port("10.0.0.5", 8443, verify_tls=True)
assert client_cls.call_args.kwargs["verify"] is True
# ── probe_open_ports ─────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_probe_open_ports_enriches_each_port():
async def fake_get(self, url, **kw):
if ":8096" in url:
return _response("<title>Jellyfin</title>")
return _response("")
ports = [{"port": 8096, "protocol": "tcp"}, {"port": 9999, "protocol": "tcp"}]
with patch("httpx.AsyncClient.get", new=fake_get):
result = await probe_open_ports("10.0.0.5", ports)
by_port = {p["port"]: p for p in result}
assert by_port[8096]["http_signals"]["title"] == "Jellyfin"
assert by_port[9999]["http_signals"] is None