Pouzor
974e782057
- Add dict[str, Any] / list[Any] type params throughout (fingerprint, models, schemas, scanner, status_checker) - Add return type annotations to all route functions (nodes, edges, canvas, scan, auth, status, main) - Fix no-any-return in security.py: cast pwd/jwt results to bool/str explicitly - Fix canvas.py: use model_validate() for NodeResponse/EdgeResponse, rename db_node/db_edge upsert vars - Fix scheduler.py: rename 'result' → 'check_result' to avoid type collision - Fix get_db() return type: AsyncGenerator[AsyncSession, None] - Add types-PyYAML for yaml import stubs - Fix scanner.py: remove unnecessary type: ignore comment (nmap has stubs) - Fix scan.py: wrap scalars().all() with list() for Sequence→list compatibility
69 lines
2.3 KiB
Python
69 lines
2.3 KiB
Python
"""Match nmap scan results against service_signatures.json."""
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
_SIGNATURES: list[dict[str, Any]] | None = None
|
|
|
|
|
|
def _load() -> list[dict[str, Any]]:
|
|
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[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 banner and not re.search(sig["banner_regex"], banner, re.IGNORECASE):
|
|
continue
|
|
return sig
|
|
return None
|
|
|
|
|
|
def fingerprint_ports(open_ports: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""
|
|
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, Any]]) -> 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"
|