101 lines
2.9 KiB
Python
101 lines
2.9 KiB
Python
"""httpx — HTTP probing: status, title, tech, webserver, TLS. Output: ``-json -td``."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from ..schema import HttpInfo, TlsInfo
|
|
from .base import parse_jsonl, run_tool
|
|
|
|
BINARY = "httpx"
|
|
|
|
|
|
@dataclass
|
|
class HttpProbe:
|
|
host: str | None
|
|
ip: str | None
|
|
port: int | None
|
|
info: HttpInfo
|
|
|
|
|
|
def _port_of(rec: dict) -> int | None:
|
|
port = rec.get("port")
|
|
try:
|
|
return int(port)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _tls_of(rec: dict) -> TlsInfo | None:
|
|
tls = rec.get("tls")
|
|
if not isinstance(tls, dict):
|
|
return None
|
|
issuer = tls.get("issuer_cn") or tls.get("issuer_common_name")
|
|
if not issuer:
|
|
org = tls.get("issuer_org")
|
|
if isinstance(org, list) and org:
|
|
issuer = org[0]
|
|
elif isinstance(org, str):
|
|
issuer = org
|
|
return TlsInfo(
|
|
subject_cn=tls.get("subject_cn") or tls.get("subject_common_name"),
|
|
issuer=issuer,
|
|
not_after=tls.get("not_after"),
|
|
)
|
|
|
|
|
|
def normalize(raw: str) -> list[HttpProbe]:
|
|
"""Map httpx -json lines into HttpProbe records keyed by host/ip/port."""
|
|
out: list[HttpProbe] = []
|
|
for rec in parse_jsonl(raw):
|
|
tech = rec.get("tech") or rec.get("technologies") or []
|
|
if isinstance(tech, str):
|
|
tech = [tech]
|
|
info = HttpInfo(
|
|
url=rec.get("url"),
|
|
status=rec.get("status_code") or rec.get("status-code"),
|
|
title=rec.get("title"),
|
|
webserver=rec.get("webserver"),
|
|
technologies=[str(t) for t in tech],
|
|
tls=_tls_of(rec),
|
|
)
|
|
out.append(
|
|
HttpProbe(
|
|
host=(rec.get("input") or rec.get("host") or "").lower() or None,
|
|
ip=rec.get("host") if _looks_like_ip(rec.get("host")) else rec.get("a", [None])[0] if isinstance(rec.get("a"), list) else None,
|
|
port=_port_of(rec),
|
|
info=info,
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
def _looks_like_ip(v) -> bool:
|
|
import ipaddress
|
|
|
|
if not isinstance(v, str):
|
|
return False
|
|
try:
|
|
ipaddress.ip_address(v)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def run(
|
|
targets: list[str],
|
|
*,
|
|
timeout: float = 180.0,
|
|
rate_limit: int | None = None,
|
|
ports: list[int] | None = None,
|
|
) -> tuple[list[HttpProbe], object]:
|
|
cmd = [BINARY, "-json", "-td", "-silent", "-title", "-tech-detect", "-web-server", "-status-code"]
|
|
# Without -p, httpx only probes 80/443. Feed it the ports discovered by
|
|
# naabu/nmap so web apps on non-standard ports (e.g. 5000, 8080) are probed.
|
|
if ports:
|
|
cmd += ["-p", ",".join(str(p) for p in sorted(set(ports)))]
|
|
if rate_limit:
|
|
cmd += ["-rate-limit", str(rate_limit)]
|
|
res = run_tool(cmd, timeout=timeout, input_text="\n".join(targets) + "\n")
|
|
return normalize(res.stdout), res
|