108 lines
4.0 KiB
Python
108 lines
4.0 KiB
Python
"""Client Secret Router."""
|
|
|
|
# pyright: reportUnusedFunction=false
|
|
|
|
import logging
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from sshecret_backend.api.common import FlexID
|
|
from sshecret_backend.types import AsyncDBSessionDep
|
|
from sshecret_backend.api.secrets.operations import (
|
|
ClientSecretOperations,
|
|
resolve_client_secret_clients,
|
|
resolve_client_secret_mapping,
|
|
)
|
|
from sshecret_backend.api.schemas import BodyValue
|
|
from sshecret_backend.api.secrets.schemas import (
|
|
ClientSecretDetailList,
|
|
ClientSecretPublic,
|
|
ClientSecretResponse,
|
|
)
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
def create_client_secrets_router(get_db_session: AsyncDBSessionDep) -> APIRouter:
|
|
"""Create client secret router."""
|
|
router = APIRouter()
|
|
|
|
@router.post("/clients/{client_identifier}/secrets/")
|
|
async def add_secret_to_client(
|
|
request: Request,
|
|
client_identifier: str,
|
|
client_secret: ClientSecretPublic,
|
|
session: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> None:
|
|
"""Add secret to a client."""
|
|
client = FlexID.from_string(client_identifier)
|
|
client_op = ClientSecretOperations(session, request, client)
|
|
await client_op.create_client_secret(
|
|
client_secret.name, client_secret.secret, client_secret.description
|
|
)
|
|
|
|
@router.put("/clients/{client_identifier}/secrets/{secret_identifier}")
|
|
async def update_client_secret(
|
|
request: Request,
|
|
client_identifier: str,
|
|
secret_identifier: str,
|
|
secret_data: BodyValue,
|
|
session: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> ClientSecretResponse:
|
|
"""Update client secret."""
|
|
client = FlexID.from_string(client_identifier)
|
|
secret = FlexID.from_string(secret_identifier)
|
|
client_op = ClientSecretOperations(session, request, client)
|
|
return await client_op.update_client_secret_value(secret, secret_data.value)
|
|
|
|
@router.get("/clients/{client_identifier}/secrets/{secret_identifier}")
|
|
async def request_client_secret_named(
|
|
request: Request,
|
|
client_identifier: str,
|
|
secret_identifier: str,
|
|
session: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> ClientSecretResponse:
|
|
"""Get a named secret from a named client."""
|
|
client = FlexID.from_string(client_identifier)
|
|
secret = FlexID.from_string(secret_identifier)
|
|
LOG.debug("Resolved client FlexID: %r", client)
|
|
LOG.debug("Resolved secret FlexID: %r", secret)
|
|
client_op = ClientSecretOperations(session, request, client)
|
|
return await client_op.get_client_secret(secret)
|
|
|
|
# TODO: delete_client_secret
|
|
@router.delete("/clients/{client_identifier}/secrets/{secret_identifier}")
|
|
async def delete_client_secret(
|
|
request: Request,
|
|
client_identifier: str,
|
|
secret_identifier: str,
|
|
session: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> None:
|
|
"""Delete client secret."""
|
|
client = FlexID.from_string(client_identifier)
|
|
secret = FlexID.from_string(secret_identifier)
|
|
client_op = ClientSecretOperations(session, request, client)
|
|
await client_op.delete_client_secret(secret)
|
|
|
|
@router.get("/secrets/")
|
|
async def get_secret_map(
|
|
session: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> list[ClientSecretDetailList]:
|
|
"""Get a list of secrets and which clients have them."""
|
|
return await resolve_client_secret_mapping(session)
|
|
|
|
@router.get("/secrets/{name}")
|
|
async def get_secret_clients(
|
|
name: str,
|
|
session: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> ClientSecretDetailList:
|
|
"""Get a list of which clients has a named secret."""
|
|
result = await resolve_client_secret_clients(session, name)
|
|
if not result:
|
|
raise HTTPException(status_code=404, detail="Secret not found.")
|
|
return result
|
|
|
|
return router
|