""" Analisis de topologia y vista desde dispositivo. El portatil no ejecuta codigo DENTRO del NVR/router/switch. Si reconstruye que deberia ver cada NVR en una LAN plana y por que un NVR concreto veria menos camaras (multicast, subred, puertos, WiFi). """ from __future__ import annotations from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from typing import Dict, List, Optional, Set from .models import Device, NetInfo, classify_device, now_str from .onvif_rtsp import camera_service_probe from .ports import scan_ports from .ws_discovery import ws_discovery_probe_both @dataclass class Finding: severity: str # CRITICAL / WARNING / INFO code: str title: str detail: str actions: List[str] = field(default_factory=list) @dataclass class DeviceViewpoint: device_ip: str device_type: str role: str same_subnet_cameras: List[str] = field(default_factory=list) reachable_unicast: List[str] = field(default_factory=list) discoverable_onvif: List[str] = field(default_factory=list) missing_vs_scan: List[str] = field(default_factory=list) missing_discovery: List[str] = field(default_factory=list) estimated_visible: int = 0 capacity_hint: str = "" findings: List[Finding] = field(default_factory=list) @dataclass class TopologyReport: generated: str net: Optional[NetInfo] inventory: Dict[str, List[Device]] discovery_ips: List[str] discovery_ok: bool discovery_error: str discovery_note: str viewpoints: List[DeviceViewpoint] findings: List[Finding] summary: str limitation_note: str def as_text(self) -> str: lines = [ "ANALISIS DE TOPOLOGIA / VISTA DESDE DISPOSITIVOS", "=" * 56, f"Generado: {self.generated}", "", self.limitation_note, "", "RESUMEN", "-" * 40, self.summary, "", "INVENTARIO POR ROL", "-" * 40, ] for role, devs in self.inventory.items(): if not devs: continue lines.append(f"[{role}] ({len(devs)})") for d in devs: ports = ",".join(str(p) for p, o in sorted(d.ports.items()) if o) or "-" lat = d.latency_ms if d.latency_ms is not None else "-" lines.append( f" {d.ip:15} {d.mac or '-':17} {(d.vendor or '-'):12} " f"{d.device_type:14} lat={lat} ports={ports}" ) lines.append("") lines.append("ONVIF WS-DISCOVERY (equivalente a Buscar del NVR)") lines.append("-" * 40) if self.discovery_ok: lines.append(f"Respuestas: {len(self.discovery_ips)} IP(s)") lines.append(", ".join(self.discovery_ips) if self.discovery_ips else "(ninguna)") else: lines.append(f"Fallo discovery: {self.discovery_error or 'sin respuestas'}") lines.append(self.discovery_note) lines.append("") lines.append("VISTA DESDE CADA NVR / ROUTER CLAVE") lines.append("-" * 40) for vp in self.viewpoints: lines.append(f"\n>>> Desde {vp.role} {vp.device_ip} ({vp.device_type})") lines.append(f" Camaras misma /24 (escaneo): {len(vp.same_subnet_cameras)}") lines.append(f" Con ping/puerto desde PC (unicast): {len(vp.reachable_unicast)}") lines.append(f" Anunciadas por WS-Discovery: {len(vp.discoverable_onvif)}") lines.append(f" Estimacion visible en Buscar NVR: {vp.estimated_visible}") if vp.capacity_hint: lines.append(f" {vp.capacity_hint}") if vp.missing_discovery: shown = ", ".join(vp.missing_discovery[:20]) more = "..." if len(vp.missing_discovery) > 20 else "" lines.append( f" Ping OK pero NO en discovery ({len(vp.missing_discovery)}): {shown}{more}" ) if vp.missing_vs_scan: shown = ", ".join(vp.missing_vs_scan[:20]) lines.append( f" En inventario pero no alcanzables ahora ({len(vp.missing_vs_scan)}): {shown}" ) for f in vp.findings: lines.append(f" [{f.severity}] {f.title}") lines.append(f" {f.detail}") for a in f.actions: lines.append(f" -> {a}") lines.append("") lines.append("HALLAZGOS GLOBALES") lines.append("-" * 40) if not self.findings: lines.append("Sin hallazgos criticos de topologia en esta pasada.") for f in self.findings: lines.append(f"[{f.severity}] {f.title}") lines.append(f" {f.detail}") for a in f.actions: lines.append(f" -> {a}") lines.append("") return "\n".join(lines) def refine_role(d: Device, gateway: str = "") -> str: t = d.device_type v = (d.vendor or "").lower() ports = {p for p, o in (d.ports or {}).items() if o} if d.ip == gateway: return "Router/Gateway" if t == "NVR" or 37777 in ports: return "NVR" if t == "Camara" or 554 in ports: if d.latency_ms is not None and d.latency_ms >= 8: return "Camara IP (posible WiFi)" return "Camara IP (posible LAN/PoE)" if t == "Router/AP" or any(x in v for x in ("tenda", "tp-link", "askey", "huawei", "cudy")): if "cudy" in v: return "AP/Router mesh (Cudy)" if "tenda" in v: return "AP/Router intermedio (Tenda)" if "tp-link" in v: return "AP/Repetidor (TP-Link)" if "askey" in v: return "Router Movistar (Askey)" return "Router/AP" if "switch" in (d.hostname or "").lower(): return "Switch" return t or "Desconocido" def same_l3_subnet(ip_a: str, ip_b: str, mask_bits: int = 24) -> bool: try: a = [int(x) for x in ip_a.split(".")] b = [int(x) for x in ip_b.split(".")] if mask_bits <= 16: return a[:2] == b[:2] return a[:3] == b[:3] except Exception: return False def _is_camera(d: Device) -> bool: if d.device_type == "Camara": return True ports = {p for p, o in (d.ports or {}).items() if o} return 554 in ports and 37777 not in ports def _is_nvr(d: Device) -> bool: if d.device_type == "NVR": return True ports = {p for p, o in (d.ports or {}).items() if o} v = (d.vendor or "").lower() if 37777 in ports and ("pni" in v or "dahua" in v or "xiongmai" in v): return True if 37777 in ports and 554 not in ports: return True return False def enrich_device_ports(devices: List[Device], workers: int = 32) -> None: key_ports = [80, 443, 554, 8000, 8080, 8899, 37777] def work(d: Device) -> None: if d.ports and any(d.ports.values()): missing = [p for p in key_ports if p not in d.ports] if missing: d.ports.update(scan_ports(d.ip, missing, timeout=1.0)) else: d.ports = scan_ports(d.ip, key_ports, timeout=1.0) d.device_type = classify_device(d.vendor, d.ports, d.hostname) with ThreadPoolExecutor(max_workers=workers) as ex: list(ex.map(work, devices)) def build_inventory(devices: List[Device], net: Optional[NetInfo]) -> Dict[str, List[Device]]: gw = net.gateway if net else "" groups: Dict[str, List[Device]] = { "NVR": [], "Camaras": [], "Routers/AP": [], "Gateway": [], "Otros": [], } for d in devices: role = refine_role(d, gw) if d.ip == gw: groups["Gateway"].append(d) elif _is_nvr(d) or role == "NVR": groups["NVR"].append(d) elif _is_camera(d) or "Camara" in role: groups["Camaras"].append(d) elif "Router" in role or "AP" in role or "Repetidor" in role: groups["Routers/AP"].append(d) else: groups["Otros"].append(d) return groups def analyze_viewpoint( focus: Device, cameras: List[Device], discovery_ips: Set[str], net: Optional[NetInfo], ) -> DeviceViewpoint: gw = net.gateway if net else "" role = refine_role(focus, gw) same = [c for c in cameras if same_l3_subnet(focus.ip, c.ip)] reachable = [ c.ip for c in same if c.status != "OFFLINE" and ( c.latency_ms is not None or c.status in ("ONLINE", "DEGRADADO") or any((c.ports or {}).values()) ) ] disc = [c.ip for c in same if c.ip in discovery_ips] missing_disc = [ip for ip in reachable if ip not in discovery_ips] missing_scan = [c.ip for c in same if c.status == "OFFLINE"] vp = DeviceViewpoint( device_ip=focus.ip, device_type=focus.device_type, role=role, same_subnet_cameras=[c.ip for c in same], reachable_unicast=reachable, discoverable_onvif=disc, missing_vs_scan=missing_scan, missing_discovery=missing_disc, estimated_visible=len(disc), ) if _is_nvr(focus): n = len(same) vp.capacity_hint = ( f"Camaras en misma subred que este NVR: {n}. " "NVR PNI IP816 suele tener 16 canales IP; si hay mas camaras que canales " "o varios NVR, cada uno solo montara un subconjunto." ) if n >= 8 and len(disc) < max(1, int(n * 0.7)) and len(reachable) >= max(1, int(n * 0.75)): vp.findings.append( Finding( "CRITICAL", "NVR_PARTIAL_DISCOVERY", f"El NVR {focus.ip} probablemente solo Buscaria ~{len(disc)} de {n} camaras", ( f"Hay {n} camaras en la misma /24, {len(reachable)} responden por unicast, " f"pero solo {len(disc)} anuncian ONVIF por multicast. Un NVR con Buscar/ONVIF " f"vera aproximadamente esas {len(disc)}. Las demas: anadir por IP fija." ), [ "En el NVR: anadir manualmente las IPs que faltan (no solo Buscar).", "Comprobar aislamiento WiFi / filtro multicast en switch o AP.", "Unificar camaras WiFi en un solo AP (Tenda), sin SSID _EXT.", "Verificar que este NVR esta por cable en el mismo switch que el resto.", ], ) ) elif n >= 8 and len(reachable) < max(1, int(n * 0.7)): vp.findings.append( Finding( "CRITICAL", "NVR_PARTIAL_REACH", f"Muchas camaras de la subred de {focus.ip} no responden ahora", ( f"Solo {len(reachable)}/{n} camaras alcanzables. Si el circuito parece cerrado " "pero faltan hosts: segmento L2 distinto, WiFi caido, IP incorrecta, " "o el portatil no esta en el mismo dominio de difusion que ese NVR." ), [ "Conecte el portatil por cable al MISMO switch que ese NVR y vuelva a escanear.", "Compare MAC/ARP de las camaras que faltan en clientes del Askey/Tenda.", "Monitorice 15 min: si caen por WiFi, el NVR tambien las perdera.", ], ) ) if cameras and not same_l3_subnet(focus.ip, cameras[0].ip): # solo si NINGUNA camara comparte subred if not same: vp.findings.append( Finding( "CRITICAL", "NVR_SUBNET_MISMATCH", f"NVR {focus.ip} no comparte /24 con camaras", "Sin enrutamiento entre subredes, el NVR no vera esas camaras.", ["Ponga NVR y camaras en 192.168.1.x con la misma mascara y gateway."], ) ) return vp def run_topology_analysis( devices: List[Device], net: Optional[NetInfo], enrich: bool = True, discovery_sec: float = 5.0, probe_services: bool = False, ) -> TopologyReport: devices = list(devices) if enrich and devices: enrich_device_ports(devices) inventory = build_inventory(devices, net) cams = inventory["Camaras"] nvrs = inventory["NVR"] if not nvrs: for d in list(devices): if (d.ports or {}).get(37777): d.device_type = "NVR" if d not in nvrs: nvrs.append(d) if d in inventory["Otros"]: inventory["Otros"].remove(d) if d not in inventory["NVR"]: inventory["NVR"].append(d) local_ip = net.ip if net else "" disc = ws_discovery_probe_both(local_ip=local_ip, listen_sec=discovery_sec) discovery_set: Set[str] = set(disc.get("ips") or []) if probe_services: sample = [c for c in cams if c.ip not in discovery_set][:8] with ThreadPoolExecutor(max_workers=8) as ex: futs = {ex.submit(camera_service_probe, c.ip): c for c in sample} for fut in as_completed(futs): c = futs[fut] try: svc = fut.result() if svc.get("onvif", {}).get("ok") or svc.get("rtsp", {}).get("ok"): c.notes = (c.notes + " | ONVIF/RTSP unicast OK").strip(" |") except Exception: pass findings: List[Finding] = [] viewpoints: List[DeviceViewpoint] = [] cam_ips = {c.ip for c in cams} disc_cams = discovery_set & cam_ips if cams and len(cams) >= 6 and len(disc_cams) < max(1, int(len(cams) * 0.6)): findings.append( Finding( "CRITICAL", "MULTICAST_GAP", f"WS-Discovery solo ve {len(disc_cams)}/{len(cams)} camaras del inventario", ( "Esto explica un NVR que solo encuentra la mitad: el Buscar del NVR usa " "el mismo multicast. El ping unicast puede funcionar y aun asi Buscar falla. " "Causas tipicas: aislamiento de clientes WiFi, IGMP snooping agresivo, " "AP/repetidor que no reenvia multicast, o camaras en otro segmento." ), [ "Desactive aislamiento AP / client isolation en Tenda y router.", "Pruebe Buscar en el NVR y compare con discovery de esta app.", "Anada por IP fija las camaras que faltan en discovery.", "Evite repetidor _EXT para CCTV; un solo AP cableado.", ], ) ) if len(nvrs) >= 2: findings.append( Finding( "WARNING", "MULTI_NVR", f"Hay {len(nvrs)} NVR en la misma red", ( "Es normal (un NVR por TV). Cada uno debe tener canales anadidos " "manualmente o por busqueda. Si un NVR ve menos, compare su vista " "en esta herramienta con la de otro NVR." ), [ "Ejecute analisis de topologia y compare estimacion por NVR.", "Todos los NVR por cable al mismo switch (no WiFi).", "Evite saturar: muchos streams 5MP a 8 NVR = cortes.", ], ) ) routers = inventory["Routers/AP"] if len(routers) + (1 if inventory["Gateway"] else 0) >= 2: findings.append( Finding( "WARNING", "MULTI_AP", "Varios routers/AP detectados", ( "Routers intermedios o repetidores pueden partir el dominio de difusion " "o filtrar multicast. Un NVR en el switch principal puede no descubrir " "camaras colgadas solo del WiFi del repetidor." ), [ "Tenda solo modo AP (DHCP off), cable LAN al Askey.", "Camaras solo en SSID del Tenda, no en _EXT.", "Un unico DHCP: Askey.", ], ) ) dhcp = net.dhcp_server if net else "" gw = net.gateway if net else "" if dhcp and gw and dhcp != gw: findings.append( Finding( "CRITICAL", "DUAL_DHCP", "Servidor DHCP distinto del gateway", f"DHCP={dhcp} gateway={gw}. Dos equipos repartiendo red -> IPs inconsistentes " "y cada NVR puede ver un subconjunto distinto de camaras.", ["Deje un solo DHCP (Askey). Desactive DHCP en Tenda y repetidor."], ) ) for nvr in nvrs: viewpoints.append(analyze_viewpoint(nvr, cams, discovery_set, net)) if net and net.gateway: gw_dev = next((d for d in devices if d.ip == net.gateway), None) if not gw_dev: gw_dev = Device(ip=net.gateway, device_type="Router/AP", hostname="gateway") viewpoints.append(analyze_viewpoint(gw_dev, cams, discovery_set, net)) n_nvr = len(nvrs) n_cam = len(cams) worst = None for vp in viewpoints: if vp.role == "NVR" and (worst is None or vp.estimated_visible < worst.estimated_visible): worst = vp summary_parts = [ f"Inventario: {n_cam} camaras, {n_nvr} NVR, {len(routers)} router/AP.", f"WS-Discovery (Buscar): {len(disc_cams)} camaras anunciadas de {n_cam} inventariadas.", ] if worst and n_cam: summary_parts.append( f"NVR con peor estimacion de busqueda: {worst.device_ip} " f"-> ~{worst.estimated_visible}/{len(worst.same_subnet_cameras)} via discovery " f"({len(worst.reachable_unicast)} alcanzables por unicast)." ) if any(f.severity == "CRITICAL" for f in findings): summary_parts.append("Estado topologia: PROBLEMA CRITICO (ver hallazgos).") elif findings: summary_parts.append("Estado topologia: ADVERTENCIA.") else: summary_parts.append("Estado topologia: sin brechas graves en esta pasada.") limitation = ( "LIMITACION: esta app no se ejecuta dentro del NVR, del Askey ni del switch. " "Se coloca en la misma LAN y reconstruye (1) inventario unicast, " "(2) lo que un NVR veria con Buscar/ONVIF multicast, (3) diferencias por NVR/subred. " "Es lo maximo verificable sin agente en cada aparato. " "Para coincidir con la pantalla de un NVR: conecte el PC por cable al mismo switch " "que ese NVR, ejecute este analisis y compare con Buscar en el NVR." ) for vp in viewpoints: for f in vp.findings: if f.severity == "CRITICAL": findings.append(f) return TopologyReport( generated=now_str(), net=net, inventory=inventory, discovery_ips=list(disc.get("ips") or []), discovery_ok=bool(disc.get("ok")), discovery_error=str(disc.get("error") or ""), discovery_note=str(disc.get("note") or ""), viewpoints=viewpoints, findings=findings, summary=" ".join(summary_parts), limitation_note=limitation, )