135 lines
4.1 KiB
Python
135 lines
4.1 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
import httpx
|
|
|
|
from app.core.database import get_db
|
|
from app.core.security import verify_password, create_access_token, hash_password, get_current_user
|
|
from app.core.config import settings
|
|
from app.models.user import User
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class Token(BaseModel):
|
|
access_token: str
|
|
token_type: str
|
|
user_id: int
|
|
username: str
|
|
is_admin: bool
|
|
|
|
|
|
class UserOut(BaseModel):
|
|
id: int
|
|
username: str
|
|
email: Optional[str]
|
|
is_admin: bool
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
@router.post("/token", response_model=Token)
|
|
async def login(
|
|
form_data: OAuth2PasswordRequestForm = Depends(),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(User).where(User.username == form_data.username)
|
|
)
|
|
user = result.scalar_one_or_none()
|
|
|
|
if not user or not user.hashed_password:
|
|
raise HTTPException(status_code=400, detail="Invalid credentials")
|
|
if not verify_password(form_data.password, user.hashed_password):
|
|
raise HTTPException(status_code=400, detail="Invalid credentials")
|
|
|
|
token = create_access_token({"sub": str(user.id)})
|
|
return Token(
|
|
access_token=token,
|
|
token_type="bearer",
|
|
user_id=user.id,
|
|
username=user.username,
|
|
is_admin=user.is_admin,
|
|
)
|
|
|
|
|
|
@router.get("/me", response_model=UserOut)
|
|
async def get_me(current_user: User = Depends(get_current_user)):
|
|
return current_user
|
|
|
|
|
|
@router.get("/pocketid/available")
|
|
async def pocketid_available():
|
|
return {"available": bool(settings.pocketid_issuer and settings.pocketid_client_id)}
|
|
|
|
|
|
@router.get("/pocketid/login-url")
|
|
async def pocketid_login_url():
|
|
"""Return the OIDC authorization URL for PocketID."""
|
|
if not settings.pocketid_issuer:
|
|
raise HTTPException(status_code=404, detail="PocketID not configured")
|
|
|
|
params = {
|
|
"client_id": settings.pocketid_client_id,
|
|
"redirect_uri": "/api/auth/pocketid/callback",
|
|
"response_type": "code",
|
|
"scope": "openid profile email",
|
|
}
|
|
from urllib.parse import urlencode
|
|
url = f"{settings.pocketid_issuer}/authorize?{urlencode(params)}"
|
|
return {"url": url}
|
|
|
|
|
|
@router.get("/pocketid/callback")
|
|
async def pocketid_callback(code: str, db: AsyncSession = Depends(get_db)):
|
|
"""Exchange OIDC code for tokens and create/login user."""
|
|
if not settings.pocketid_issuer:
|
|
raise HTTPException(status_code=404, detail="PocketID not configured")
|
|
|
|
# Exchange code for tokens
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
f"{settings.pocketid_issuer}/token",
|
|
data={
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": "/api/auth/pocketid/callback",
|
|
"client_id": settings.pocketid_client_id,
|
|
"client_secret": settings.pocketid_client_secret,
|
|
},
|
|
)
|
|
if resp.status_code != 200:
|
|
raise HTTPException(status_code=400, detail="Token exchange failed")
|
|
|
|
tokens = resp.json()
|
|
userinfo_resp = await client.get(
|
|
f"{settings.pocketid_issuer}/userinfo",
|
|
headers={"Authorization": f"Bearer {tokens['access_token']}"},
|
|
)
|
|
userinfo = userinfo_resp.json()
|
|
|
|
sub = userinfo.get("sub")
|
|
email = userinfo.get("email")
|
|
preferred_username = userinfo.get("preferred_username") or email
|
|
|
|
result = await db.execute(select(User).where(User.pocketid_sub == sub))
|
|
user = result.scalar_one_or_none()
|
|
|
|
if not user:
|
|
user = User(
|
|
username=preferred_username,
|
|
email=email,
|
|
pocketid_sub=sub,
|
|
)
|
|
db.add(user)
|
|
await db.flush()
|
|
|
|
token = create_access_token({"sub": str(user.id)})
|
|
# Redirect to frontend with token
|
|
from fastapi.responses import RedirectResponse
|
|
return RedirectResponse(url=f"/?token={token}")
|