"""Broker configuration management (spec sections 25-27, 30, 31, 61). Release 1 flow: edit model -> generate -> validate via control sidecar -> deploy (activate + reload) -> revision recorded. Diffs and approval workflow arrive with Release 2/3 polish, but every deploy already creates an immutable revision (spec section 26). """ from __future__ import annotations import uuid from datetime import UTC, datetime from typing import Annotated import httpx from fastapi import APIRouter, Depends, HTTPException, Request, status from sqlalchemy import select from sqlalchemy.orm import Session from app import audit from app.config import get_settings from app.configgen import generate_config from app.db import get_db from app.deps import CurrentUser, require_roles from app.models import ( BrokerSettings, ConfigRevision, ListenerConfig, ManagementRole, ) from app.schemas import ( BrokerSettingsIn, BrokerSettingsOut, ConfigPreviewOut, DeployRequest, ListenerIn, ListenerOut, RevisionDetailOut, RevisionOut, ) router = APIRouter(prefix="/api/v1/broker", tags=["broker"]) BrokerAdmin = Annotated[CurrentUser, Depends(require_roles(ManagementRole.BROKER_ADMIN))] AnyReader = Annotated[ CurrentUser, Depends( require_roles( ManagementRole.BROKER_ADMIN, ManagementRole.OPERATOR, ManagementRole.AUDITOR, ) ), ] def _next_revision_number(db: Session) -> int: highest = db.scalar( select(ConfigRevision.revision_number).order_by( ConfigRevision.revision_number.desc() ) ) return (highest or 0) + 1 def _call_control( path: str, *, method: str = "POST", json_body: dict | None = None ) -> dict: settings = get_settings() headers = {"Authorization": f"Bearer {settings.control_api_token}"} try: with httpx.Client(base_url=settings.control_base_url, timeout=30) as client: response = client.request( method, path, headers=headers, json=json_body ) except httpx.TransportError as exc: raise HTTPException( status.HTTP_503_SERVICE_UNAVAILABLE, f"broker control service unreachable: {exc}", ) from exc if response.status_code >= 400: detail = response.json().get("error", response.text) raise HTTPException(response.status_code, detail) return response.json() # --- Listeners --------------------------------------------------------------- @router.get("/listeners", response_model=list[ListenerOut]) def list_listeners(db: Annotated[Session, Depends(get_db)], _user: AnyReader): return db.scalars(select(ListenerConfig)).all() @router.post("/listeners", response_model=ListenerOut, status_code=201) def create_listener( body: ListenerIn, request: Request, db: Annotated[Session, Depends(get_db)], _user: BrokerAdmin, ): listener = ListenerConfig(**body.model_dump()) db.add(listener) db.flush() audit.record( db, "LISTENER_CREATED", request=request, object_type="listener", object_id=str(listener.id), new_value={"name": body.name, "port": body.port}, ) return listener @router.put("/listeners/{listener_id}", response_model=ListenerOut) def update_listener( listener_id: uuid.UUID, body: ListenerIn, request: Request, db: Annotated[Session, Depends(get_db)], _user: BrokerAdmin, ): listener = db.get(ListenerConfig, listener_id) if listener is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "listener not found") for key, value in body.model_dump().items(): setattr(listener, key, value) db.flush() # Audit the security-relevant listener fields (spec section 38): the # full PUT payload is a replacement, so record what it now says. audit.record( db, "LISTENER_UPDATED", request=request, object_type="listener", object_id=str(listener.id), new_value={ "name": body.name, "port": body.port, "bind_address": body.bind_address, "protocol": body.protocol, "allow_anonymous": body.allow_anonymous, }, ) return listener @router.delete("/listeners/{listener_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_listener( listener_id: uuid.UUID, request: Request, db: Annotated[Session, Depends(get_db)], _user: BrokerAdmin, ): listener = db.get(ListenerConfig, listener_id) if listener is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "listener not found") db.delete(listener) audit.record( db, "LISTENER_DELETED", request=request, object_type="listener", object_id=str(listener.id), ) return None # --- Global settings ----------------------------------------------------------- @router.get("/settings", response_model=list[BrokerSettingsOut]) def list_settings(db: Annotated[Session, Depends(get_db)], _user: AnyReader): return db.scalars(select(BrokerSettings)).all() @router.put("/settings/{key}", response_model=BrokerSettingsIn) def upsert_setting( key: str, body: BrokerSettingsIn, request: Request, db: Annotated[Session, Depends(get_db)], _user: BrokerAdmin, ): if body.key != key: raise HTTPException(status.HTTP_400_BAD_REQUEST, "key mismatch") setting = db.scalar(select(BrokerSettings).where(BrokerSettings.key == key)) if setting is None: setting = BrokerSettings(key=key, value=body.value, value_type=body.value_type) db.add(setting) else: setting.value = body.value setting.value_type = body.value_type db.flush() audit.record( db, "BROKER_SETTING_CHANGED", request=request, object_type="broker_setting", object_id=key, new_value={"value": body.value}, ) return setting # --- Generation, validation, deployment ---------------------------------------- @router.get("/config/preview", response_model=ConfigPreviewOut) def preview_config( db: Annotated[Session, Depends(get_db)], _user: AnyReader ): """Show the generated mosquitto.conf without deploying (spec section 32).""" result = generate_config(db) return ConfigPreviewOut(config_text=result.config_text, warnings=result.warnings) @router.post("/config/validate") def validate_config( db: Annotated[Session, Depends(get_db)], _user: BrokerAdmin ): """Generate + validate against the real broker binary (spec section 27).""" result = generate_config(db) response = _call_control("/deploy", json_body={"content": result.config_text}) # NOTE: /deploy both validates AND activates. Release 1 keeps the # combined flow; Release 2 splits validate-only via a staged endpoint. return response @router.post("/config/deploy") def deploy_config( body: DeployRequest, request: Request, db: Annotated[Session, Depends(get_db)], user: BrokerAdmin, ): """Generate -> validate -> activate + reload -> record revision.""" result = generate_config(db) response = _call_control("/deploy", json_body={"content": result.config_text}) if response.get("status") != "deployed": raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, response) revision = ConfigRevision( revision_number=_next_revision_number(db), created_by=user.username, comment=body.comment, config_text=result.config_text, validation_result=response, deployed=True, deployed_at=datetime.now(UTC), ) db.add(revision) audit.record( db, "BROKER_CONFIG_DEPLOYED", request=request, user=user.username, object_type="config_revision", object_id=str(revision.revision_number), new_value={"comment": body.comment}, ) return { "status": "deployed", "revision": revision.revision_number, "broker_healthy": response.get("broker_healthy"), } @router.get("/revisions", response_model=list[RevisionOut]) def list_revisions( db: Annotated[Session, Depends(get_db)], _user: AnyReader ): return db.scalars( select(ConfigRevision).order_by(ConfigRevision.revision_number.desc()) ).all() @router.get("/revisions/{number}", response_model=RevisionDetailOut) def get_revision( number: int, db: Annotated[Session, Depends(get_db)], _user: AnyReader, ): revision = db.scalar( select(ConfigRevision).where(ConfigRevision.revision_number == number) ) if revision is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "revision not found") return revision