"""Pipeline fingerprint_web_stack. One-shot que materializa el flujo "averiguar la tecnologia web (stack) de una URL" estilo Wappalyzer: hace el fetch HTTP de las senales (cabeceras, HTML, cookies, titulo, servidor) y matchea las firmas para devolver las tecnologias detectadas (servidor, lenguaje, CMS, frameworks JS, librerias, analytics, CDN, e-commerce, WAF). Opcionalmente archiva la evidencia en OSINT. Convierte el patron de 2 llamadas (fetch_http_fingerprint -> detect_web_tech) en una sola invocacion. Compone funciones del registry del dominio cybersecurity; no reescribe ninguna logica de fetch, matching de firmas ni persistencia. Funciones del registry compuestas (importadas, no reimplementadas): fetch_http_fingerprint, detect_web_tech, save_scan_to_osint """ from urllib.parse import urlparse from cybersecurity import ( fetch_http_fingerprint, detect_web_tech, save_scan_to_osint, ) def _build_raw( url: str, final_url: str, status_code: int, server: str, title: str, technologies: list[dict], ) -> str: """Construye una tabla legible TECNOLOGIA/CATEGORIA/VERSION/CONFIDENCE para evidencia. NO incluye el HTML entero ni valores de cookie: solo metadatos de respuesta y la matriz de tecnologias detectadas. Args: url: URL solicitada. final_url: URL final tras redirects. status_code: codigo HTTP de la respuesta. server: cadena del servidor (cabecera Server), puede ser "". title: titulo de la pagina, puede ser "". technologies: lista de dicts de tecnologia (ver fingerprint_web_stack). Returns: Bloque de texto multi-linea con cabecera y una fila por tecnologia. """ header_lines = [ f"# fingerprint_web_stack {url}", "", f"url: {url}", f"final_url: {final_url}", f"status_code: {status_code}", f"server: {server or '-'}", f"title: {title or '-'}", "", ] cols = f"{'TECHNOLOGY':<24}{'CATEGORY':<22}{'VERSION':<14}CONFIDENCE" lines = header_lines + [cols] for t in technologies: name = str(t.get("name", "")) category = str(t.get("category", "")) version = str(t.get("version") or "") confidence = str(t.get("confidence", "")) lines.append(f"{name:<24}{category:<22}{version:<14}{confidence}") if not technologies: lines.append("(no technologies detected)") return "\n".join(lines) def _target_from_url(url: str, final_url: str) -> str: """Deriva el target (host) para el archivado OSINT a partir de la URL. Prefiere el host de la URL solicitada; si no se puede parsear, cae al host de la URL final tras redirects; si tampoco, devuelve la cadena cruda. Args: url: URL solicitada. final_url: URL final tras redirects. Returns: El host (sin esquema ni path), o la URL cruda si no se pudo extraer. """ for candidate in (url, final_url): if not candidate: continue try: host = urlparse(candidate).hostname except ValueError: host = None if host: return host return (url or final_url or "unknown").strip() def fingerprint_web_stack( url: str, timeout_s: float = 15.0, verify_tls: bool = True, max_html_bytes: int = 500_000, save: bool = True, ) -> dict: """Detecta la tecnologia web (stack) de una URL en un solo paso (estilo Wappalyzer). Compone, en una sola invocacion: 1. ``fetch_http_fingerprint(url, ...)`` para recoger las senales crudas de la respuesta (cabeceras, HTML, cookies, titulo, servidor). 2. ``detect_web_tech(headers, html, cookies, final_url)`` (PURA) para matchear esas senales contra la tabla de firmas y obtener las tecnologias detectadas. 3. Si ``save`` es True, archiva una tabla de evidencia en OSINT via ``save_scan_to_osint`` con ``scan_type="web_tech"`` (target = host de la URL). Nunca lanza excepciones: cualquier fallo se refleja en la clave ``status`` del dict devuelto. Args: url: URL objetivo. Sin esquema se asume https:// (fallback a http://), tal como hace fetch_http_fingerprint. timeout_s: timeout de la peticion HTTP en segundos. Default 15.0. Se pasa tal cual a fetch_http_fingerprint. verify_tls: si False, no verifica el certificado TLS (inseguro, solo para hosts propios con cert self-signed). Default True. Se pasa a fetch_http_fingerprint. max_html_bytes: corta el HTML leido a este tamano para no descargar megas. Default 500_000 (500 KB). Se pasa a fetch_http_fingerprint. save: si True (default) archiva la evidencia en OSINT via save_scan_to_osint con scan_type="web_tech"; si False solo ejecuta el fetch + matching y no toca el vault ni el service osint_db. Politica recon: todo scan se archiva. Si el sink falla, el resultado degrada sin romper (saved.status="error"). Returns: dict de estado. Nunca lanza. ok:: { "status": "ok", "url": , "final_url": , "status_code": int, "server": str, # cabecera Server, "" si no hay "title": str, # titulo de la pagina, "" si no hay "technologies": [ # tal cual de detect_web_tech {"name", "category", "version", "confidence", "evidence"}, ... ], "by_category": {: [, ...], ...}, "count": int, "saved": | None, "raw": "# fingerprint_web_stack ...\nTECHNOLOGY ...", } error (el fetch HTTP fallo: host no resuelve, conexion rechazada, timeout):: {"status": "error", "stage": "fetch", "url": , "fetch": } """ # 1. Fetch de senales. Si el fetch falla del todo, propagamos sin continuar. fp = fetch_http_fingerprint( url, timeout_s=timeout_s, verify_tls=verify_tls, max_html_bytes=max_html_bytes, ) if fp.get("status") != "ok": return { "status": "error", "stage": "fetch", "url": url, "fetch": fp, } final_url = fp.get("final_url", "") or "" status_code = fp.get("status_code", 0) server = fp.get("server") or "" title = fp.get("title") or "" # 2. Matching de firmas (puro): no toca red, solo aplica regex deterministas. detection = detect_web_tech( fp.get("headers") or {}, html=fp.get("html") or "", cookies=fp.get("cookies") or [], final_url=final_url, ) technologies = detection.get("technologies", []) by_category = detection.get("by_category", {}) count = detection.get("count", len(technologies)) raw = _build_raw(url, final_url, status_code, server, title, technologies) # 3. Archiva la evidencia en OSINT si procede (degrada sin romper). saved = None if save: target = _target_from_url(url, final_url) summary = { "count": count, "by_category": by_category, "server": server, "status_code": status_code, } saved = save_scan_to_osint( target, "web_tech", raw, summary=summary, tool="fingerprint_web_stack", ) return { "status": "ok", "url": url, "final_url": final_url, "status_code": status_code, "server": server, "title": title, "technologies": technologies, "by_category": by_category, "count": count, "saved": saved, "raw": raw, } def _parse_cli(argv: list[str]) -> dict: """Parsea los args de CLI: [--no-save] [--no-verify-tls]. Devuelve un dict de kwargs para fingerprint_web_stack. """ positional: list[str] = [] save = True verify_tls = True for arg in argv: if arg == "--no-save": save = False elif arg == "--no-verify-tls": verify_tls = False else: positional.append(arg) return {"positional": positional, "save": save, "verify_tls": verify_tls} if __name__ == "__main__": import sys parsed = _parse_cli(sys.argv[1:]) positional = parsed["positional"] target_url = positional[0] if len(positional) >= 1 else "https://example.com" try: result = fingerprint_web_stack( target_url, verify_tls=parsed["verify_tls"], save=parsed["save"], ) print("status:", result.get("status")) if result.get("status") == "ok": print(f"url: {result['url']} -> {result['final_url']} ({result['status_code']})") print("server:", result["server"] or "-") print("--- technologies ---") print(result["raw"]) saved = result.get("saved") or {} if saved: print("note_path:", saved.get("note_path")) print("registered:", saved.get("registered")) else: print("error:", result.get("fetch", {}).get("error")) except Exception as exc: # noqa: BLE001 - smoke nunca debe romper print("smoke exception (tolerada):", repr(exc))