diff --git a/packages/sshecret-admin/src/sshecret_admin/api/endpoints/clients.py b/packages/sshecret-admin/src/sshecret_admin/api/endpoints/clients.py index c8384f4..3267289 100644 --- a/packages/sshecret-admin/src/sshecret_admin/api/endpoints/clients.py +++ b/packages/sshecret-admin/src/sshecret_admin/api/endpoints/clients.py @@ -21,7 +21,7 @@ LOG = logging.getLogger(__name__) def create_router(dependencies: AdminDependencies) -> APIRouter: """Create clients router.""" - app = APIRouter() + app = APIRouter(dependencies=[Depends(dependencies.get_current_active_user)]) @app.get("/clients/") async def get_clients( diff --git a/packages/sshecret-admin/src/sshecret_admin/api/endpoints/secrets.py b/packages/sshecret-admin/src/sshecret_admin/api/endpoints/secrets.py index 01ee01d..8b07b07 100644 --- a/packages/sshecret-admin/src/sshecret_admin/api/endpoints/secrets.py +++ b/packages/sshecret-admin/src/sshecret_admin/api/endpoints/secrets.py @@ -19,7 +19,7 @@ LOG = logging.getLogger(__name__) def create_router(dependencies: AdminDependencies) -> APIRouter: """Create secrets router.""" - app = APIRouter() + app = APIRouter(dependencies=[Depends(dependencies.get_current_active_user)]) @app.get("/secrets/") async def get_secret_names( diff --git a/packages/sshecret-admin/src/sshecret_admin/api/router.py b/packages/sshecret-admin/src/sshecret_admin/api/router.py index ae939fb..ed0e32f 100644 --- a/packages/sshecret-admin/src/sshecret_admin/api/router.py +++ b/packages/sshecret-admin/src/sshecret_admin/api/router.py @@ -55,7 +55,9 @@ def create_router(dependencies: BaseDependencies) -> APIRouter: raise HTTPException(status_code=400, detail="Inactive or disabled user") return current_user - async def get_admin_backend(session: Annotated[Session, Depends(dependencies.get_db_session)]): + async def get_admin_backend( + session: Annotated[Session, Depends(dependencies.get_db_session)] + ): """Get admin backend API.""" password_db = session.exec(select(PasswordDB).where(PasswordDB.id == 1)).first() if not password_db: @@ -65,11 +67,13 @@ def create_router(dependencies: BaseDependencies) -> APIRouter: admin = AdminBackend(dependencies.settings, password_db.encrypted_password) yield admin - app = APIRouter( - prefix=f"/api/{API_VERSION}", dependencies=[Depends(get_current_active_user)] + app = APIRouter(prefix=f"/api/{API_VERSION}") + + endpoint_deps = AdminDependencies.create( + dependencies, get_admin_backend, get_current_active_user ) - endpoint_deps = AdminDependencies.create(dependencies, get_admin_backend) + LOG.debug("Registering sub-routers") app.include_router(auth.create_router(endpoint_deps)) app.include_router(clients.create_router(endpoint_deps)) diff --git a/packages/sshecret-admin/src/sshecret_admin/core/dependencies.py b/packages/sshecret-admin/src/sshecret_admin/core/dependencies.py index d2d2137..fd783e6 100644 --- a/packages/sshecret-admin/src/sshecret_admin/core/dependencies.py +++ b/packages/sshecret-admin/src/sshecret_admin/core/dependencies.py @@ -2,9 +2,10 @@ from collections.abc import AsyncGenerator, Callable, Generator from dataclasses import dataclass -from typing import Self +from typing import Awaitable, Self from sqlmodel import Session +from sshecret_admin.auth import User from sshecret_admin.services import AdminBackend from sshecret_admin.core.settings import AdminServerSettings @@ -13,6 +14,8 @@ DBSessionDep = Callable[[], Generator[Session, None, None]] AdminDep = Callable[[Session], AsyncGenerator[AdminBackend, None]] +GetUserDep = Callable[[User], Awaitable[User]] + @dataclass class BaseDependencies: @@ -21,17 +24,25 @@ class BaseDependencies: settings: AdminServerSettings get_db_session: DBSessionDep + @dataclass class AdminDependencies(BaseDependencies): """Dependency class with admin.""" get_admin_backend: AdminDep + get_current_active_user: GetUserDep @classmethod - def create(cls, deps: BaseDependencies, get_admin_backend: AdminDep) -> Self: + def create( + cls, + deps: BaseDependencies, + get_admin_backend: AdminDep, + get_current_active_user: GetUserDep, + ) -> Self: """Create from base dependencies.""" return cls( settings=deps.settings, get_db_session=deps.get_db_session, get_admin_backend=get_admin_backend, + get_current_active_user=get_current_active_user, )