79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
"""Bot signature matching + reverse-DNS verification (Chapter 07).
|
|
|
|
Signatures are data (JSON), not hardcoded logic, so the list grows without
|
|
a code change. Verification does real DNS I/O — only ever called from the
|
|
background process-logs batch job (app/services/classification.py), never
|
|
synchronously inside a dashboard request, per Chapter 07's explicit rule.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import socket
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
SIGNATURES_PATH = Path(__file__).parent / "data" / "bot_signatures.json"
|
|
|
|
# Skip re-verifying the same IP more often than this (Ch07: DNS latency is
|
|
# a real cost on constrained hosting).
|
|
VERIFICATION_TTL = timedelta(days=7)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BotSignature:
|
|
name: str
|
|
ua_substrings: tuple[str, ...]
|
|
verify_suffixes: tuple[str, ...] # PTR hostname must end in one of these
|
|
|
|
|
|
def _load_signatures() -> list[BotSignature]:
|
|
raw = json.loads(SIGNATURES_PATH.read_text())
|
|
return [
|
|
BotSignature(name=e["name"], ua_substrings=tuple(e["ua_substrings"]), verify_suffixes=tuple(e["verify_suffixes"]))
|
|
for e in raw
|
|
]
|
|
|
|
|
|
_SIGNATURES = _load_signatures()
|
|
|
|
|
|
def classify_bot(user_agent: str | None) -> str | None:
|
|
"""Return the claimed bot name via UA substring match, or None."""
|
|
if not user_agent:
|
|
return None
|
|
ua_lower = user_agent.lower()
|
|
for sig in _SIGNATURES:
|
|
if any(sub.lower() in ua_lower for sub in sig.ua_substrings):
|
|
return sig.name
|
|
return None
|
|
|
|
|
|
def _signature_for(bot_name: str) -> BotSignature | None:
|
|
return next((s for s in _SIGNATURES if s.name == bot_name), None)
|
|
|
|
|
|
def verify_bot_ip(ip: str, bot_name: str) -> bool:
|
|
"""Reverse-DNS + forward-confirm that `ip` really belongs to `bot_name`."""
|
|
sig = _signature_for(bot_name)
|
|
if sig is None:
|
|
return False
|
|
try:
|
|
hostname, _, _ = socket.gethostbyaddr(ip)
|
|
except (socket.herror, socket.gaierror, OSError):
|
|
return False
|
|
if not any(hostname.lower().endswith(suffix) for suffix in sig.verify_suffixes):
|
|
return False
|
|
try:
|
|
forward_ips = socket.gethostbyname_ex(hostname)[2]
|
|
except (socket.herror, socket.gaierror, OSError):
|
|
return False
|
|
return ip in forward_ips
|
|
|
|
|
|
def is_verification_stale(last_verified_at: datetime | None) -> bool:
|
|
"""True if this IP needs a fresh DNS check (Ch07 caching rule)."""
|
|
if last_verified_at is None:
|
|
return True
|
|
return datetime.utcnow() - last_verified_at > VERIFICATION_TTL
|