258 lines
9.4 KiB
Python
258 lines
9.4 KiB
Python
"""Secrets sub-api factory."""
|
|
|
|
# pyright: reportUnusedFunction=false
|
|
|
|
import logging
|
|
from collections import defaultdict
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from sqlalchemy import select
|
|
from typing import Annotated
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from sshecret_backend.models import Client, ClientSecret
|
|
from sshecret_backend.view_models import (
|
|
ClientReference,
|
|
ClientSecretDetailList,
|
|
ClientSecretList,
|
|
ClientSecretPublic,
|
|
BodyValue,
|
|
ClientSecretResponse,
|
|
)
|
|
from sshecret_backend import audit
|
|
from sshecret_backend.types import AsyncDBSessionDep
|
|
from .common import get_client_by_id_or_name, get_client_by_name
|
|
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
async def lookup_client_secret(
|
|
session: AsyncSession, client: Client, name: str
|
|
) -> ClientSecret | None:
|
|
"""Look up a secret for a client."""
|
|
statement = (
|
|
select(ClientSecret)
|
|
.where(ClientSecret.client_id == client.id)
|
|
.where(ClientSecret.name == name)
|
|
)
|
|
results = await session.scalars(statement)
|
|
return results.first()
|
|
|
|
|
|
def get_secrets_api(get_db_session: AsyncDBSessionDep) -> APIRouter:
|
|
"""Construct clients sub-api."""
|
|
router = APIRouter()
|
|
|
|
@router.post("/clients/{name}/secrets/")
|
|
async def add_secret_to_client(
|
|
request: Request,
|
|
name: str,
|
|
client_secret: ClientSecretPublic,
|
|
session: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> None:
|
|
"""Add secret to a client."""
|
|
client = await get_client_by_id_or_name(session, name)
|
|
if not client:
|
|
raise HTTPException(
|
|
status_code=404, detail="Cannot find a client with the given name."
|
|
)
|
|
|
|
existing_secret = await lookup_client_secret(
|
|
session, client, client_secret.name
|
|
)
|
|
if existing_secret:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Cannot add a secret. A different secret with the same name already exists.",
|
|
)
|
|
db_secret = ClientSecret(
|
|
name=client_secret.name, client_id=client.id, secret=client_secret.secret
|
|
)
|
|
session.add(db_secret)
|
|
await session.commit()
|
|
await session.refresh(db_secret)
|
|
await audit.audit_create_secret(session, request, client, db_secret)
|
|
|
|
@router.put("/clients/{name}/secrets/{secret_name}")
|
|
async def update_client_secret(
|
|
request: Request,
|
|
name: str,
|
|
secret_name: str,
|
|
secret_data: BodyValue,
|
|
session: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> ClientSecretResponse:
|
|
"""Update a client secret.
|
|
|
|
This can also be used for destructive creates.
|
|
"""
|
|
client = await get_client_by_id_or_name(session, name)
|
|
if not client:
|
|
raise HTTPException(
|
|
status_code=404, detail="Cannot find a client with the given name."
|
|
)
|
|
|
|
existing_secret = await lookup_client_secret(session, client, secret_name)
|
|
if existing_secret:
|
|
existing_secret.secret = secret_data.value
|
|
|
|
session.add(existing_secret)
|
|
await session.commit()
|
|
await session.refresh(existing_secret)
|
|
await audit.audit_update_secret(session, request, client, existing_secret)
|
|
return ClientSecretResponse.from_client_secret(existing_secret)
|
|
|
|
db_secret = ClientSecret(
|
|
name=secret_name,
|
|
client_id=client.id,
|
|
secret=secret_data.value,
|
|
)
|
|
session.add(db_secret)
|
|
await session.commit()
|
|
await session.refresh(db_secret)
|
|
await audit.audit_create_secret(session, request, client, db_secret)
|
|
return ClientSecretResponse.from_client_secret(db_secret)
|
|
|
|
@router.get("/clients/{name}/secrets/{secret_name}")
|
|
async def request_client_secret(
|
|
request: Request,
|
|
name: str,
|
|
secret_name: str,
|
|
session: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> ClientSecretResponse:
|
|
"""Get a client secret."""
|
|
client = await get_client_by_id_or_name(session, name)
|
|
if not client:
|
|
raise HTTPException(
|
|
status_code=404, detail="Cannot find a client with the given name."
|
|
)
|
|
|
|
secret = await lookup_client_secret(session, client, secret_name)
|
|
if not secret:
|
|
raise HTTPException(
|
|
status_code=404, detail="Cannot find a secret with the given name."
|
|
)
|
|
|
|
response_model = ClientSecretResponse.from_client_secret(secret)
|
|
await audit.audit_access_secret(session, request, client, secret)
|
|
return response_model
|
|
|
|
@router.delete("/clients/{name}/secrets/{secret_name}")
|
|
async def delete_client_secret(
|
|
request: Request,
|
|
name: str,
|
|
secret_name: str,
|
|
session: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> None:
|
|
"""Delete a secret."""
|
|
client = await get_client_by_name(session, name)
|
|
if not client:
|
|
raise HTTPException(
|
|
status_code=404, detail="Cannot find a client with the given name."
|
|
)
|
|
|
|
secret = await lookup_client_secret(session, client, secret_name)
|
|
if not secret:
|
|
raise HTTPException(
|
|
status_code=404, detail="Cannot find a secret with the given name."
|
|
)
|
|
|
|
await session.delete(secret)
|
|
await session.commit()
|
|
await audit.audit_delete_secret(session, request, client, secret)
|
|
|
|
@router.get("/secrets/")
|
|
async def get_secret_map(
|
|
session: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> list[ClientSecretList]:
|
|
"""Get a list of all secrets and which clients have them."""
|
|
client_secret_map: defaultdict[str, list[str]] = defaultdict(list)
|
|
client_secrets = await session.scalars(
|
|
select(ClientSecret).options(selectinload(ClientSecret.client))
|
|
)
|
|
for client_secret in client_secrets.all():
|
|
if not client_secret.client:
|
|
if client_secret.name not in client_secret_map:
|
|
client_secret_map[client_secret.name] = []
|
|
continue
|
|
client_secret_map[client_secret.name].append(client_secret.client.name)
|
|
# audit.audit_client_secret_list(session, request)
|
|
return [
|
|
ClientSecretList(name=secret_name, clients=clients)
|
|
for secret_name, clients in client_secret_map.items()
|
|
]
|
|
|
|
@router.get("/secrets/detailed/")
|
|
async def get_detailed_secret_map(
|
|
session: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> list[ClientSecretDetailList]:
|
|
"""Get a list of all secrets and which clients have them."""
|
|
client_secrets: dict[str, ClientSecretDetailList] = {}
|
|
all_client_secrets = await session.execute(
|
|
select(ClientSecret).options(selectinload(ClientSecret.client))
|
|
)
|
|
for client_secret in all_client_secrets.scalars().all():
|
|
if client_secret.name not in client_secrets:
|
|
client_secrets[client_secret.name] = ClientSecretDetailList(
|
|
name=client_secret.name
|
|
)
|
|
client_secrets[client_secret.name].ids.append(str(client_secret.id))
|
|
if not client_secret.client:
|
|
continue
|
|
client_secrets[client_secret.name].clients.append(
|
|
ClientReference(
|
|
id=str(client_secret.client.id), name=client_secret.client.name
|
|
)
|
|
)
|
|
# `audit.audit_client_secret_list(session, request)
|
|
return list(client_secrets.values())
|
|
|
|
@router.get("/secrets/{name}")
|
|
async def get_secret_clients(
|
|
name: str,
|
|
session: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> ClientSecretList:
|
|
"""Get a list of which clients has a named secret."""
|
|
clients: list[str] = []
|
|
client_secrets = await session.scalars(
|
|
select(ClientSecret)
|
|
.join(ClientSecret.client)
|
|
.options(selectinload(ClientSecret.client))
|
|
.where(ClientSecret.name == name)
|
|
.where(Client.is_active.is_(True))
|
|
)
|
|
for client_secret in client_secrets.all():
|
|
if not client_secret.client:
|
|
continue
|
|
clients.append(client_secret.client.name)
|
|
|
|
return ClientSecretList(name=name, clients=clients)
|
|
|
|
@router.get("/secrets/{name}/detailed")
|
|
async def get_secret_clients_detailed(
|
|
name: str,
|
|
session: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> ClientSecretDetailList:
|
|
"""Get a list of which clients has a named secret."""
|
|
detail_list = ClientSecretDetailList(name=name)
|
|
client_secrets = await session.scalars(
|
|
select(ClientSecret)
|
|
.options(selectinload(ClientSecret.client))
|
|
.where(ClientSecret.name == name)
|
|
.where(ClientSecret.client.is_(Client.is_active))
|
|
)
|
|
for client_secret in client_secrets.all():
|
|
if not client_secret.client:
|
|
continue
|
|
detail_list.ids.append(str(client_secret.id))
|
|
detail_list.clients.append(
|
|
ClientReference(
|
|
id=str(client_secret.client.id), name=client_secret.client.name
|
|
)
|
|
)
|
|
|
|
return detail_list
|
|
|
|
return router
|