144 lines
4.9 KiB
Python
144 lines
4.9 KiB
Python
"""Optionally enrich observed endpoints with DNS, port, and TLS checks.
|
|
|
|
Passive capture evidence remains authoritative. Active checks are opt-in and
|
|
limited to a small port set because operators must have authorization to scan.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Callable, Optional, Sequence, Union
|
|
from urllib.parse import urlparse
|
|
|
|
from ..artifacts import UrlArtifact
|
|
|
|
DEFAULT_PORTS = (22, 80, 443, 8000, 8080)
|
|
|
|
|
|
@dataclass
|
|
class UrlProbe:
|
|
url: str
|
|
host: str
|
|
resolved_ips: list[str] = field(default_factory=list)
|
|
open_ports: list[int] = field(default_factory=list)
|
|
tls_versions: list[str] = field(default_factory=list)
|
|
cipher_suites: list[str] = field(default_factory=list)
|
|
certificate: Optional[str] = None
|
|
active: bool = False
|
|
error: Optional[str] = None
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"url": self.url, "host": self.host,
|
|
"resolved_ips": self.resolved_ips, "open_ports": self.open_ports,
|
|
"tls_versions": self.tls_versions, "cipher_suites": self.cipher_suites,
|
|
"certificate": self.certificate, "active": self.active, "error": self.error,
|
|
}
|
|
|
|
|
|
def host_of(url: str) -> str:
|
|
"""Reduce a URL or domain value to its hostname."""
|
|
s = str(url).strip()
|
|
if "://" not in s:
|
|
s = "//" + s
|
|
netloc = urlparse(s).netloc or ""
|
|
host = netloc.split("@")[-1].split(":")[0]
|
|
return host or str(url).strip()
|
|
|
|
|
|
def probe_url(
|
|
artifact: Union[UrlArtifact, str],
|
|
*,
|
|
active: bool = False,
|
|
ports: Sequence[int] = DEFAULT_PORTS,
|
|
dns_resolver: Optional[Callable[[str], list]] = None,
|
|
port_scanner: Optional[Callable[[str, Sequence[int]], list]] = None,
|
|
tls_scanner: Optional[Callable[[str], dict]] = None,
|
|
) -> UrlProbe:
|
|
"""Return passive observations and optionally perform authorized active checks."""
|
|
if isinstance(artifact, UrlArtifact):
|
|
url = artifact.value
|
|
probe = UrlProbe(url=url, host=host_of(url))
|
|
probe.open_ports = sorted(artifact.observed_ports)
|
|
probe.tls_versions = sorted(artifact.observed_tls)
|
|
else:
|
|
url = str(artifact)
|
|
probe = UrlProbe(url=url, host=host_of(url))
|
|
|
|
if not active:
|
|
return probe
|
|
|
|
probe.active = True
|
|
host = probe.host
|
|
try:
|
|
resolver = dns_resolver or _default_dns_resolver
|
|
probe.resolved_ips = resolver(host) or []
|
|
except Exception as exc:
|
|
probe.error = f"DNS: {exc}"
|
|
|
|
try:
|
|
scanner = port_scanner or _default_port_scanner
|
|
found = scanner(host, ports) or []
|
|
probe.open_ports = sorted(set(probe.open_ports) | set(found))
|
|
except Exception as exc:
|
|
probe.error = (probe.error or "") + f" Ports: {exc}"
|
|
|
|
try:
|
|
tls = tls_scanner or _default_tls_scanner
|
|
info = tls(host) or {}
|
|
if info.get("tls_version"):
|
|
probe.tls_versions = sorted(set(probe.tls_versions) | {info["tls_version"]})
|
|
if info.get("cipher"):
|
|
probe.cipher_suites = sorted(set(probe.cipher_suites) | {info["cipher"]})
|
|
probe.certificate = info.get("certificate")
|
|
except Exception as exc:
|
|
probe.error = (probe.error or "") + f" TLS: {exc}"
|
|
|
|
return probe
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Lazy defaults are imported only when active checks are requested.
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def _default_dns_resolver(host: str) -> list[str]:
|
|
import dns.resolver # lazy
|
|
answers = dns.resolver.resolve(host, "A")
|
|
return [str(r) for r in answers]
|
|
|
|
|
|
def _default_port_scanner(host: str, ports: Sequence[int]) -> list[int]:
|
|
import nmap # lazy
|
|
nm = nmap.PortScanner()
|
|
port_arg = ",".join(str(p) for p in ports)
|
|
# TCP connect scanning works without raw-socket privileges; -Pn skips ICMP discovery.
|
|
nm.scan(hosts=host, arguments=f"-sT -Pn -p {port_arg} --max-retries 1 --host-timeout 20s")
|
|
open_ports: set[int] = set()
|
|
# nmap indexes results by resolved address, so iterate over every returned host.
|
|
for h in nm.all_hosts():
|
|
try:
|
|
for port in nm[h].all_tcp():
|
|
if nm[h]["tcp"][port]["state"] == "open":
|
|
open_ports.add(int(port))
|
|
except Exception:
|
|
continue
|
|
return sorted(open_ports)
|
|
|
|
|
|
def _default_tls_scanner(host: str) -> dict:
|
|
import socket
|
|
import ssl
|
|
ctx = ssl.create_default_context()
|
|
try:
|
|
with socket.create_connection((host, 443), timeout=5) as sock:
|
|
with ctx.wrap_socket(sock, server_hostname=host) as ssock:
|
|
cert = ssock.getpeercert()
|
|
cipher = ssock.cipher()
|
|
return {
|
|
"tls_version": ssock.version(),
|
|
"cipher": cipher[0] if cipher else None,
|
|
"certificate": "Present" if cert else "Not Present",
|
|
}
|
|
except Exception:
|
|
return {}
|