Volver a la wiki

Helper de clave JWT dedicada (agent_jwt_secret)

Descripción

Módulo: terminal/api/auth.py
Función: _agent_jwt_secret() -> str
Propósito: Devolver la clave de firma para JWT del Local Agent, con fallback inerte a SECRET_KEY.

Qué hace

def _agent_jwt_secret() -> str:
    """Signing key for the Agent JWT: dedicated AGENT_JWT_SECRET if set, else the
    Django SECRET_KEY (inert fallback — identical to the previous behaviour)."""
    return getattr(settings, "AGENT_JWT_SECRET", "") or settings.SECRET_KEY

Lógica:

  1. Si settings.AGENT_JWT_SECRET está definida y no vacía → devuelve su valor
  2. Si no → devuelve settings.SECRET_KEY (comportamiento anterior, sin breaking change)

Fuentes de datos

Lectura

Invocaciones

REST (emisión de tokens)

terminal/api/auth.py:

  1. generate_agent_token(agent_id, tenant_id, user_email, expiry_hours=24)

    payload = {
    "agent_id": agent_id,
    "tenant_id": tenant_id,
    "email": user_email,
    "iat": datetime.now(UTC),
    "exp": datetime.now(UTC) + timedelta(hours=expiry_hours),
    "type": "access",
}
return jwt.encode(payload, _agent_jwt_secret(), algorithm="HS256")
    • Endpoint (probablemente): POST /api/agents/token/ o similar
    • Flujo: User inicia sesión, Agent genera token con esta clave

  2. generate_refresh_token(agent_id, tenant_id)

    payload = {
    "agent_id": agent_id,
    "tenant_id": tenant_id,
    "iat": datetime.now(UTC),
    "exp": datetime.now(UTC) + timedelta(days=AGENT_REFRESH_TOKEN_EXPIRY_DAYS),
    "type": "refresh",
}
return jwt.encode(payload, _agent_jwt_secret(), algorithm="HS256")
    • Duración: 180 días (configurable AGENT_REFRESH_TOKEN_EXPIRY_DAYS)

REST (verificación de tokens)

terminal/api/auth.py:

verify_agent_token(token: str) -> dict | None

try:
    payload = jwt.decode(token, _agent_jwt_secret(), algorithms=["HS256"])
    return payload
except jwt.ExpiredSignatureError:
    logger.warning("Agent token expired")
    return None
except jwt.InvalidSignatureError:
    logger.warning("Agent token invalid")
    return None
except Exception as e:
    logger.error("Agent token verification error: %s", e)
    return None

WebSocket (conexión agente)

terminal/consumers.py (async WebSocket consumer):

async def connect(self):
    """WebSocket connect: validate Agent JWT token."""
    ...
    try:
        from terminal.api.auth import _agent_jwt_secret
        payload = jwt.decode(token, _agent_jwt_secret(), algorithms=["HS256"])
        
        # Verify agent_id matches token
        if payload.get("agent_id") != self.agent_id:
            await self.close(code=4003)
            return
        
        self.agent_id = self.agent_id
        self.tenant_id = payload.get("tenant_id")
        await self.accept()
    except jwt.InvalidSignatureError:
        await self.close(code=4001)
        return

Configuración

Setup inicial (config/settings/base.py)

# Dedicated signing key for the Local Agent JWT, decoupled from SECRET_KEY
# (same rationale as CREDENTIAL_ENCRYPTION_KEY): rotating the Django SECRET_KEY
# should not invalidate every Agent token, and a leaked SECRET_KEY should not let
# an attacker forge Agent tokens. Empty → terminal/api/auth.py falls back to
# SECRET_KEY (inert: identical behaviour to today). Set this env var in Dokploy
# to activate the split (forces a one-time re-auth of the fleet, as expected
# when rotating a signing key).
AGENT_JWT_SECRET = os.getenv("AGENT_JWT_SECRET", "")

Activación en Dokploy

Antes de activar:

Al activar:

  1. Declarar en Dokploy: AGENT_JWT_SECRET=<new-random-key> (RFC 5234, HS256 needs >256 bits)
  2. Deploy → nueva clave en uso
  3. Tokens antiguos (firmados con SECRET_KEY) ya no verifican
  4. Flota se re-autentica automáticamente (esperado en cualquier rotación de clave)
  5. A partir de ahí: tokens nuevos firmados con AGENT_JWT_SECRET

Seguridad

Rationale: Por qué separar del SECRET_KEY

  1. Rotación sin cascada: Rotar SECRET_KEY (p.ej. por cambio de hosting) invalidaba todo token del Agent. Con clave dedicada, rotan independientemente.

  2. SPOF reducido: Si SECRET_KEY se filtra:

    • Antes: Atacante puede forjar tokens del Agent, cifrar datos de credenciales, firmar cookies de sesión
    • Después: Atacante puede cifrar/sesiones, pero no forjar tokens del Agent (usa clave distinta)

  3. Compliance: Muchos estándares requieren que claves de firma de autenticación estén separadas de claves generales.

Fallback inerte

Flujo de vida de un token

1. User inicia sesión en dashboard
2. generate_agent_token(...) → jwt.encode(..., _agent_jwt_secret())
3. Token se devuelve al cliente (Agent)
4. Agent: POST /api/... + Authorization: Bearer <token>
5. verify_agent_token(token) → jwt.decode(..., _agent_jwt_secret())
   → Si clave = SECRET_KEY (env var no declarada): verifica OK
   → Si clave = AGENT_JWT_SECRET (env var declarada): verifica OK
6. Agent: WebSocket ws://...
7. consumers.py connect() → jwt.decode(..., _agent_jwt_secret())
   → Mismo helper → misma clave → sincronía REST+WebSocket

Datetime aware

El helper se usa con timestamps “aware” (UTC):

payload = {
    "iat": datetime.now(UTC),  # Aware (deprecado: datetime.utcnow())
    "exp": datetime.now(UTC) + timedelta(hours=24),
    ...
}

Cambio complementario: datetime.utcnow() → datetime.now(UTC) en los payloads JWT (fix de deprecación).

Tests

tests/api/test_config_ia.py:

  1. test_agent_jwt_secret_fallback(settings)

    settings.AGENT_JWT_SECRET = ""
assert auth._agent_jwt_secret() == settings.SECRET_KEY

settings.AGENT_JWT_SECRET = "a-dedicated-agent-key"
assert auth._agent_jwt_secret() == "a-dedicated-agent-key"
    • Verifica: Fallback inerte
    • Verifica: Usa clave dedicada si está declarada

  2. test_agent_jwt_roundtrip_with_dedicated_key(settings)

    settings.AGENT_JWT_SECRET = "a-dedicated-agent-key"
token = auth.generate_agent_token("agent-1", 1, "edu@example.com")
payload = auth.verify_agent_token(token)
assert payload["agent_id"] == "agent-1"

# Rotación invalida token anterior
settings.AGENT_JWT_SECRET = "a-rotated-key"
assert auth.verify_agent_token(token) is None
    • Verifica: Round-trip (genera + verifica)
    • Verifica: Rotación invalida token anterior (comportamiento esperado)

Véase también

Subir