"""Optional OIDC auth module."""
# pyright: reportUnusedFunction=false
import logging
from datetime import datetime
from typing import Annotated
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from pydantic import ValidationError
from sqlalchemy.ext.asyncio import AsyncSession
from sshecret_admin.auth import create_access_token, create_refresh_token
from sshecret_admin.auth.authentication import generate_user_info, handle_oidc_claim
from sshecret_admin.auth.exceptions import AuthenticationFailedError
from sshecret_admin.auth.oidc import AdminOidc
from sshecret_admin.frontend.exceptions import RedirectException
from sshecret_admin.services import AdminBackend
from starlette.datastructures import URL
from sshecret.backend.models import Operation
from ..dependencies import FrontendDependencies
LOG = logging.getLogger(__name__)
async def audit_login_failure(
admin: AdminBackend,
error_message: str,
request: Request,
) -> None:
"""Write login failure to audit log."""
origin: str | None = None
if request.client:
origin = request.client.host
await admin.write_audit_message(
operation=Operation.DENY,
message="Login failed",
origin=origin or "UNKNOWN",
provider_error_message=error_message,
)
def create_router(dependencies: FrontendDependencies) -> APIRouter:
"""Create auth router."""
app = APIRouter()
def get_oidc_client() -> AdminOidc:
"""Get OIDC client dependency."""
if not dependencies.settings.oidc:
raise RuntimeError("OIDC authentication not configured.")
oidc = AdminOidc(dependencies.settings.oidc)
return oidc
@app.get("/oidc/login")
async def oidc_login(
request: Request, oidc: Annotated[AdminOidc, Depends(get_oidc_client)]
) -> RedirectResponse:
"""Redirect to oidc login."""
redirect_url = request.url_for("oidc_auth")
return await oidc.start_auth(request, redirect_url)
@app.get("/oidc/auth")
async def oidc_auth(
request: Request,
session: Annotated[AsyncSession, Depends(dependencies.get_async_session)],
admin: Annotated[AdminBackend, Depends(dependencies.get_admin_backend)],
oidc: Annotated[AdminOidc, Depends(get_oidc_client)],
):
"""Handle OIDC auth callback."""
try:
claims = await oidc.handle_auth_callback(request)
except AuthenticationFailedError as error:
raise RedirectException(
to=URL("/login").include_query_params(
error_title="Login error from external provider",
error_message=str(error),
)
)
except ValidationError as error:
LOG.error("Validation error: %s", error, exc_info=True)
raise RedirectException(
to=URL("/login").include_query_params(
error_title="Error parsing claim",
error_message="One or more required parameters were not included in the claim.",
)
)
# We now have a IdentityClaims object.
# We need to check if this matches an existing user, or we need to create a new one.
user = await handle_oidc_claim(session, claims)
user.last_login = datetime.now()
session.add(user)
await session.commit()
# Set cookies
token_data: dict[str, str] = {"sub": claims.sub}
access_token = create_access_token(
dependencies.settings, data=token_data, provider=claims.provider
)
refresh_token = create_refresh_token(
dependencies.settings, data=token_data, provider=claims.provider
)
user_info = generate_user_info(user)
response = HTMLResponse("""
Login successful. Redirecting...
""")
response.set_cookie(
"access_token",
value=access_token,
httponly=True,
secure=False,
samesite="strict",
)
response.set_cookie(
"refresh_token",
value=refresh_token,
httponly=True,
secure=False,
samesite="strict",
)
origin = "UNKNOWN"
if request.client:
origin = request.client.host
await admin.write_audit_message(
operation=Operation.LOGIN,
message="Logged in to admin frontend",
origin=origin,
username=user_info.display_name,
oidc=claims.provider,
)
return response
return app