Support API tokens #13

New issue

Closed

opened 2026-08-23 17:57:58 +02:00 by blacklight · 0 comments

blacklight commented 2026-08-23 17:57:58 +02:00

Owner

Copy link

Generic API Tokens for Songhive

Analysis of the Existing Auth System

Current mechanisms

Songhive currently ships three authentication paths, all gated through the same
Authorization: Bearer <token> header:

Path	Token type	Backing store	Expiry
Login / Refresh	HS256 JWT (access) + opaque refresh token	Refresh token in Redis (SHA-256 keyed)	Access: configurable minutes; refresh: configurable days
OAuth2 Authorization Code + PKCE	Opaque access + refresh tokens	Both in Redis	Governed by auth.access_token_expiry_minutes / refresh_token_expiry_days
OAuth2 Client Credentials	(not implemented, grant type rejected)	–	–

JWT infrastructure (songhive/api/middleware/auth.py)

create_access_token(user_id, secret_key, expires_minutes=15) builds a minimal
payload {sub, iat, [exp]} and signs it with HS256. When expires_minutes=None
the exp claim is omitted — the function already supports non-expiring tokens.

decode_access_token(token, secret_key) returns only the sub (user id) or
None; it discards all other claims.

Dependency resolution (songhive/api/deps.py)

_get_current_user calls decode_access_token → get_user_by_id. It knows
nothing beyond the sub claim: there is no notion of token type, origin, or
revocation status for access tokens. Revocation today is only possible for
opaque refresh tokens stored in Redis.

What is missing for API tokens

1. Persistent record — a DB row the user can list, name, and revoke.
2. Unique token identity (jti claim) so the DB record can be looked up
   during request validation without storing the entire JWT.
3. Absolute expiry — API tokens need a calendar date, not a relative
   duration in minutes.
4. Revocation — JWTs are stateless; revocation requires a DB check keyed
   on jti.
5. Type discrimination — token_type: "api_token" claim inside the JWT
   so the middleware can route to the DB check only for this token class,
   leaving ordinary short-lived access tokens unaffected.
6. CRUD API — endpoints for creating, listing, and revoking tokens.

---

Implementation Plan

Goal

Extend Songhive so authenticated users can create long-lived API tokens that
are issued as HS256 JWTs, carry an optional absolute expiry, are stored in the
database for audit and revocation, and are validated transparently by the
existing Bearer-token middleware.

Acceptance Criteria

• A user can POST /auth/api-tokens to create a named token; the raw JWT is
  returned once in the response.
• The token works immediately as a Bearer credential against any protected
  endpoint.
• A user can GET /auth/api-tokens to list their tokens (metadata only; raw
  JWT is never returned after creation).
• A user can DELETE /auth/api-tokens/{token_id} to revoke a specific token;
  revoked tokens are rejected within the same request cycle (no delay).
• Tokens with an expires_at in the past are rejected at auth time.
• Tokens with expires_at: null never expire unless explicitly revoked.
• Existing login / refresh / OAuth2 flows are unaffected.
• All new code is covered by unit and integration tests.

Assumptions and Constraints

• Tech stack: Python 3.14, FastAPI, SQLAlchemy 2 async, Alembic, Redis,
  PyJWT, bcrypt — all already present.
• The jti claim is stored in plain form in the DB (no additional hashing
  needed; the JWT itself is the secret, not the jti).
• Revocation is checked synchronously against the DB on every API-token
  request. A Redis cache layer for jti blocklisting is optional and
  explicitly out of scope for this plan.
• last_used_at is updated optimistically (fire-and-forget, acceptable to
  miss under concurrent load).
• Token names must be unique per user.
• Admin users may revoke any user's token via the existing admin route
  namespace (also out of scope; admin APIs can call the service layer directly).
• Migrations use Alembic autogenerate (alembic revision --autogenerate);
  a skeleton command is included in each migration step.

---

Section Map

#	Section	Depends on	Output
1	ApiToken DB model	—	songhive/models/api_token.py, migration
2	API token service	Section 1	songhive/users/api_tokens.py
3	Middleware & deps extension	Section 2	edits to middleware/auth.py, api/deps.py
4	HTTP routes	Section 3	songhive/api/routes/api_tokens.py, router registration
5	Tests	Sections 1–4	tests/test_api_tokens.py

---

Section 1: ApiToken SQLAlchemy Model + Migration

Goal

Define the persistent record for an API token and add it to the database.

Prerequisites

None.

Steps

1. Create songhive/models/api_token.py with a class ApiToken(Base).
   Required columns:

   Column	Type	Constraints
   id	str (UUID)	PK — inherited from Base
   user_id	str	FK → users.id ON DELETE CASCADE, indexed
   jti	str(64)	UNIQUE, indexed — the JWT ID claim value
   name	str(128)	NOT NULL — human label
   expires_at	datetime (tz-aware)	nullable — None = never expires
   revoked_at	datetime (tz-aware)	nullable — None = active
   last_used_at	datetime (tz-aware)	nullable
   created_at / updated_at	inherited from Base	—

   Add a UniqueConstraint("user_id", "name") so names are unique per user.

   Add a convenience property is_active → bool:
   revoked_at is None and (expires_at is None or expires_at > utcnow()).

2. Export ApiToken from songhive/models/__init__.py so Alembic can
   detect it during autogenerate.

3. Generate an Alembic migration:

   alembic revision --autogenerate -m "add api_tokens table"
   

   Review the generated file; confirm op.create_table("api_tokens", ...) with
   all columns and a FK to users.

4. Apply the migration in local dev:

   alembic upgrade head
   

Acceptance Criteria

• alembic upgrade head completes without errors.
• alembic downgrade -1 removes the table cleanly.
• ApiToken can be imported from songhive.models.

Tests / Verification

No unit tests for this section; table creation is exercised by Sections 2 & 5.

---

Section 2: API Token Service (songhive/users/api_tokens.py)

Goal

Encapsulate all API token business logic: JWT creation, DB persistence,
validation, and revocation.

Prerequisites

Section 1 (the ApiToken model).

Steps

1. Add create_api_token_jwt(user_id, secret_key, jti, expires_at) to
   songhive/api/middleware/auth.py (alongside create_access_token).

   # illustrative pseudo-signature
   def create_api_token_jwt(
       user_id: str,
       secret_key: str,
       jti: str,
       expires_at: Optional[datetime],   # absolute UTC datetime or None
   ) -> str:
       payload = {
           "sub": user_id,
           "jti": jti,
           "token_type": "api_token",
           "iat": utcnow(),
       }
       if expires_at is not None:
           payload["exp"] = expires_at
       return jwt.encode(payload, secret_key, algorithm="HS256")
   

   Also add decode_token_payload(token, secret_key) -> Optional[dict] that
   returns the full decoded payload (including token_type, jti) or None.
   It must pass options={"verify_exp": False} so that an expired api_token
   JWT can still be looked up in the DB (expiry is enforced by the service, not
   PyJWT, for api_tokens — see Section 3).

   Note: decode_access_token is left unchanged for backward compat.

2. Create songhive/users/api_tokens.py and export the following from
   its __all__:

   • issue_api_token(db, user, config, name, expires_at) -> tuple[ApiToken, str]

     • Generate jti = secrets.token_urlsafe(32).
     • Call create_api_token_jwt(user.id, config.auth.secret_key, jti, expires_at).
     • Insert an ApiToken row with that jti, name, expires_at, user_id.
     • Return (api_token_orm_instance, raw_jwt_string).
     • Raise ApiTokenError("Name already in use") if the (user_id, name) unique constraint would be violated (catch IntegrityError and re-raise).

   • get_api_token_by_jti(db, jti) -> Optional[ApiToken]

     • SELECT * FROM api_tokens WHERE jti = ? with selectinload or eager
       load unnecessary (no relationships).

   • validate_api_token(db, jti) -> Optional[ApiToken]

     • Calls get_api_token_by_jti.
     • Returns None if not found, revoked, or expired.
     • Otherwise updates last_used_at (best-effort, no error on failure) and
       returns the instance.

   • list_user_api_tokens(db, user_id, *, limit=100, offset=0) -> list[ApiToken]

     • SELECT * FROM api_tokens WHERE user_id = ? ORDER BY created_at DESC.

   • revoke_api_token(db, token_id, user_id) -> bool

     • SELECT * FROM api_tokens WHERE id = ? AND user_id = ?.
     • If not found → return False.
     • Set revoked_at = utcnow(), flush, return True.

   • count_user_api_tokens(db, user_id) -> int

   • ApiTokenError(ValueError) — custom exception with optional status_code.

3. Export ApiToken service helpers from the module's __all__.

Acceptance Criteria

• issue_api_token creates a row in the DB and returns a decodable JWT with
  token_type == "api_token" and a valid jti.
• validate_api_token returns None for revoked or expired tokens.
• revoke_api_token returns False when the token belongs to a different user.

Tests / Verification

Covered in Section 5 (test_api_tokens.py); also exercise at the service level:

• pytest tests/test_api_tokens.py -k "service"

---

Section 3: Middleware & Dependency Extension

Goal

Make _get_current_user transparently accept API tokens alongside existing
short-lived JWTs without breaking any current behaviour.

Prerequisites

Section 2 (service layer + decode_token_payload).

Steps

1. Update _get_current_user in songhive/api/deps.py:

   # illustrative flow (not full code)
   token = _token_from_credentials_or_request(request, credentials)
   if not token:
       return None
   
   # Fast path: try existing access-token decode (HS256, checks exp).
   user_id = decode_access_token(token, config.auth.secret_key)
   
   if user_id is not None:
       # Regular short-lived JWT — existing path unchanged.
       return await get_user_by_id(db, user_id) or None (check is_active)
   
   # Slow path: decode full payload without exp enforcement.
   payload = decode_token_payload(token, config.auth.secret_key)
   if payload is None or payload.get("token_type") != "api_token":
       return None   # invalid signature or unknown type
   
   jti = payload.get("jti")
   if not jti:
       return None
   
   api_token = await validate_api_token(db, jti)
   if api_token is None:
       return None
   
   return await get_user_by_id(db, api_token.user_id) — check is_active
   

   Why the two-step decode? Regular JWTs are checked by PyJWT including
   exp; if they pass that check, user_id is non-None and we never hit Redis
   or the DB. API tokens bypass PyJWT's exp check but do the DB lookup, which
   enforces expiry at the row level.

2. No changes required to get_current_user, get_current_user_optional,
   or any other dependency. The change is entirely within _get_current_user.

3. No changes required to the auth routes for login/refresh/OAuth2.

Acceptance Criteria

• An API token JWT passes get_current_user and returns the correct User.
• A revoked or expired API token returns None from _get_current_user (→ 401).
• A login access token (short-lived) still bypasses the DB lookup entirely.
• No regression on existing auth tests.

Tests / Verification

• Run full test suite: pytest --tb=short
• Specifically: pytest tests/test_auth.py tests/test_tokens.py

---

Section 4: HTTP Routes (songhive/api/routes/api_tokens.py)

Goal

Expose CRUD endpoints for API tokens, scoped to the authenticated user.

Prerequisites

Section 3 (working middleware).

Steps

1. Create songhive/api/routes/api_tokens.py with router = APIRouter(prefix="/auth/api-tokens", tags=["API Tokens"]).

2. Pydantic schemas:

   • ApiTokenCreateRequest: name: str (1–128), expires_at: Optional[datetime] = None
     — validate that expires_at, when provided, is in the future.
   • ApiTokenCreateResponse: id, name, token (raw JWT), expires_at, created_at
   • ApiTokenSummary: id, name, expires_at, last_used_at, created_at,
     is_active — no token field.
   • ApiTokenListResponse: items: list[ApiTokenSummary], total: int
   • RevokeApiTokenResponse: success: bool = True

3. Endpoints:

   POST /auth/api-tokens (requires get_current_user, rate-limited)

   • Deserialize ApiTokenCreateRequest.
   • Call issue_api_token(db, user, config, name, expires_at).
   • On ApiTokenError → 409 Conflict.
   • Return ApiTokenCreateResponse with the raw JWT.

   GET /auth/api-tokens (requires get_current_user)

   • Optional query params: limit (default 100, max 500), offset (default 0).
   • Call list_user_api_tokens + count_user_api_tokens.
   • Return ApiTokenListResponse.

   DELETE /auth/api-tokens/{token_id} (requires get_current_user)

   • Call revoke_api_token(db, token_id, user.id).
   • Returns 404 if False (not found or not owned by caller).
   • Returns RevokeApiTokenResponse(success=True) on success.

4. Register the router in the main API router (wherever auth.router is
   included — typically songhive/api/routes/__init__.py or songhive/api/app.py).
   Include api_tokens.router alongside the existing auth.router.

Acceptance Criteria

• POST /auth/api-tokens returns a JWT that works as a Bearer credential.
• GET /auth/api-tokens never returns the raw token.
• DELETE /auth/api-tokens/{token_id} → subsequent requests with that token get 401.
• Requests from a different user to delete another user's token get 404.
• expires_at in the past is rejected at create time with 422.

Tests / Verification

• pytest tests/test_api_tokens.py -k "route"
• Manual smoke: httpx or curl against a local dev server.

---

Section 5: Tests (tests/test_api_tokens.py)

Goal

Full test coverage for the service layer and HTTP routes.

Prerequisites

Sections 1–4.

Steps

1. Fixtures — reuse db_session, config, fake_redis, client
   (AsyncClient against the FastAPI app) from the existing test suite.
   Add a api_token_user fixture (a live User row + a created API token).

2. Service-level tests (no HTTP):

   Test	Assertion
   test_issue_api_token_returns_jwt_and_orm	JWT decodes; jti matches DB row
   test_issue_api_token_duplicate_name_raises	ApiTokenError on duplicate name per user
   test_validate_api_token_returns_none_when_revoked	after revoke_api_token, validate → None
   test_validate_api_token_returns_none_when_expired	set expires_at in the past; validate → None
   test_validate_api_token_updates_last_used_at	last_used_at is set after validation
   test_revoke_api_token_wrong_user_returns_false	revoke with different user_id → False

3. Middleware/deps tests:

   Test	Assertion
   test_api_token_authenticates_request	Bearer=api_token → 200, correct user resolved
   test_api_token_revoked_returns_401	revoke → next request → 401
   test_api_token_expired_returns_401	expired token → 401
   test_login_access_token_still_works	regular JWT unaffected

4. Route-level tests (HTTP via AsyncClient):

   Test	Assertion
   test_create_api_token_returns_jwt	POST → 201, body contains token
   test_create_api_token_no_expiry	expires_at=null → accepted
   test_create_api_token_past_expiry_rejected	expires_at in the past → 422
   test_list_api_tokens_no_token_in_response	GET → no token field in items
   test_delete_api_token_revokes_it	DELETE → 200; subsequent auth → 401
   test_delete_api_token_other_user_404	different user → 404
   test_unauthenticated_create_returns_401	no Bearer → 401

5. Run pytest tests/test_api_tokens.py -v and confirm all green.

---

Integration and Final Verification

After all sections are implemented:

1. Run the full test suite:

   pytest --tb=short -q
   

2. Type-check:

   mypy songhive/
   

3. Lint:

   ruff check songhive/ tests/
   

4. Manual end-to-end smoke (local server):

# Create a token
TOKEN=$(curl -s -X POST http://localhost:8000/api/v1/auth/api-tokens \
  -H "Authorization: Bearer $LOGIN_JWT" \
  -H "Content-Type: application/json" \
  -d '{"name": "my-cli-token"}' | jq -r '.token')

# Use the token
curl -s http://localhost:8000/api/v1/users/me \
  -H "Authorization: Bearer $TOKEN"

# List tokens
curl -s http://localhost:8000/api/v1/auth/api-tokens \
  -H "Authorization: Bearer $LOGIN_JWT"

# Revoke the token (get id from list)
curl -s -X DELETE http://localhost:8000/api/v1/auth/api-tokens/$TOKEN_ID \
  -H "Authorization: Bearer $LOGIN_JWT"

# Confirm revoked → 401
curl -s http://localhost:8000/api/v1/users/me \
  -H "Authorization: Bearer $TOKEN"

5. Confirm migration round-trip:

alembic downgrade -1 && alembic upgrade head

Handoff

The next step is to implement this contract. Run /implement against this file
to implement the whole plan, or hand individual ## Section N blocks to separate
agents.

# Generic API Tokens for Songhive ## Analysis of the Existing Auth System ### Current mechanisms Songhive currently ships three authentication paths, all gated through the same `Authorization: Bearer <token>` header: | Path | Token type | Backing store | Expiry | |------|-----------|---------------|--------| | **Login / Refresh** | HS256 JWT (access) + opaque refresh token | Refresh token in Redis (SHA-256 keyed) | Access: configurable minutes; refresh: configurable days | | **OAuth2 Authorization Code + PKCE** | Opaque access + refresh tokens | Both in Redis | Governed by `auth.access_token_expiry_minutes` / `refresh_token_expiry_days` | | **OAuth2 Client Credentials** | _(not implemented, grant type rejected)_ | – | – | #### JWT infrastructure (`songhive/api/middleware/auth.py`) `create_access_token(user_id, secret_key, expires_minutes=15)` builds a minimal payload `{sub, iat, [exp]}` and signs it with HS256. When `expires_minutes=None` the `exp` claim is omitted — the function already supports non-expiring tokens. `decode_access_token(token, secret_key)` returns only the `sub` (user id) or `None`; it discards all other claims. #### Dependency resolution (`songhive/api/deps.py`) `_get_current_user` calls `decode_access_token` → `get_user_by_id`. It knows nothing beyond the `sub` claim: there is no notion of token type, origin, or revocation status for access tokens. Revocation today is only possible for opaque refresh tokens stored in Redis. #### What is missing for API tokens 1. **Persistent record** — a DB row the user can list, name, and revoke. 2. **Unique token identity** (`jti` claim) so the DB record can be looked up during request validation without storing the entire JWT. 3. **Absolute expiry** — API tokens need a calendar date, not a relative duration in minutes. 4. **Revocation** — JWTs are stateless; revocation requires a DB check keyed on `jti`. 5. **Type discrimination** — `token_type: "api_token"` claim inside the JWT so the middleware can route to the DB check only for this token class, leaving ordinary short-lived access tokens unaffected. 6. **CRUD API** — endpoints for creating, listing, and revoking tokens. --- ## Implementation Plan ### Goal Extend Songhive so authenticated users can create long-lived API tokens that are issued as HS256 JWTs, carry an optional absolute expiry, are stored in the database for audit and revocation, and are validated transparently by the existing Bearer-token middleware. ### Acceptance Criteria - A user can `POST /auth/api-tokens` to create a named token; the raw JWT is returned **once** in the response. - The token works immediately as a `Bearer` credential against any protected endpoint. - A user can `GET /auth/api-tokens` to list their tokens (metadata only; raw JWT is never returned after creation). - A user can `DELETE /auth/api-tokens/{token_id}` to revoke a specific token; revoked tokens are rejected within the same request cycle (no delay). - Tokens with an `expires_at` in the past are rejected at auth time. - Tokens with `expires_at: null` never expire unless explicitly revoked. - Existing login / refresh / OAuth2 flows are unaffected. - All new code is covered by unit and integration tests. ### Assumptions and Constraints - Tech stack: Python 3.14, FastAPI, SQLAlchemy 2 async, Alembic, Redis, `PyJWT`, `bcrypt` — all already present. - The `jti` claim is stored in plain form in the DB (no additional hashing needed; the JWT itself is the secret, not the `jti`). - Revocation is checked synchronously against the DB on every API-token request. A Redis cache layer for `jti` blocklisting is optional and explicitly out of scope for this plan. - `last_used_at` is updated optimistically (fire-and-forget, acceptable to miss under concurrent load). - Token names must be unique per user. - Admin users may revoke any user's token via the existing admin route namespace (also out of scope; admin APIs can call the service layer directly). - Migrations use Alembic autogenerate (`alembic revision --autogenerate`); a skeleton command is included in each migration step. --- ### Section Map | # | Section | Depends on | Output | |---|---------|-----------|--------| | 1 | `ApiToken` DB model | — | `songhive/models/api_token.py`, migration | | 2 | API token service | Section 1 | `songhive/users/api_tokens.py` | | 3 | Middleware & deps extension | Section 2 | edits to `middleware/auth.py`, `api/deps.py` | | 4 | HTTP routes | Section 3 | `songhive/api/routes/api_tokens.py`, router registration | | 5 | Tests | Sections 1–4 | `tests/test_api_tokens.py` | --- ## Section 1: `ApiToken` SQLAlchemy Model + Migration ### Goal Define the persistent record for an API token and add it to the database. ### Prerequisites None. ### Steps 1. **Create `songhive/models/api_token.py`** with a class `ApiToken(Base)`. Required columns: | Column | Type | Constraints | |--------|------|-------------| | `id` | `str` (UUID) | PK — inherited from `Base` | | `user_id` | `str` | FK → `users.id` `ON DELETE CASCADE`, indexed | | `jti` | `str(64)` | `UNIQUE`, indexed — the JWT ID claim value | | `name` | `str(128)` | `NOT NULL` — human label | | `expires_at` | `datetime` (tz-aware) | nullable — `None` = never expires | | `revoked_at` | `datetime` (tz-aware) | nullable — `None` = active | | `last_used_at` | `datetime` (tz-aware) | nullable | | `created_at` / `updated_at` | inherited from `Base` | — | Add a `UniqueConstraint("user_id", "name")` so names are unique per user. Add a convenience property `is_active` → `bool`: `revoked_at is None and (expires_at is None or expires_at > utcnow())`. 2. **Export `ApiToken`** from `songhive/models/__init__.py` so Alembic can detect it during autogenerate. 3. **Generate an Alembic migration**: ``` alembic revision --autogenerate -m "add api_tokens table" ``` Review the generated file; confirm `op.create_table("api_tokens", ...)` with all columns and a FK to `users`. 4. **Apply the migration** in local dev: ``` alembic upgrade head ``` ### Acceptance Criteria - `alembic upgrade head` completes without errors. - `alembic downgrade -1` removes the table cleanly. - `ApiToken` can be imported from `songhive.models`. ### Tests / Verification No unit tests for this section; table creation is exercised by Sections 2 & 5. --- ## Section 2: API Token Service (`songhive/users/api_tokens.py`) ### Goal Encapsulate all API token business logic: JWT creation, DB persistence, validation, and revocation. ### Prerequisites Section 1 (the `ApiToken` model). ### Steps 1. **Add `create_api_token_jwt(user_id, secret_key, jti, expires_at)`** to `songhive/api/middleware/auth.py` (alongside `create_access_token`). ```text # illustrative pseudo-signature def create_api_token_jwt( user_id: str, secret_key: str, jti: str, expires_at: Optional[datetime], # absolute UTC datetime or None ) -> str: payload = { "sub": user_id, "jti": jti, "token_type": "api_token", "iat": utcnow(), } if expires_at is not None: payload["exp"] = expires_at return jwt.encode(payload, secret_key, algorithm="HS256") ``` Also add `decode_token_payload(token, secret_key) -> Optional[dict]` that returns the full decoded payload (including `token_type`, `jti`) or `None`. It must pass `options={"verify_exp": False}` so that an expired api_token JWT can still be looked up in the DB (expiry is enforced by the service, not PyJWT, for api_tokens — see Section 3). > **Note:** `decode_access_token` is left unchanged for backward compat. 2. **Create `songhive/users/api_tokens.py`** and export the following from its `__all__`: - `issue_api_token(db, user, config, name, expires_at) -> tuple[ApiToken, str]` - Generate `jti = secrets.token_urlsafe(32)`. - Call `create_api_token_jwt(user.id, config.auth.secret_key, jti, expires_at)`. - Insert an `ApiToken` row with that `jti`, `name`, `expires_at`, `user_id`. - Return `(api_token_orm_instance, raw_jwt_string)`. - Raise `ApiTokenError("Name already in use")` if the `(user_id, name)` unique constraint would be violated (catch `IntegrityError` and re-raise). - `get_api_token_by_jti(db, jti) -> Optional[ApiToken]` - `SELECT * FROM api_tokens WHERE jti = ?` with `selectinload` or eager load unnecessary (no relationships). - `validate_api_token(db, jti) -> Optional[ApiToken]` - Calls `get_api_token_by_jti`. - Returns `None` if not found, revoked, or expired. - Otherwise updates `last_used_at` (best-effort, no error on failure) and returns the instance. - `list_user_api_tokens(db, user_id, *, limit=100, offset=0) -> list[ApiToken]` - `SELECT * FROM api_tokens WHERE user_id = ? ORDER BY created_at DESC`. - `revoke_api_token(db, token_id, user_id) -> bool` - `SELECT * FROM api_tokens WHERE id = ? AND user_id = ?`. - If not found → return `False`. - Set `revoked_at = utcnow()`, flush, return `True`. - `count_user_api_tokens(db, user_id) -> int` - `ApiTokenError(ValueError)` — custom exception with optional `status_code`. 3. **Export** `ApiToken` service helpers from the module's `__all__`. ### Acceptance Criteria - `issue_api_token` creates a row in the DB and returns a decodable JWT with `token_type == "api_token"` and a valid `jti`. - `validate_api_token` returns `None` for revoked or expired tokens. - `revoke_api_token` returns `False` when the token belongs to a different user. ### Tests / Verification Covered in Section 5 (`test_api_tokens.py`); also exercise at the service level: - `pytest tests/test_api_tokens.py -k "service"` --- ## Section 3: Middleware & Dependency Extension ### Goal Make `_get_current_user` transparently accept API tokens alongside existing short-lived JWTs without breaking any current behaviour. ### Prerequisites Section 2 (service layer + `decode_token_payload`). ### Steps 1. **Update `_get_current_user` in `songhive/api/deps.py`**: ```text # illustrative flow (not full code) token = _token_from_credentials_or_request(request, credentials) if not token: return None # Fast path: try existing access-token decode (HS256, checks exp). user_id = decode_access_token(token, config.auth.secret_key) if user_id is not None: # Regular short-lived JWT — existing path unchanged. return await get_user_by_id(db, user_id) or None (check is_active) # Slow path: decode full payload without exp enforcement. payload = decode_token_payload(token, config.auth.secret_key) if payload is None or payload.get("token_type") != "api_token": return None # invalid signature or unknown type jti = payload.get("jti") if not jti: return None api_token = await validate_api_token(db, jti) if api_token is None: return None return await get_user_by_id(db, api_token.user_id) — check is_active ``` **Why the two-step decode?** Regular JWTs are checked by PyJWT including `exp`; if they pass that check, `user_id` is non-None and we never hit Redis or the DB. API tokens bypass PyJWT's `exp` check but do the DB lookup, which enforces expiry at the row level. 2. **No changes required** to `get_current_user`, `get_current_user_optional`, or any other dependency. The change is entirely within `_get_current_user`. 3. **No changes required** to the auth routes for login/refresh/OAuth2. ### Acceptance Criteria - An API token JWT passes `get_current_user` and returns the correct `User`. - A revoked or expired API token returns `None` from `_get_current_user` (→ 401). - A login access token (short-lived) still bypasses the DB lookup entirely. - No regression on existing auth tests. ### Tests / Verification - Run full test suite: `pytest --tb=short` - Specifically: `pytest tests/test_auth.py tests/test_tokens.py` --- ## Section 4: HTTP Routes (`songhive/api/routes/api_tokens.py`) ### Goal Expose CRUD endpoints for API tokens, scoped to the authenticated user. ### Prerequisites Section 3 (working middleware). ### Steps 1. **Create `songhive/api/routes/api_tokens.py`** with `router = APIRouter(prefix="/auth/api-tokens", tags=["API Tokens"])`. 2. **Pydantic schemas**: - `ApiTokenCreateRequest`: `name: str (1–128)`, `expires_at: Optional[datetime] = None` — validate that `expires_at`, when provided, is in the future. - `ApiTokenCreateResponse`: `id`, `name`, `token` (raw JWT), `expires_at`, `created_at` - `ApiTokenSummary`: `id`, `name`, `expires_at`, `last_used_at`, `created_at`, `is_active` — **no `token` field**. - `ApiTokenListResponse`: `items: list[ApiTokenSummary]`, `total: int` - `RevokeApiTokenResponse`: `success: bool = True` 3. **Endpoints**: **`POST /auth/api-tokens`** (requires `get_current_user`, rate-limited) - Deserialize `ApiTokenCreateRequest`. - Call `issue_api_token(db, user, config, name, expires_at)`. - On `ApiTokenError` → 409 Conflict. - Return `ApiTokenCreateResponse` with the raw JWT. **`GET /auth/api-tokens`** (requires `get_current_user`) - Optional query params: `limit` (default 100, max 500), `offset` (default 0). - Call `list_user_api_tokens` + `count_user_api_tokens`. - Return `ApiTokenListResponse`. **`DELETE /auth/api-tokens/{token_id}`** (requires `get_current_user`) - Call `revoke_api_token(db, token_id, user.id)`. - Returns 404 if `False` (not found or not owned by caller). - Returns `RevokeApiTokenResponse(success=True)` on success. 4. **Register the router** in the main API router (wherever `auth.router` is included — typically `songhive/api/routes/__init__.py` or `songhive/api/app.py`). Include `api_tokens.router` alongside the existing `auth.router`. ### Acceptance Criteria - `POST /auth/api-tokens` returns a JWT that works as a Bearer credential. - `GET /auth/api-tokens` never returns the raw token. - `DELETE /auth/api-tokens/{token_id}` → subsequent requests with that token get 401. - Requests from a different user to delete another user's token get 404. - `expires_at` in the past is rejected at create time with 422. ### Tests / Verification - `pytest tests/test_api_tokens.py -k "route"` - Manual smoke: `httpx` or `curl` against a local dev server. --- ## Section 5: Tests (`tests/test_api_tokens.py`) ### Goal Full test coverage for the service layer and HTTP routes. ### Prerequisites Sections 1–4. ### Steps 1. **Fixtures** — reuse `db_session`, `config`, `fake_redis`, `client` (AsyncClient against the FastAPI app) from the existing test suite. Add a `api_token_user` fixture (a live `User` row + a created API token). 2. **Service-level tests** (no HTTP): | Test | Assertion | |------|-----------| | `test_issue_api_token_returns_jwt_and_orm` | JWT decodes; `jti` matches DB row | | `test_issue_api_token_duplicate_name_raises` | `ApiTokenError` on duplicate name per user | | `test_validate_api_token_returns_none_when_revoked` | after `revoke_api_token`, `validate` → `None` | | `test_validate_api_token_returns_none_when_expired` | set `expires_at` in the past; `validate` → `None` | | `test_validate_api_token_updates_last_used_at` | `last_used_at` is set after validation | | `test_revoke_api_token_wrong_user_returns_false` | revoke with different `user_id` → `False` | 3. **Middleware/deps tests**: | Test | Assertion | |------|-----------| | `test_api_token_authenticates_request` | Bearer=api_token → 200, correct user resolved | | `test_api_token_revoked_returns_401` | revoke → next request → 401 | | `test_api_token_expired_returns_401` | expired token → 401 | | `test_login_access_token_still_works` | regular JWT unaffected | 4. **Route-level tests** (HTTP via `AsyncClient`): | Test | Assertion | |------|-----------| | `test_create_api_token_returns_jwt` | POST → 201, body contains `token` | | `test_create_api_token_no_expiry` | `expires_at=null` → accepted | | `test_create_api_token_past_expiry_rejected` | `expires_at` in the past → 422 | | `test_list_api_tokens_no_token_in_response` | GET → no `token` field in items | | `test_delete_api_token_revokes_it` | DELETE → 200; subsequent auth → 401 | | `test_delete_api_token_other_user_404` | different user → 404 | | `test_unauthenticated_create_returns_401` | no Bearer → 401 | 5. Run `pytest tests/test_api_tokens.py -v` and confirm all green. --- ## Integration and Final Verification After all sections are implemented: 1. Run the full test suite: ``` pytest --tb=short -q ``` 2. Type-check: ``` mypy songhive/ ``` 3. Lint: ``` ruff check songhive/ tests/ ``` 4. Manual end-to-end smoke (local server): ```bash # Create a token TOKEN=$(curl -s -X POST http://localhost:8000/api/v1/auth/api-tokens \ -H "Authorization: Bearer $LOGIN_JWT" \ -H "Content-Type: application/json" \ -d '{"name": "my-cli-token"}' | jq -r '.token') # Use the token curl -s http://localhost:8000/api/v1/users/me \ -H "Authorization: Bearer $TOKEN" # List tokens curl -s http://localhost:8000/api/v1/auth/api-tokens \ -H "Authorization: Bearer $LOGIN_JWT" # Revoke the token (get id from list) curl -s -X DELETE http://localhost:8000/api/v1/auth/api-tokens/$TOKEN_ID \ -H "Authorization: Bearer $LOGIN_JWT" # Confirm revoked → 401 curl -s http://localhost:8000/api/v1/users/me \ -H "Authorization: Bearer $TOKEN" ``` 5. Confirm migration round-trip: ``` alembic downgrade -1 && alembic upgrade head ``` ## Handoff The next step is to implement this contract. Run `/implement` against this file to implement the whole plan, or hand individual `## Section N` blocks to separate agents.

blacklight added the

new feature

api

backend

auth

labels 2026-08-23 17:57:58 +02:00

blacklight referenced this issue from a pull request that will close it, 2026-08-23 20:17:15 +02:00

feat: add user API tokens for long-lived auth #14

blacklight closed this issue 2026-08-24 00:51:11 +02:00

Sign in to join this conversation.

No Branch/Tag specified

Branches Tags

main

v0.0.15

v0.0.14

v0.0.13

v0.0.12

v0.0.11

v0.0.10

v0.0.9

v0.0.8

v0.0.7

v0.0.6

v0.0.5

v0.0.4

v0.0.3

v0.0.2

Labels

Clear labels   

api

API-related changes

  

auth

Authorization and authentication

  

backend

Backend-related tasks

  

bug

Something is not working

  

duplicate

This issue or pull request already exists

  

enhancement

Enhancement over an existing feature

  

federation

ActivityPub federation tasks

  

frontend

Frontend-related tasks

  

help wanted

Need some help

  

invalid

Something is wrong

  

new feature

New feature requests

  

question

More information is needed

  

storage

Media storage tasks

  

stream

Media streaming tasks

  

wontfix

This won't be fixed

No labels

api

auth

backend

bug

duplicate

enhancement

federation

frontend

help wanted

invalid

new feature

question

storage

stream

wontfix

Milestone

Clear milestone

No items

No milestone

Projects

Clear projects

No items

No project

Assignees

Clear assignees

No assignees

1 participant

Notifications

Subscribe

Due date

The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set

Reference

blacklight/songhive#13

No description provided.