fix: resolve all 64 mypy errors across backend
- 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
This commit is contained in:
@@ -2,11 +2,12 @@
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
_SIGNATURES: list[dict] | None = None
|
||||
_SIGNATURES: list[dict[str, Any]] | None = None
|
||||
|
||||
|
||||
def _load() -> list[dict]:
|
||||
def _load() -> list[dict[str, Any]]:
|
||||
global _SIGNATURES
|
||||
if _SIGNATURES is None:
|
||||
path = Path(__file__).parent.parent.parent / "data" / "service_signatures.json"
|
||||
@@ -15,7 +16,7 @@ def _load() -> list[dict]:
|
||||
return _SIGNATURES
|
||||
|
||||
|
||||
def match_port(port: int, protocol: str, banner: str | None = None) -> dict | None:
|
||||
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:
|
||||
@@ -26,7 +27,7 @@ def match_port(port: int, protocol: str, banner: str | None = None) -> dict | No
|
||||
return None
|
||||
|
||||
|
||||
def fingerprint_ports(open_ports: list[dict]) -> list[dict]:
|
||||
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.
|
||||
@@ -53,7 +54,7 @@ def fingerprint_ports(open_ports: list[dict]) -> list[dict]:
|
||||
return results
|
||||
|
||||
|
||||
def suggest_node_type(open_ports: list[dict]) -> str:
|
||||
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()
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
import logging
|
||||
import socket
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
@@ -11,14 +12,14 @@ from app.services.fingerprint import fingerprint_ports, suggest_node_type
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
try:
|
||||
import nmap # type: ignore[import-untyped]
|
||||
import nmap
|
||||
_NMAP_AVAILABLE = True
|
||||
except ImportError:
|
||||
_NMAP_AVAILABLE = False
|
||||
logger.warning("python-nmap not available — scanner will run in mock mode")
|
||||
|
||||
|
||||
def _nmap_scan(target: str) -> list[dict]:
|
||||
def _nmap_scan(target: str) -> list[dict[str, Any]]:
|
||||
"""Run nmap -sV --open on target, return list of host dicts."""
|
||||
if not _NMAP_AVAILABLE:
|
||||
return _mock_scan(target)
|
||||
@@ -64,13 +65,13 @@ def _extract_os(nm: object, host: str) -> str | None:
|
||||
try:
|
||||
osmatch = nm[host].get("osmatch", []) # type: ignore[index]
|
||||
if osmatch:
|
||||
return osmatch[0]["name"]
|
||||
return str(osmatch[0]["name"])
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def _mock_scan(target: str) -> list[dict]:
|
||||
def _mock_scan(target: str) -> list[dict[str, Any]]:
|
||||
"""Return fake results for dev/test environments without nmap."""
|
||||
return [
|
||||
{
|
||||
|
||||
@@ -3,13 +3,14 @@ import asyncio
|
||||
import logging
|
||||
import socket
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def check_node(check_method: str, target: str | None, ip: str | None) -> dict:
|
||||
async def check_node(check_method: str, target: str | None, ip: str | None) -> dict[str, Any]:
|
||||
"""
|
||||
Run the appropriate check and return {status, response_time_ms}.
|
||||
status is one of: online, offline, unknown.
|
||||
|
||||
Reference in New Issue
Block a user