feat: Phase 3 & 4 — monitoring, discovery, polish, deployment
Phase 3 — Discovery & Monitoring: - Network scanner: nmap wrapper + mock fallback, fingerprint service (35 signatures) - Status checker: ping/http/https/tcp/ssh/prometheus/health per-node checks - APScheduler: status checks every 60s, WebSocket broadcast - WebSocket /ws/status: live node status updates to frontend - Sidebar panels: Pending Devices, Hidden Devices, Scan History - Auth token persisted to localStorage (survive page refresh) - 24 new backend tests (scan flow + status_checker) Phase 4 — Polish & Deployment: - Auto-layout: Dagre hierarchical TB via Toolbar button - Export PNG: html-to-image download via Toolbar button - Scan config modal: CIDR ranges + check interval, GET/POST /api/v1/scan/config - Dockerfile.backend (Python 3.13 slim + nmap), Dockerfile.frontend (nginx) - docker-compose.yml with data volume and NET_RAW cap for ping - scripts/lxc-install.sh: Proxmox VE systemd bootstrap - README.md: quick-start, config reference, stack overview
This commit is contained in:
@@ -0,0 +1,67 @@
|
||||
"""Match nmap scan results against service_signatures.json."""
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
_SIGNATURES: list[dict] | None = None
|
||||
|
||||
|
||||
def _load() -> list[dict]:
|
||||
global _SIGNATURES
|
||||
if _SIGNATURES is None:
|
||||
path = Path(__file__).parent.parent.parent / "data" / "service_signatures.json"
|
||||
with open(path) as f:
|
||||
_SIGNATURES = json.load(f)
|
||||
return _SIGNATURES
|
||||
|
||||
|
||||
def match_port(port: int, protocol: str, banner: str | None = None) -> dict | 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 banner and not re.search(sig["banner_regex"], banner, re.IGNORECASE):
|
||||
continue
|
||||
return sig
|
||||
return None
|
||||
|
||||
|
||||
def fingerprint_ports(open_ports: list[dict]) -> list[dict]:
|
||||
"""
|
||||
Given a list of {port, protocol, banner?} 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"))
|
||||
if sig:
|
||||
results.append({
|
||||
"port": p["port"],
|
||||
"protocol": p.get("protocol", "tcp"),
|
||||
"service_name": sig["service_name"],
|
||||
"icon": sig.get("icon"),
|
||||
"category": sig.get("category"),
|
||||
})
|
||||
else:
|
||||
results.append({
|
||||
"port": p["port"],
|
||||
"protocol": p.get("protocol", "tcp"),
|
||||
"service_name": "unknown_service",
|
||||
"icon": None,
|
||||
"category": None,
|
||||
})
|
||||
return results
|
||||
|
||||
|
||||
def suggest_node_type(open_ports: list[dict]) -> str:
|
||||
"""Suggest a node type based on the most specific matched signature."""
|
||||
priority = ["proxmox", "nas", "router", "lxc", "vm", "server", "ap", "iot", "switch"]
|
||||
found: set[str] = set()
|
||||
for p in open_ports:
|
||||
sig = match_port(p["port"], p.get("protocol", "tcp"))
|
||||
if sig and sig.get("suggested_node_type"):
|
||||
found.add(sig["suggested_node_type"])
|
||||
for t in priority:
|
||||
if t in found:
|
||||
return t
|
||||
return "generic"
|
||||
Reference in New Issue
Block a user