Servicio: consulta de avisos de seguridad en OSV.dev

Ficha técnica

Firma de la función

def resolve_vulns(deps: list[Dep], workers: int = 8) -> None:
    """Cruza cada dependencia con OSV.dev (Regla 15: fuente real).

    La versión consultada es la instalada (PROD > runner > lockfile) o, si no
    hay, la **resuelta** del spec. Falla suave: si OSV no responde, 0 avisos.
    """

Flujo de ejecución

Fase 1: Filtrado de dependencias consultables

queryable: list[tuple[Dep, str]] = []
for d in deps:
    ver = d.installed or d.resolved
    if ver and d.ecosystem in OSV_ECOSYSTEM:   # {"pypi": "PyPI", "npm": "npm"}
        queryable.append((d, ver))

Entrada: list[Dep] — las dependencias ya deduplicadas, con resolved relleno por resolve_specs() y installed por collect().
Filtro: Solo deps con (installed o resolved) Y ecosystem ∈ {PyPI, npm}.
Salida: list[(Dep, str)] — tuplas (dep, versión en uso).

Por qué este filtro: OSV.dev devuelve avisos de CUALQUIER versión del paquete si la versión no se especifica; es ruido.

Ojo con lo que se manda (cambio del 28-08-2026, task #275): hasta esa fecha se enviaba base_from_spec(d.pinned), es decir el mínimo de la horquilla (requests>=2.31.0 → «2.31.0»). Ninguna instalación real corre esa versión: pip instala la mayor que el spec admite. El resultado eran 30 avisos inventados (22 HIGH de aiohttp, 6 de requests, 2 de pytest). Ahora se manda d.resolved, que calcula deps_specs.resolve() contra la lista completa de versiones publicadas.

Fase 2: Batch POST a /v1/querybatch

for i in range(0, len(queryable), OSV_BATCH_SIZE):  # OSV_BATCH_SIZE = 100
    chunk = queryable[i : i + OSV_BATCH_SIZE]
    body = {
        "queries": [
            {
                "package": {"name": d.package, "ecosystem": osv_eco[d.ecosystem]},
                "version": ver
            } for d, ver in chunk
        ]
    }
    resp = _post_json(OSV_BATCH_URL, body)
    batch = (resp or {}).get("results") or []
    batch += [{}] * (len(chunk) - len(batch))  # Rellenar gaps si OSV corto
    results += batch

Endpoint: POST https://api.osv.dev/v1/querybatch
Payload: hasta 100 queries con nombre, ecosistema y versión de cada dep.
Respuesta: lista de results con avisos por query (típico 0-5 por dep).

Falla suave: Si OSV no responde (timeout, 500), results es vacío o None; se rellena con dicts vacíos {} para que zip no falle.

Fase 3: Colecta de IDs de avisos

all_ids: set[str] = set()
for (d, _ver), r in zip(queryable, results, strict=False):
    ids = sorted({v["id"] for v in (r.get("vulns") or []) if v.get("id")})
    if ids:
        d.vuln_count = len(ids)
        d.vuln_ids = ",".join(ids)
        all_ids.update(ids)

Extrae: IDs únicos de cada aviso (ej. GHSA-abc-def-ghi, PYSEC-2026-123).
Actualiza Dep: vuln_count, vuln_ids (string separado por comas).
Acumula: all_ids para la siguiente fase (enriquecimiento paralelo de severidad).

Fase 4: Fetch paralelo del detalle de cada aviso

def fetch_detail(vuln_id: str) -> tuple[str, dict]:
    return vuln_id, _fetch_json(OSV_VULN_URL.format(vuln_id=urllib.parse.quote(vuln_id))) or {}

detail: dict[str, dict] = {}
with ThreadPoolExecutor(max_workers=workers) as pool:
    for vuln_id, data in pool.map(fetch_detail, sorted(all_ids)):
        detail[vuln_id] = data

Endpoint: GET https://api.osv.dev/v1/vulns/{vuln_id}
Se guarda el aviso entero (desde 28-08-2026), no solo la severidad, porque de él salen dos cosas:

• database_specific.severity → la gravedad (de GHSA, Cloudflare Radar, etc.). Valores típicos: critical, high, moderate (alias medium), low, unknown.
• affected[].ranges[].events[].fixed → las versiones con arreglo, que alimentan fix_within_spec.

Paralelismo: 8 workers (HTTPs no está limitado por GIL en urllib).

Fase 5: Peor severidad y fix_within_spec por dep

for d in deps:
    if not d.vuln_ids:
        continue
    ids = d.vuln_ids.split(",")
    sevs = [((detail.get(v) or {}).get("database_specific") or {}).get("severity") or "unknown" for v in ids]
    d.vuln_severity = min((s.lower() for s in sevs), key=lambda s: SEV_ORDER.get(s, 4))
    # ¿Cada aviso tiene una versión con fix que la horquilla ACTUAL admite?
    known: list[bool] = []
    for vuln_id in ids:
        fixed = _fixed_versions(detail.get(vuln_id) or {}, d.package, d.ecosystem)
        if fixed:
            known.append(any(specs.satisfies(d.pinned, f, d.ecosystem) for f in fixed))
    d.fix_within_spec = all(known) if known else None

Severidad: para cada dep con avisos, la peor de la lista (crítica > alta > moderada > baja > desconocida).
SEV_ORDER (definido en el módulo):

SEV_ORDER = {"critical": 0, "high": 1, "moderate": 2, "medium": 2, "low": 3, "unknown": 4}

fix_within_spec es tri-estado y responde a «¿hace falta tocar el manifiesto?»:

El helper _fixed_versions(vuln, package, ecosystem) filtra los bloques affected por nombre de paquete (normalizado PEP 503 en PyPI) y ecosistema antes de recoger los eventos fixed — un aviso puede cubrir varios paquetes a la vez.

Sub-componentes (helpers)

_post_json(url: str, payload: dict) -> dict | None

Propósito: POST genérico con headers correctos.

Entrada: URL, payload Python (convertido a JSON).
Headers:

• User-Agent: crearack-deps-freshness/2.0 (+https://workspace.crearack.com)
• Content-Type: application/json
• Accept: application/json

Timeout: 20 segundos.
Falla suave: Excepciones (urllib.error.URLError, json.JSONDecodeError, TimeoutError) → retorna None.

Dataclass auxiliar: Dep

Desde el 28-08-2026 vive en scripts/deps_sources.py (el troceo por la Regla 5: deps_freshness.py rondaba las 500 LOC).

@dataclass
class Dep:
    repo: str  # 'crearack-pro' | 'workspace'
    ecosystem: str  # 'pypi' | 'npm'
    manifest: str  # ruta relativa (el más restrictivo tras dedup)
    package: str  # nombre del paquete
    pinned: str  # spec crudo del manifiesto (==6.0.4, ^9.3.0, >=2.10.0...)
    installed: str | None = None
    latest: str | None = None
    lag_kind: str = "unknown"  # "major", "minor", "patch", etc.
    is_dev: bool = False  # True si es test/dev dependency
    vuln_count: int = 0  # número de avisos
    vuln_ids: str | None = None  # ej "GHSA-abc,PYSEC-123"
    vuln_severity: str | None = None  # peor severidad
    # --- campos nuevos (v2.0, task #275) ---
    manifests: str = ""  # todos los manifiestos donde aparece, separados por coma
    resolved: str | None = None  # mayor versión publicada que satisface `pinned`
    installed_source: str | None = None  # prod | runner | lockfile
    unbounded: bool = False  # el spec no pone techo superior
    unbounded_major: bool = False  # ...y ya corre en un major distinto al declarado
    fix_within_spec: bool | None = None  # True: redeploy basta · False: cambiar pin
    _versions: list[str] = field(default_factory=list, repr=False)  # no se publica

_versions (la lista completa de versiones publicadas del registro) es de uso interno: _payload_items() descarta todo campo que empiece por _ antes de mandar el snapshot al Workspace.

Casos de uso

Caso 1: Dep con versión instalada resoluble

Input:  Dep(package="django", pinned="==6.0.6", installed="6.0.6", installed_source="prod")
Phases:
  1. Queryable: ("django", "6.0.6")
  2. Batch: OSV responde 3 avisos (GHSA-...)
  3. Colecta: vuln_count=3, vuln_ids="GHSA-...,GHSA-...,GHSA-..."
  4. Fetch: detalle de cada uno → severidades ["high", "high", "moderate"]
  5. Asigna: vuln_severity="high" (peor); ningún fix cabe en "==6.0.6"
Output: Dep(..., vuln_count=3, vuln_severity="high", fix_within_spec=False)
        → la UI dice "cambiar pin"

Caso 2: Dep sin versión instalada — se usa la resuelta

Input:  Dep(package="pytest", pinned=">=7.0", installed=None)
        (es un manifiesto de tests: is_dev=True, no se instala en el job)
Phases:
  1. resolve_specs(): resolved = mayor publicada que cumple ">=7.0" (ej. "9.2.1")
  2. Queryable: ("pytest", "9.2.1")   # NO "7.0"
  3. OSV: sin avisos
Output: Dep(..., vuln_count=0, vuln_ids=None, vuln_severity=None)

Este es exactamente el caso que antes producía falsos positivos: con «7.0» OSV devolvía 2 avisos que no afectan a ninguna instalación real.

Caso 2b: aviso que se tapa con un redeploy

Input:  Dep(package="aiohttp", pinned=">=3.14.0,<3.15", installed="3.14.0")
Phases:
  1-3. OSV devuelve 1 aviso HIGH sobre 3.14.0
  4.   El aviso publica fixed = "3.14.3"
  5.   specs.satisfies(">=3.14.0,<3.15", "3.14.3") → True
Output: Dep(..., fix_within_spec=True)  → la UI dice "basta redeploy"

Caso 3: Dep sin versión resoluble (spec opaco)

Input:  Dep(package="algun-paquete-interno", pinned="workspace:*", installed=None)
Phases:
  1. specs.resolve("workspace:*", ...) → None   # no se resuelve contra el registro
  2. No queryable: se omite
Output: Dep(..., vuln_count=0)  # Sin consultar OSV

Caso 4: OSV no responde

Input:  Dep(package="rarelib", pinned="==1.0.0", installed="1.0.0")
Phases:
  1-2. Batch POST falla (timeout)
  3-5. results[i] es {} vacío
Output: Dep(..., vuln_count=0)  # Falla suave

Integración en collect()

El orden importa: resolve_vulns() corre después de resolver las horquillas y de deduplicar, porque necesita resolved (y la horquilla ganadora, la más restrictiva) para no inventar avisos.

def collect(...) -> list[Dep]:
    # ... parseo de manifiestos (parse_requirements, parse_package_json) ...
    # ... "instalada": PROD -> runner -> lockfile ...

    resolve_registry(deps)     # .latest + lista completa de versiones publicadas
    resolve_specs(deps)        # .resolved, .unbounded, .unbounded_major
    deps = dedupe(deps)        # una fila por (repo, ecosistema, paquete)
    resolve_vulns(deps)        # vuln_count, vuln_ids, vuln_severity, fix_within_spec

    for d in deps:
        d.lag_kind = lag_kind(d.installed or d.resolved, d.latest)
    
    # Orden: con aviso primero (peor severidad arriba), luego desfase
    deps.sort(
        key=lambda d: (
            0 if d.vuln_count else 1,
            SEV_ORDER.get(d.vuln_severity or "unknown", 4),
            LAG_ORDER.get(d.lag_kind, 9),
            d.repo,
            d.package.lower(),
        )
    )
    return deps

Observabilidad

Línea de salida

N dependencias (deduplicadas) · major=0 · minor=4 · patch=8 · unknown=2 · uptodate=105 ·
con avisos=12 · sin techo=21 (de ellas con major nuevo=3) · 'instalada' desde PROD=94

Indica: total tras deduplicar, conteos por tipo de desfase, deps con avisos, horquillas sin techo (y cuántas ya saltaron de major) y cuántas filas traen la instalada de producción — este último número es el que delata que falta el secret INTERNAL_TOOLS_TOKEN (saldría 0).

Tabla

Formato de las columnas (filas de ejemplo, no de una pasada concreta):

PKG          REPO          PINEADA           INSTALADA          RESUELTA  ÚLTIMA   DESFASE  SEGURIDAD              NOTAS
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
vitest       workspace     2.1.9             2.1.9 (runner)     2.1.9     2.1.10   minor    1 (critical) cambiar pin
django       crearack-pro  ==6.0.8           6.0.8 (prod)       6.0.8     6.0.8    uptodate —
anthropic    crearack-pro  >=0.52            0.122.0 (prod)     0.122.0   0.122.0  uptodate —                      SIN TECHO · major nuevo
pytest       crearack-pro  >=8.0             —                  9.2.1     9.2.1    uptodate —                      en 2 manifiestos

Columnas nuevas desde el 28-08-2026: RESUELTA, la fuente entre paréntesis en INSTALADA (prod/runner/lockfile), la coletilla del arreglo en SEGURIDAD (redeploy basta / cambiar pin / sin fix) y la columna NOTAS (sin techo · en N manifiestos).

Reglas y restricciones

1. Regla 15: OSV.dev es fuente real — se consulta en cada run contra la base de GHSA/PYSEC oficial.
2. Sin API key: El servicio es público; no hay credenciales que gestionar.
3. Falla suave: Si OSV cae, el snapshot se publica sin severidad (vuln_count=0), no bloquea el robot.
4. Versión real, nunca el mínimo del spec: a OSV se manda installed or resolved. Mandar la base de la horquilla (>=2.31.0 → «2.31.0») producía avisos que no existen en ninguna instalación — 30 falsos positivos el 28-08-2026.
5. Lotes de 100: Optimización estándar de OSV.dev para latencia y throughput.
6. Después del dedup: se consulta una sola vez por (repo, ecosistema, paquete), con la horquilla más restrictiva de las que declaran los manifiestos.

Véase también

• [[feature—harness—osv-security-deps-freshness]]
• [[entity—ci—service—deps-freshness-collector]]
• [[entity—ci—workflow—deps-freshness]]
• [[entity—api—endpoint—installed-packages]]
• [[entity—migrations—table—deps-freshness]]
• [[runbook—workspace-tech—tool-dependencias]]
• [[concept—infra—supply-chain-security]]