Volver a la wiki

Servicio net_guard

Módulo: monitoring/services/net_guard.py | Versión: Etapa 3 (commit cb4fe5e+) | Status: Centralizado, en uso por http_service + notification_service

Propósito

Módulo centralizado de validación y pinning de destinos de red para eliminar SSRF (Server-Side Request Forgery) e inyecciones de IP internas, y DNS-rebinding / TOCTOU (Time-of-Check-Time-of-Use), garantizando que los clientes HTTP nunca re-resuelven a una dirección diferente después de la validación inicial.

API Pública

Capa 1: Validación temprana (cheap boolean checks)

def validate_destination_host(hostname: str, port: int = None) -> tuple[bool, str]:
    """Resuelve hostname, rechaza si alguna IP es interna. Cheap."""

def validate_destination_url(url: str) -> tuple[bool, str]:
    """Parsea URL, llama a validate_destination_host. Cheap."""

Casos:

Uso: Early gate (p.ej. test antes de queue un webhook send, sin gastar connection).

Capa 2: Resolución + Pinning (DNS-rebinding defense)

def resolve_and_validate(url: str) -> tuple[ResolvedDestination | None, str]:
    """Resuelve hostname UNA VEZ. Devuelve todas las IPs validadas (ninguna interna).
    
    Returns:
        (dest, "") si todo OK, donde dest.ips es la lista de IPs pineadas.
        (None, reason_string) si el destino es bloqueado.
    """

ResolvedDestination = namedtuple("ResolvedDestination", "scheme hostname port ips path_qs")

Datos devueltos:

Invariante: Si resolve_and_validate devuelve éxito, TODAS las IPs en dest.ips son públicas válidas. El cliente NUNCA re-resuelve; solo intenta conectar a esas IPs.

Capa 2b: Reintentos entre IPs (multi-IP resiliencia)

def try_each_ip(dest: ResolvedDestination, attempt: callable, deadline: float | None = None) -> Any:
    """Ejecuta attempt(ip) sobre cada IP pinneada; devuelve el primer éxito.

    attempt(ip) debe ser un callable que devuelva algo o lance excepción.
    Solo reintenta ante fallo de CONEXIÓN (OSError, ssl.SSLError, httpx.ConnectError,
    httpx.ConnectTimeout) — un ReadTimeout a mitad de respuesta o un error de
    programación NO reintenta, se propaga (cola auditoria A, 04-09-2026: antes
    cualquier Exception disparaba retry y solo el error de la ÚLTIMA IP era visible).
    `deadline` (corte de time.monotonic()) para el presupuesto compartido de la sonda;
    si se agota sin ningún intento exitoso, lanza TimeoutError explícito en vez del
    RuntimeError genérico de "no validated IPs".
    """

async def try_each_ip_async(dest: ResolvedDestination, attempt: async_callable, deadline: float | None = None) -> Any:
    """Variante async."""

Patrón típico:

def _attempt(ip):
    # Usa ip, presenta hostname para TLS/Host header.
    resp = client.post(pinned_url(dest, ip), ..., headers=pin["headers"], extensions=pin["extensions"])
    return resp.status_code

status_code = try_each_ip(dest, _attempt)  # Retry automático entre IPs.

Capa 2c: Presentación de hostname original (TLS SNI, cert verification, Host header)

def pinned_url(dest: ResolvedDestination, ip: str) -> str:
    """Rebuild URL pointing at pinned IP, e.g. \"https://[2606:4700:...]:443/path?q=1\"."""

def httpx_pin_kwargs(dest: ResolvedDestination) -> dict:
    """Per-request kwargs para httpx: Host header + SNI hostname."""
    # {"headers": {"Host": "example.com:8080"}, "extensions": {"sni_hostname": "example.com"}}

def pinned_requests_session(dest: ResolvedDestination) -> requests.Session:
    """requests.Session con HTTPAdapter pineado (server_hostname/assert_hostname para TLS)."""

def host_header(dest: ResolvedDestination) -> str:
    """Host header value: hostname[:nondefault_port]."""

Flujo anti-rebinding

  1. Resolve + Validate once: resolve_and_validate(url) → lista de IPs pineadas.
  2. Connect a IP pinneada: try_each_ip(dest, _attempt_fn), donde cada _attempt_fn usa pinned_url(dest, ip) + headers/extensions de httpx_pin_kwargs o pinned_requests_session.
  3. Present original hostname: Host header, TLS SNI, cert verification contra el hostname original (no la IP).
  4. Retry logic: Si IP#1 falla a nivel conexión, intenta IP#2, etc. Sin fallback a re-resolving.
  5. Result: Una conexión segura, sin ventana TOCTOU, multi-IP resiliente.

Limitaciones conocidas (documentadas)

Implementación interna

Ranges bloqueados

_INTERNAL_IPV4_RANGES = [
    ipaddress.ip_network("0.0.0.0/8"),      # This network
    ipaddress.ip_network("10.0.0.0/8"),     # Private
    ipaddress.ip_network("127.0.0.0/8"),    # Loopback
    ipaddress.ip_network("169.254.0.0/16"), # Link-local (metadata)
    ipaddress.ip_network("198.18.0.0/15"),  # RFC 2544 benchmarking (cola auditoria A)
    ipaddress.ip_network("224.0.0.0/4"),    # Multicast (cola auditoria A)
    # ...
]

_INTERNAL_IPV6_RANGES = [
    ipaddress.ip_network("::1/128"),  # Loopback
    # ...
]

_BLOCKED_HOSTNAMES = {
    "localhost", "localhost.localdomain", 
    "example.com", "example.org", "example.net",
    # ...
}

Función interna: _ip_blocked(ip: str) -> CIDR | None

Retorna el rango CIDR si la IP está bloqueada, else None. Compara también la forma ipv4_mapped de IPs IPv6 (Tanda 9, 01-09-2026): ::ffff:127.0.0.1 bloquea igual que 127.0.0.1.

Función interna: _is_ip_literal(hostname: str) -> bool

Detecta si hostname es una dirección IPv4 o IPv6 literal.

Función interna: _strip_trailing_dot(hostname: str) -> str

Normaliza localhost. (forma FQDN con punto final, resuelve igual que localhost pero no casaba con _BLOCKED_HOSTNAMES por comparación exacta) antes de validar — añadida en la cola auditoria A (04-09-2026).

Clientes principales

Ambos usan resolve_and_validate() + try_each_ip[_async]() + httpx_pin_kwargs() / pinned_requests_session().

Véase también

Subir