278 lines
9.6 KiB
Python
278 lines
9.6 KiB
Python
"""Extract endpoint and crackable hash artifacts from a capture context.
|
||
|
||
Extraction is independent of Django, tshark, and I/O. Active endpoint probing
|
||
and hash cracking remain isolated in ``analyzer.enrich``.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Optional
|
||
|
||
from .context import CaptureContext
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Artifact data models
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
@dataclass
|
||
class UrlArtifact:
|
||
"""A domain, host, or URL observed in the capture."""
|
||
|
||
value: str
|
||
kind: str = "unknown" # "dns" | "sni" | "http_host" | "payload"
|
||
src_ip: Optional[str] = None
|
||
dst_ip: Optional[str] = None
|
||
timestamp: Optional[float] = None
|
||
packet_number: Optional[int] = None
|
||
# Passive observations are preferred over active probing results.
|
||
observed_ports: set[int] = field(default_factory=set)
|
||
observed_tls: set[str] = field(default_factory=set)
|
||
|
||
def to_dict(self) -> dict:
|
||
return {
|
||
"value": self.value, "kind": self.kind,
|
||
"src_ip": self.src_ip, "dst_ip": self.dst_ip,
|
||
"timestamp": self.timestamp, "packet_number": self.packet_number,
|
||
"observed_ports": sorted(self.observed_ports),
|
||
"observed_tls": sorted(self.observed_tls),
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class HashArtifact:
|
||
"""A hash or crackable authentication artifact from the capture."""
|
||
|
||
value: str # Raw hash or complete hashcat line
|
||
hashcat_mode: Optional[int] = None
|
||
hash_format: Optional[str] = None # Human-readable format name
|
||
kind: str = "raw_hex" # "raw_hex" | "netntlmv2" | "krb5asrep" | "krb5tgs" | "wpa" | "digest"
|
||
account: Optional[str] = None
|
||
src_ip: Optional[str] = None
|
||
dst_ip: Optional[str] = None
|
||
timestamp: Optional[float] = None
|
||
packet_number: Optional[int] = None
|
||
|
||
def to_dict(self) -> dict:
|
||
return {
|
||
"value": self.value, "hashcat_mode": self.hashcat_mode,
|
||
"hash_format": self.hash_format, "kind": self.kind,
|
||
"account": self.account, "src_ip": self.src_ip, "dst_ip": self.dst_ip,
|
||
"timestamp": self.timestamp, "packet_number": self.packet_number,
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Hashcat mode detection
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
# Hex length to hashcat mode and format name.
|
||
_LEN_MODES = {
|
||
32: (0, "md5"),
|
||
40: (100, "sha1"),
|
||
56: (300, "sha224"),
|
||
64: (1400, "sha256"),
|
||
96: (6100, "sha384"),
|
||
128: (1700, "sha512"),
|
||
}
|
||
|
||
# Structured format name to hashcat mode.
|
||
_FORMAT_MODES = {
|
||
"netntlmv1": 5500,
|
||
"netntlmv2": 5600,
|
||
"krb5asrep": 18200,
|
||
"krb5tgs": 13100,
|
||
"wpa": 22000,
|
||
"eapol": 22000,
|
||
}
|
||
|
||
_NETNTLMV2_RE = re.compile(r"^[^:]+::[^:]*:[0-9a-fA-F]{16}:[0-9a-fA-F]{32}:[0-9a-fA-F]+$")
|
||
_NETNTLMV1_RE = re.compile(r"^[^:]+::[^:]*:[0-9a-fA-F]{48}:[0-9a-fA-F]{48}:[0-9a-fA-F]{16}$")
|
||
|
||
|
||
def identify_hash_mode(value: str, hint_format: Optional[str] = None):
|
||
"""Return a hashcat mode and format using structure before hex length."""
|
||
v = (value or "").strip()
|
||
if not v:
|
||
return None, None
|
||
|
||
if hint_format:
|
||
base = str(hint_format).split("(")[0].strip().lower()
|
||
if base in _FORMAT_MODES:
|
||
return _FORMAT_MODES[base], hint_format
|
||
|
||
low = v.lower()
|
||
if low.startswith("$krb5asrep$"):
|
||
return 18200, "krb5asrep"
|
||
if low.startswith("$krb5tgs$"):
|
||
return 13100, "krb5tgs"
|
||
if low.startswith("$wpapsk$") or low.startswith("wpa*") or low.startswith("$wpa$"):
|
||
return 22000, "wpa"
|
||
if _NETNTLMV2_RE.match(v):
|
||
return 5600, "NetNTLMv2"
|
||
if _NETNTLMV1_RE.match(v):
|
||
return 5500, "NetNTLMv1"
|
||
|
||
if all(c in "0123456789abcdefABCDEF" for c in v) and len(v) in _LEN_MODES:
|
||
return _LEN_MODES[len(v)]
|
||
return None, None
|
||
|
||
|
||
def _kind_from_mode(mode: Optional[int]) -> str:
|
||
return {
|
||
5500: "netntlmv1", 5600: "netntlmv2",
|
||
18200: "krb5asrep", 13100: "krb5tgs", 22000: "wpa",
|
||
}.get(mode, "raw_hex")
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Endpoint extraction
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
_URL_RE = re.compile(
|
||
r"\b(?:"
|
||
r"(?:https?://|www\.)(?:[a-zA-Z0-9.-]+\.[a-zA-Z]{2,6})" # URL or www host
|
||
r"|"
|
||
r"(?:[a-zA-Z0-9-]+\.){2,}[a-zA-Z]{2,6}" # Bare multi-label domain
|
||
r"|"
|
||
r"https?://\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}" # URL with IPv4 host
|
||
r")"
|
||
r"(?:/[a-zA-Z0-9._%/-]*)?\b"
|
||
)
|
||
|
||
|
||
# Endpoints are only trustworthy from cleartext application data. Scanning
|
||
# encrypted/binary payloads (TLS, QUIC, SSH, raw TCP) with a loose domain regex
|
||
# invents host-shaped noise from random bytes, so payload extraction is limited
|
||
# to protocols whose bytes are actually text.
|
||
_CLEARTEXT_PROTOS = {
|
||
"http", "http2", "telnet", "ftp", "ftp-data", "ftp_data", "smtp", "pop",
|
||
"pop3", "imap", "snmp", "tftp", "ldap", "sip", "irc", "syslog", "rtsp",
|
||
"xmpp", "mqtt", "urlencoded-form", "json", "xml", "data-text-lines",
|
||
}
|
||
|
||
_IPV4_RE = re.compile(r"^\d{1,3}(?:\.\d{1,3}){3}$")
|
||
|
||
|
||
def _looks_like_host(value: str) -> bool:
|
||
"""Reject byte noise that the URL regex mistakes for a bare domain."""
|
||
v = value.split("://")[-1].split("/")[0]
|
||
if _IPV4_RE.match(v):
|
||
return all(0 <= int(o) <= 255 for o in v.split("."))
|
||
if len(v) < 4 or ".." in v:
|
||
return False
|
||
labels = v.split(".")
|
||
if len(labels) < 2:
|
||
return False
|
||
tld = labels[-1]
|
||
if not tld.isalpha() or not (2 <= len(tld) <= 24):
|
||
return False
|
||
# At least one substantive label — kills noise like "4v.b.di" / "q.p.fh".
|
||
if not any(len(lbl) >= 3 for lbl in labels[:-1]):
|
||
return False
|
||
return True
|
||
|
||
|
||
def _normalize_host(value: Optional[str]) -> Optional[str]:
|
||
if not value:
|
||
return None
|
||
v = str(value).strip().lower()
|
||
v = v.rstrip(".")
|
||
return v or None
|
||
|
||
|
||
def extract_urls(ctx: CaptureContext) -> list[UrlArtifact]:
|
||
"""Collect endpoints and attach passively observed ports and TLS versions."""
|
||
by_value: dict[str, UrlArtifact] = {}
|
||
|
||
def add(value, kind, **kw) -> Optional[UrlArtifact]:
|
||
v = _normalize_host(value)
|
||
if not v:
|
||
return None
|
||
art = by_value.get(v)
|
||
if art is None:
|
||
art = UrlArtifact(value=v, kind=kind, **kw)
|
||
by_value[v] = art
|
||
return art
|
||
|
||
for q in ctx.dns_queries:
|
||
add(q.query_name, "dns", src_ip=q.src_ip, dst_ip=q.dst_ip,
|
||
timestamp=q.timestamp, packet_number=q.packet_number)
|
||
|
||
for t in ctx.tls_sessions:
|
||
if t.sni:
|
||
art = add(t.sni, "sni", src_ip=t.src_ip, dst_ip=t.dst_ip,
|
||
timestamp=t.timestamp, packet_number=t.packet_number)
|
||
if art:
|
||
if t.dst_port:
|
||
art.observed_ports.add(t.dst_port)
|
||
if t.version:
|
||
art.observed_tls.add(t.version)
|
||
|
||
for r in ctx.http_requests:
|
||
if r.host:
|
||
art = add(r.host, "http_host", src_ip=r.src_ip, dst_ip=r.dst_ip,
|
||
timestamp=r.timestamp, packet_number=r.packet_number)
|
||
if art and r.dst_port:
|
||
art.observed_ports.add(r.dst_port)
|
||
|
||
for p in ctx.text_payloads:
|
||
if (p.get("protocol") or "").lower() not in _CLEARTEXT_PROTOS:
|
||
continue
|
||
text = p.get("text") or ""
|
||
for m in _URL_RE.findall(text):
|
||
if not _looks_like_host(m):
|
||
continue
|
||
add(m, "payload", src_ip=p.get("src_ip"), dst_ip=p.get("dst_ip"),
|
||
timestamp=p.get("timestamp"), packet_number=p.get("packet_number"))
|
||
|
||
return list(by_value.values())
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Hash extraction
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
_HASH_HEX_RE = re.compile(r"\b[a-fA-F0-9]{32,64}\b")
|
||
|
||
|
||
def extract_hashes(ctx: CaptureContext) -> list[HashArtifact]:
|
||
"""Collect structured authentication material and raw hexadecimal hashes."""
|
||
out: list[HashArtifact] = []
|
||
seen: set[str] = set()
|
||
|
||
def add(value, *, mode, fmt, kind, **kw) -> None:
|
||
if not value or value in seen:
|
||
return
|
||
seen.add(value)
|
||
out.append(HashArtifact(value=value, hashcat_mode=mode, hash_format=fmt,
|
||
kind=kind, **kw))
|
||
|
||
# Structured material already reconstructed by the packet extractor.
|
||
for am in ctx.auth_material:
|
||
if getattr(am, "hashcat", None):
|
||
mode = am.hashcat_mode
|
||
fmt = am.hash_format
|
||
if mode is None:
|
||
mode, fmt = identify_hash_mode(am.hashcat, hint_format=fmt)
|
||
add(am.hashcat, mode=mode, fmt=fmt, kind=_kind_from_mode(mode),
|
||
account=am.account, src_ip=am.src_ip, dst_ip=am.dst_ip,
|
||
timestamp=am.timestamp, packet_number=am.packet_number)
|
||
|
||
# Raw 32–64 character hex values use a mode only when the length is known.
|
||
# Only cleartext payloads are scanned; TLS/QUIC ciphertext is full of 32–64
|
||
# char hex runs that are not hashes.
|
||
for p in ctx.text_payloads:
|
||
if (p.get("protocol") or "").lower() not in _CLEARTEXT_PROTOS:
|
||
continue
|
||
text = p.get("text") or ""
|
||
for m in _HASH_HEX_RE.findall(text):
|
||
mode, fmt = identify_hash_mode(m)
|
||
add(m.lower(), mode=mode, fmt=fmt, kind="raw_hex",
|
||
src_ip=p.get("src_ip"), dst_ip=p.get("dst_ip"),
|
||
timestamp=p.get("timestamp"), packet_number=p.get("packet_number"))
|
||
|
||
return out
|