222 lines
8.1 KiB
Python
222 lines
8.1 KiB
Python
import secrets
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
from urllib.parse import urlencode
|
|
|
|
import httpx
|
|
import jwt
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
|
|
from fastapi.responses import RedirectResponse
|
|
from google.auth.transport import requests as google_requests
|
|
from google.oauth2 import id_token
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .auth import COOKIE_NAME, create_token, current_user, normalize_email, password_hash
|
|
from .config import settings
|
|
from .database import get_db
|
|
from .models import OidcIdentity, PasswordCredential, UserAccount
|
|
from .schemas import LoginRequest, RegisterRequest, SetPasswordRequest, UserResponse
|
|
|
|
router = APIRouter(prefix="/api/v1/auth", tags=["authentication"])
|
|
GOOGLE_ISSUER = "https://accounts.google.com"
|
|
|
|
|
|
def public_user(user: UserAccount) -> UserResponse:
|
|
return UserResponse(
|
|
id=str(user.id),
|
|
email=user.email,
|
|
name=user.display_name,
|
|
avatar_url=user.avatar_url,
|
|
email_verified=user.email_verified,
|
|
has_password=user.password_credential is not None,
|
|
providers=["google" for identity in user.oidc_identities if identity.issuer == GOOGLE_ISSUER],
|
|
)
|
|
|
|
|
|
def set_session(response: Response, user: UserAccount) -> None:
|
|
response.set_cookie(
|
|
COOKIE_NAME,
|
|
create_token(user.id),
|
|
httponly=True,
|
|
secure=settings.auth_cookie_secure,
|
|
samesite="lax",
|
|
max_age=settings.access_token_minutes * 60,
|
|
path="/",
|
|
)
|
|
|
|
|
|
@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
|
|
def register(payload: RegisterRequest, response: Response, db: Session = Depends(get_db)) -> UserResponse:
|
|
email = normalize_email(str(payload.email))
|
|
user = db.scalar(select(UserAccount).where(UserAccount.email == email))
|
|
if user:
|
|
detail = (
|
|
"Sign in with Google, then add a password from Sign-in & Security"
|
|
if user.oidc_identities and not user.password_credential
|
|
else "An account already exists"
|
|
)
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=detail)
|
|
user = UserAccount(email=email, display_name=payload.name.strip(), email_verified=False)
|
|
db.add(user)
|
|
db.flush()
|
|
user.password_credential = PasswordCredential(password_hash=password_hash.hash(payload.password))
|
|
db.commit()
|
|
db.refresh(user)
|
|
set_session(response, user)
|
|
return public_user(user)
|
|
|
|
|
|
@router.post("/login", response_model=UserResponse)
|
|
def login(payload: LoginRequest, response: Response, db: Session = Depends(get_db)) -> UserResponse:
|
|
user = db.scalar(select(UserAccount).where(UserAccount.email == normalize_email(str(payload.email))))
|
|
if not user or not user.password_credential:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid email or password")
|
|
if not password_hash.verify(payload.password, user.password_credential.password_hash):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid email or password")
|
|
set_session(response, user)
|
|
return public_user(user)
|
|
|
|
|
|
@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
|
|
def logout(response: Response) -> None:
|
|
response.delete_cookie(COOKIE_NAME, path="/")
|
|
|
|
|
|
@router.get("/me", response_model=UserResponse)
|
|
def me(user: UserAccount = Depends(current_user)) -> UserResponse:
|
|
return public_user(user)
|
|
|
|
|
|
@router.post("/password", response_model=UserResponse)
|
|
def configure_password(
|
|
payload: SetPasswordRequest,
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> UserResponse:
|
|
if user.password_credential:
|
|
if not payload.current_password or not password_hash.verify(
|
|
payload.current_password, user.password_credential.password_hash
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Current password is incorrect",
|
|
)
|
|
user.password_credential.password_hash = password_hash.hash(payload.password)
|
|
else:
|
|
user.password_credential = PasswordCredential(password_hash=password_hash.hash(payload.password))
|
|
db.commit()
|
|
db.refresh(user)
|
|
return public_user(user)
|
|
|
|
|
|
@router.get("/google/start")
|
|
def google_start(next_path: str = Query(default="/console", pattern=r"^/")) -> RedirectResponse:
|
|
if not settings.google_client_id or not settings.google_client_secret:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="Google sign-in is not configured",
|
|
)
|
|
now = datetime.now(timezone.utc)
|
|
nonce = secrets.token_urlsafe(24)
|
|
state = jwt.encode(
|
|
{
|
|
"purpose": "google_oauth",
|
|
"nonce": nonce,
|
|
"next": next_path,
|
|
"iat": now,
|
|
"exp": now + timedelta(minutes=10),
|
|
},
|
|
settings.auth_secret,
|
|
algorithm="HS256",
|
|
)
|
|
query = urlencode(
|
|
{
|
|
"client_id": settings.google_client_id,
|
|
"redirect_uri": settings.google_redirect_uri,
|
|
"response_type": "code",
|
|
"scope": "openid email profile",
|
|
"state": state,
|
|
"nonce": nonce,
|
|
"prompt": "select_account",
|
|
}
|
|
)
|
|
return RedirectResponse(f"https://accounts.google.com/o/oauth2/v2/auth?{query}")
|
|
|
|
|
|
@router.get("/google/callback")
|
|
def google_callback(
|
|
code: str,
|
|
state: str,
|
|
db: Session = Depends(get_db),
|
|
) -> RedirectResponse:
|
|
try:
|
|
state_data = jwt.decode(state, settings.auth_secret, algorithms=["HS256"])
|
|
if state_data.get("purpose") != "google_oauth":
|
|
raise ValueError("invalid state")
|
|
except (jwt.PyJWTError, ValueError) as exc:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid OAuth state") from exc
|
|
|
|
token_response = httpx.post(
|
|
"https://oauth2.googleapis.com/token",
|
|
data={
|
|
"code": code,
|
|
"client_id": settings.google_client_id,
|
|
"client_secret": settings.google_client_secret,
|
|
"redirect_uri": settings.google_redirect_uri,
|
|
"grant_type": "authorization_code",
|
|
},
|
|
timeout=15,
|
|
)
|
|
if token_response.is_error:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Google authentication failed")
|
|
|
|
try:
|
|
claims = id_token.verify_oauth2_token(
|
|
token_response.json()["id_token"],
|
|
google_requests.Request(),
|
|
settings.google_client_id,
|
|
)
|
|
if claims.get("nonce") != state_data["nonce"] or not claims.get("email_verified"):
|
|
raise ValueError("unverified Google identity")
|
|
except (KeyError, ValueError) as exc:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid Google identity") from exc
|
|
|
|
subject = str(claims["sub"])
|
|
identity = db.scalar(
|
|
select(OidcIdentity).where(
|
|
OidcIdentity.issuer == GOOGLE_ISSUER,
|
|
OidcIdentity.subject == subject,
|
|
)
|
|
)
|
|
if identity:
|
|
user = identity.user
|
|
identity.last_login_at = datetime.now(timezone.utc)
|
|
else:
|
|
email = normalize_email(str(claims["email"]))
|
|
user = db.scalar(select(UserAccount).where(UserAccount.email == email))
|
|
if not user:
|
|
user = UserAccount(
|
|
email=email,
|
|
display_name=str(claims.get("name") or email.split("@")[0]),
|
|
avatar_url=claims.get("picture"),
|
|
email_verified=True,
|
|
)
|
|
db.add(user)
|
|
db.flush()
|
|
else:
|
|
user.email_verified = True
|
|
user.avatar_url = user.avatar_url or claims.get("picture")
|
|
db.add(
|
|
OidcIdentity(
|
|
user_id=user.id,
|
|
issuer=GOOGLE_ISSUER,
|
|
subject=subject,
|
|
provider_email=email,
|
|
)
|
|
)
|
|
db.commit()
|
|
destination = str(state_data.get("next", "/console"))
|
|
response = RedirectResponse(f"{settings.frontend_origin}{destination}")
|
|
set_session(response, user)
|
|
return response
|