165 lines
5.6 KiB
Python
165 lines
5.6 KiB
Python
"""Frontend router."""
|
|
|
|
# pyright: reportUnusedFunction=false
|
|
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
|
|
from jinja2_fragments.fastapi import Jinja2Blocks
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import Session
|
|
from sshecret_admin.auth.authentication import generate_user_info
|
|
from sshecret_admin.auth.models import AuthProvider, IdentityClaims, LocalUserInfo
|
|
from starlette.datastructures import URL
|
|
|
|
|
|
from sshecret_admin.auth import PasswordDB, User, decode_token
|
|
from sshecret_admin.auth.constants import LOCAL_ISSUER
|
|
|
|
from sshecret_admin.core.dependencies import BaseDependencies
|
|
from sshecret_admin.services.admin_backend import AdminBackend
|
|
from sshecret_admin.core.db import DatabaseSessionManager
|
|
|
|
from .dependencies import FrontendDependencies
|
|
from .exceptions import RedirectException
|
|
from .views import audit, auth, clients, index, secrets, oidc_auth
|
|
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
access_token = "access_token"
|
|
refresh_token = "refresh_token"
|
|
|
|
|
|
def create_router(dependencies: BaseDependencies) -> APIRouter:
|
|
"""Create frontend router."""
|
|
|
|
app = APIRouter(include_in_schema=False)
|
|
|
|
script_path = Path(os.path.dirname(os.path.realpath(__file__)))
|
|
|
|
template_path = script_path / "templates"
|
|
|
|
templates = Jinja2Blocks(directory=template_path)
|
|
|
|
async def get_admin_backend(
|
|
session: Annotated[Session, Depends(dependencies.get_db_session)],
|
|
):
|
|
"""Get admin backend API."""
|
|
password_db = session.scalars(
|
|
select(PasswordDB).where(PasswordDB.id == 1)
|
|
).first()
|
|
if not password_db:
|
|
raise HTTPException(
|
|
500, detail="Error: The password manager has not yet been set up."
|
|
)
|
|
admin = AdminBackend(dependencies.settings, password_db.encrypted_password)
|
|
yield admin
|
|
|
|
def get_identity_claims(request: Request) -> IdentityClaims:
|
|
"""Get identity claim from session."""
|
|
token = request.cookies.get("access_token")
|
|
next = URL("/refresh").include_query_params(next=request.url.path)
|
|
credentials_error = RedirectException(to=next)
|
|
if not token:
|
|
raise credentials_error
|
|
claims = decode_token(dependencies.settings, token)
|
|
if not claims:
|
|
raise credentials_error
|
|
return claims
|
|
|
|
def refresh_identity_claims(request: Request) -> IdentityClaims:
|
|
"""Get identity claim from session for refreshing the token."""
|
|
token = request.cookies.get("refresh_token")
|
|
next = URL("/login").include_query_params(next=request.url.path)
|
|
credentials_error = RedirectException(to=next)
|
|
if not token:
|
|
raise credentials_error
|
|
claims = decode_token(dependencies.settings, token)
|
|
if not claims:
|
|
raise credentials_error
|
|
return claims
|
|
|
|
async def get_login_status(request: Request) -> bool:
|
|
"""Get login status."""
|
|
token = request.cookies.get("access_token")
|
|
if not token:
|
|
return False
|
|
|
|
claims = decode_token(dependencies.settings, token)
|
|
return claims is not None
|
|
|
|
async def require_login(request: Request) -> None:
|
|
"""Enforce login requirement."""
|
|
token = request.cookies.get("access_token")
|
|
LOG.info("User has no cookie")
|
|
if not token:
|
|
url = URL("/login").include_query_params(next=request.url.path)
|
|
raise RedirectException(to=url)
|
|
is_logged_in = await get_login_status(request)
|
|
if not is_logged_in:
|
|
next = URL("/refresh").include_query_params(next=request.url.path)
|
|
raise RedirectException(to=next)
|
|
|
|
async def get_async_session():
|
|
"""Get async session."""
|
|
sessionmanager = DatabaseSessionManager(dependencies.settings.async_db_url)
|
|
async with sessionmanager.session() as session:
|
|
yield session
|
|
|
|
async def get_user_info(
|
|
request: Request, session: Annotated[AsyncSession, Depends(get_async_session)]
|
|
) -> LocalUserInfo:
|
|
"""Get User information."""
|
|
claims = get_identity_claims(request)
|
|
if claims.provider == LOCAL_ISSUER:
|
|
LOG.info("Local user, finding username %s", claims.sub)
|
|
query = (
|
|
select(User)
|
|
.where(User.username == claims.sub)
|
|
.where(User.provider == AuthProvider.LOCAL)
|
|
)
|
|
else:
|
|
query = (
|
|
select(User)
|
|
.where(User.oidc_issuer == claims.provider)
|
|
.where(User.oidc_sub == claims.sub)
|
|
)
|
|
|
|
result = await session.scalars(query)
|
|
if user := result.first():
|
|
if user.disabled:
|
|
raise RedirectException(to=URL("/logout"))
|
|
return generate_user_info(user)
|
|
|
|
next = URL("/refresh").include_query_params(next=request.url.path)
|
|
raise RedirectException(to=next)
|
|
|
|
view_dependencies = FrontendDependencies.create(
|
|
dependencies,
|
|
get_admin_backend,
|
|
templates,
|
|
refresh_identity_claims,
|
|
get_login_status,
|
|
get_user_info,
|
|
get_async_session,
|
|
require_login,
|
|
)
|
|
|
|
app.include_router(audit.create_router(view_dependencies))
|
|
app.include_router(auth.create_router(view_dependencies))
|
|
app.include_router(clients.create_router(view_dependencies))
|
|
app.include_router(index.create_router(view_dependencies))
|
|
app.include_router(secrets.create_router(view_dependencies))
|
|
if dependencies.settings.oidc:
|
|
app.include_router(oidc_auth.create_router(view_dependencies))
|
|
|
|
return app
|