# JTI Denylist / Token Revocation Pattern

Instant access-token revocation behind a Clean Architecture port, with in-memory and Redis backends.

## Overview

Access tokens are stateless JWTs with a short TTL (15 min by default). When a user logs out, the refresh token is revoked in the database, but the access token remains valid until it expires. The JTI denylist closes this gap by checking every access token's unique `jti` (JWT ID) claim against a shared denylist.

## Port Design

```python
from typing import Protocol

class JtiDenylist(Protocol):
    async def add(self, jti: str, ttl_seconds: int) -> None: ...
    async def is_denylisted(self, jti: str) -> bool: ...
    async def clear(self) -> None: ...
```

## JTI in JWT Tokens

Access tokens already carry a `jti` claim — generated as `str(uuid4())` in `create_access_token()`:

```python
def create_access_token(subject, expires_delta=None):
    to_encode = {
        "sub": str(subject),
        "jti": str(uuid4()),  # Unique per token
        "exp": ...,
        "iat": ...,
        "typ": "access",
        # ...
    }
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")
```

The JTI is a UUID string — it's unique, non-enumerable, and carries no user information.

## In-Memory Implementation

```python
class InMemoryJtiDenylist:
    def __init__(self):
        self._store: OrderedDict[str, float] = OrderedDict()

    async def add(self, jti: str, ttl_seconds: int) -> None:
        if ttl_seconds < 1:
            return
        self._store[jti] = time.monotonic() + ttl_seconds

    async def is_denylisted(self, jti: str) -> bool:
        expires_at = self._store.get(jti)
        if expires_at is None:
            return False
        if time.monotonic() >= expires_at:
            del self._store[jti]  # Lazy expiry
            return False
        return True

    async def clear(self) -> None:
        self._store.clear()
```

**Design decisions:**
- Lazy expiry: expired entries are cleaned on access, not by a background timer
- TTL matches access token lifetime (default 900 seconds / 15 min)
- `time.monotonic()` — immune to system clock changes

## Redis Implementation

```python
class RedisJtiDenylist:
    def __init__(self, redis, key_prefix="denylist:jti:"):
        self._redis = redis
        self._key_prefix = key_prefix

    async def add(self, jti, ttl_seconds):
        if ttl_seconds < 1:
            return
        with suppress(Exception):
            await self._redis.setex(f"{self._key_prefix}{jti}", ttl_seconds, "1")

    async def is_denylisted(self, jti):
        with suppress(Exception):
            value = await self._redis.get(f"{self._key_prefix}{jti}")
            return value is not None
        return False
```

**Fail-open:** If Redis is down, `is_denylisted` returns False (token is allowed through). This prevents a Redis outage from blocking all authenticated requests.

## Singleton Factory

```python
_JTI_DENYLIST: JtiDenylist | None = None

def get_jti_denylist() -> JtiDenylist:
    global _JTI_DENYLIST
    if _JTI_DENYLIST is not None:
        return _JTI_DENYLIST

    settings = get_settings()
    if settings.REDIS_HOST:
        from redis.asyncio import Redis as AsyncRedis
        redis_client = AsyncRedis(
            host=settings.REDIS_HOST, port=settings.REDIS_PORT,
            db=settings.REDIS_DB, password=settings.REDIS_PASSWORD or None,
            decode_responses=True,
        )
        _JTI_DENYLIST = RedisJtiDenylist(redis_client)
    else:
        _JTI_DENYLIST = InMemoryJtiDenylist()
    return _JTI_DENYLIST
```

The factory lives in `core/jti_denylist.py` alongside the port and implementations — NOT in a module's presentation/dependencies.py. This avoids circular imports when multiple modules (user, streaming) need to access the denylist.

## Auth Dependency Integration

The JTI check is added to `require_user_id` dependency in each module's presentation layer:

```python
async def require_user_id(token: str) -> int:
    payload = decode_access_token(token)
    user_id = int(payload["sub"])
    jti: str = payload["jti"]

    denylist = get_jti_denylist()
    if await denylist.is_denylisted(jti):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token has been revoked",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user_id
```

**Important:** The JTI check runs AFTER JWT decode/validation. A token with an invalid signature or expired timestamp is rejected before the denylist check even runs.

## Logout Integration

On logout, the current access token's JTI is added to the denylist:

```python
@router.post("/logout", status_code=204)
async def logout(token: str = Depends(oauth2_scheme), ...):
    payload = decode_access_token(token)
    user_id = int(payload["sub"])
    jti: str = payload["jti"]

    # Denylist this access token for its remaining lifetime
    denylist = get_jti_denylist()
    await denylist.add(jti, ttl_seconds=900)  # 15 min default access token TTL

    # Also revoke all refresh sessions
    await use_cases.logout(user_id)
```

## Refresh Integration

On token refresh, the old access token's JTI (from the Bearer header) should also be denylisted:

```python
@router.post("/refresh")
async def refresh(
    data: RefreshRequest,
    token: str = Depends(oauth2_scheme),  # Old access token from Bearer header
    use_cases: ...,
    ...
):
    # Denylist the old access token's JTI
    payload = decode_access_token(token)
    denylist = get_jti_denylist()
    await denylist.add(payload["jti"], ttl_seconds=900)

    # Proceed with refresh rotation
    return await use_cases.refresh_token(data.refresh_token)
```

## Testing

```python
class _FakeRedis:
    def __init__(self):
        self._store: dict[str, str] = {}

    async def get(self, key: str) -> str | None:
        return self._store.get(key)

    async def setex(self, key: str, seconds: int, value: str) -> None:
        self._store[key] = value

    async def delete(self, *keys: str) -> int:
        removed = 0
        for k in keys:
            if k in self._store:
                del self._store[k]
                removed += 1
        return removed

class TestJtiDenylist:
    @pytest.fixture
    def denylist(self):
        return InMemoryJtiDenylist()

    async def test_not_denylisted_by_default(self, denylist):
        assert await denylist.is_denylisted("some-jti") is False

    async def test_add_and_check(self, denylist):
        await denylist.add("test-jti-123", ttl_seconds=60)
        assert await denylist.is_denylisted("test-jti-123") is True

    async def test_zero_ttl_does_not_add(self, denylist):
        await denylist.add("jti", ttl_seconds=0)
        assert await denylist.is_denylisted("jti") is False

    async def test_clear(self, denylist):
        await denylist.add("jti", ttl_seconds=60)
        await denylist.clear()
        assert await denylist.is_denylisted("jti") is False

    async def test_fail_open_on_error(self):
        class _BrokenRedis:
            async def get(self, key): raise ConnectionError("Down")
            async def setex(self, key, sec, val): raise ConnectionError("Down")
            async def delete(self, *keys): raise ConnectionError("Down")

        denylist = RedisJtiDenylist(_BrokenRedis())
        assert await denylist.is_denylisted("any") is False
        await denylist.add("any", ttl_seconds=60)  # Should not raise
```

## Integration Test: Full Logout Flow

```python
async def test_logout_revokes_token(test_client, settings):
    # Register and login
    await test_client.post("/api/v1/auth/register", json={...})
    login_resp = await test_client.post("/api/v1/auth/login", json={...})
    token = login_resp.json()["access_token"]

    # /me works before logout
    me_resp = await test_client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"})
    assert me_resp.status_code == 200

    # Logout
    logout_resp = await test_client.post("/api/v1/auth/logout", headers={"Authorization": f"Bearer {token}"})
    assert logout_resp.status_code == 204

    # /me fails after logout (token was denylisted)
    me_resp = await test_client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"})
    assert me_resp.status_code == 401
    assert "Token has been revoked" in me_resp.text
```

## Key Invariants

1. **JTI is always present** — access tokens without a `jti` claim are rejected by `decode_access_token()`'s `require` options
2. **Fail-open: Redis down → tokens allowed** — prevents Redis outage from becoming total auth outage
3. **TTL matches access token lifetime** — no need to clean up entries after the token would have expired naturally
4. **Logged-out token cannot be reused** — logout action proves possession of the token; you can't denylist a token you never saw
5. **Refresh rotation also denylists the old access token** — after refresh, the old access token is dead even if someone intercepted it

## E2E HTTP Route Tests for Token Revocation

Each authenticated route module should have a test proving that a denylisted JTI is rejected. This catches dependency wiring mistakes where a module's auth dependency updates but doesn't pass through the JTI check:

### Test Pattern

```python
from app.core.jti_denylist import get_jti_denylist, reset_jti_denylist
from app.core.security import create_access_token, decode_access_token


async def test_me_rejects_denylisted_token(client):
    """GET /auth/me rejects a denylisted access token."""
    reset_jti_denylist()                         # Isolate from other tests
    token = create_access_token(subject="1")
    payload = decode_access_token(token)
    denylist = get_jti_denylist()
    await denylist.add(payload["jti"], ttl_seconds=60)

    response = await client.get(
        "/api/v1/auth/me",
        headers={"Authorization": f"Bearer {token}"},
    )

    assert response.status_code == 401
    assert response.json()["detail"] == "Token has been revoked"
    reset_jti_denylist()                         # Clean up for next test
```

### Coverage Checklist

Every route module with authentication should have at least one test:

| Route Module | Endpoint to Test | Why |
|---|---|---|
| User (auth) | `GET /auth/me` | Core auth path — if this breaks, everything breaks |
| User (auth) | `GET /auth/sessions` | Session listing, exercises the same dep |
| Debrid accounts | `GET /debrid/accounts` | Proves debrid's `require_user_id` wires JTI correctly |
| Streaming resolve | `POST /streams/resolve` | Proves streaming's `require_user_id` wires JTI correctly |

### Key Assertions

- Status code: `401`
- Detail message: `"Token has been revoked"` (exact string from the auth helper)
- No fallback to `"Invalid or expired token"` — that would mean the denylist check was skipped
- Use `reset_jti_denylist()` before and after the test to avoid cross-test contamination

### Pitfall: Test Isolation

The `get_jti_denylist()` singleton persists across tests. Relying on `ttl_seconds` expiry between tests is unreliable (tests run too fast). **Always call `reset_jti_denylist()` at the start and end of denylist tests.**