Robert Helewka
4a3c14d4af
Add comprehensive rule documentation for AI-assisted development covering authentication surfaces, outbound-call safety invariants, and other project conventions to guide Claude's understanding of critical system behaviors.
108 lines
3.3 KiB
Python
108 lines
3.3 KiB
Python
"""Owner-only CRUD for personal access tokens (PATs).
|
|
|
|
PATs are long-lived bearer tokens for MCP/CLI clients (Claude Desktop, Cline)
|
|
and scripted API consumers that can't refresh a short-lived Casdoor JWT. The
|
|
plaintext is shown to the caller exactly once at creation; only its SHA-256
|
|
hash is stored.
|
|
"""
|
|
|
|
import secrets
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from auth import PAT_PREFIX, OwnerUser, hash_token
|
|
from db.database import PersonalAccessToken, get_db
|
|
|
|
router = APIRouter(prefix="/api/v1/tokens", tags=["tokens"])
|
|
|
|
|
|
class TokenCreate(BaseModel):
|
|
name: str = Field(..., min_length=1, max_length=200)
|
|
|
|
|
|
class TokenOut(BaseModel):
|
|
id: str
|
|
name: str
|
|
token_prefix: str
|
|
created_at: str | None = None
|
|
last_used_at: str | None = None
|
|
expires_at: str | None = None
|
|
revoked_at: str | None = None
|
|
|
|
|
|
class TokenCreated(TokenOut):
|
|
token: str = Field(..., description="Plaintext token — shown only once. Store it now.")
|
|
|
|
|
|
def _serialize(pat: PersonalAccessToken) -> dict:
|
|
return {
|
|
"id": pat.id,
|
|
"name": pat.name,
|
|
"token_prefix": pat.token_prefix,
|
|
"created_at": pat.created_at.isoformat() if pat.created_at else None,
|
|
"last_used_at": pat.last_used_at.isoformat() if pat.last_used_at else None,
|
|
"expires_at": pat.expires_at.isoformat() if pat.expires_at else None,
|
|
"revoked_at": pat.revoked_at.isoformat() if pat.revoked_at else None,
|
|
}
|
|
|
|
|
|
@router.get("", response_model=list[TokenOut])
|
|
async def list_tokens(
|
|
user: OwnerUser,
|
|
session: AsyncSession = Depends(get_db),
|
|
) -> list[dict]:
|
|
"""List the owner's personal access tokens (no plaintext)."""
|
|
result = await session.execute(
|
|
select(PersonalAccessToken)
|
|
.where(PersonalAccessToken.user_id == user.id)
|
|
.order_by(PersonalAccessToken.created_at.desc())
|
|
)
|
|
return [_serialize(pat) for pat in result.scalars().all()]
|
|
|
|
|
|
@router.post("", response_model=TokenCreated, status_code=201)
|
|
async def create_token(
|
|
payload: TokenCreate,
|
|
user: OwnerUser,
|
|
session: AsyncSession = Depends(get_db),
|
|
) -> dict:
|
|
"""Mint a new PAT. The plaintext is returned ONCE in the response."""
|
|
plaintext = PAT_PREFIX + secrets.token_urlsafe(32)
|
|
pat = PersonalAccessToken(
|
|
id=uuid.uuid4().hex,
|
|
user_id=user.id,
|
|
name=payload.name,
|
|
token_hash=hash_token(plaintext),
|
|
token_prefix=plaintext[: len(PAT_PREFIX) + 4],
|
|
)
|
|
session.add(pat)
|
|
await session.commit()
|
|
await session.refresh(pat)
|
|
return {**_serialize(pat), "token": plaintext}
|
|
|
|
|
|
@router.delete("/{token_id}", status_code=204)
|
|
async def revoke_token(
|
|
token_id: str,
|
|
user: OwnerUser,
|
|
session: AsyncSession = Depends(get_db),
|
|
) -> None:
|
|
"""Soft-revoke a PAT (sets revoked_at)."""
|
|
result = await session.execute(
|
|
select(PersonalAccessToken).where(
|
|
PersonalAccessToken.id == token_id,
|
|
PersonalAccessToken.user_id == user.id,
|
|
)
|
|
)
|
|
pat = result.scalar_one_or_none()
|
|
if pat is None:
|
|
raise HTTPException(status_code=404, detail="Token not found")
|
|
if pat.revoked_at is None:
|
|
pat.revoked_at = datetime.now(UTC)
|
|
await session.commit()
|