feat: start adding routes that users can use directly with their mTLS cert (also useful for testing)

Author: rambo

src/rmmtxauthz/schema/userdirect.py

@@ -0,0 +1,22 @@
+"""Schemas for direct user mTLS routes"""
+
+from pydantic import Field, BaseModel, ConfigDict
+
+
+class UserCredentials(BaseModel):
+    """Request to add product interoperability."""
+
+    username: str = Field(description="MediaMTX username")
+    password: str = Field(description="MediaMTX password")
+
+    model_config = ConfigDict(
+        extra="forbid",
+        json_schema_extra={
+            "examples": [
+                {
+                    "username": "KOIRA11a",
+                    "password": "SomethingRandom",  # pragma: allowlist secret
+                },
+            ],
+        },
+    )

src/rmmtxauthz/web/application.py

@@ -17,6 +17,7 @@
 from .instructions import router as irouter
 from .interop import interoprouter
 from .health import hrouter
+from .userdirect import userrouter
 
 LOGGER = logging.getLogger(__name__)
 
@@ -46,6 +47,7 @@ def get_app_no_init() -> FastAPI:
     app.include_router(interoprouter, prefix="/api/v1/interop", tags=["interop"])
     app.include_router(crudrouter, prefix="/api/v1/users", tags=["users"])
     app.include_router(mtxrouter, prefix="/api/v1/mediamtx", tags=["mediamtx"])
+    app.include_router(userrouter, prefix="/api/v1/direct", tags=["directuser"])
     app.include_router(irouter, prefix="/api/v1", tags=["instructions"])
     app.include_router(hrouter, prefix="/api/v1", tags=["health"])
     return app

src/rmmtxauthz/web/userdirect.py

@@ -0,0 +1,26 @@
+"""APIs usable directly by the user with mTLS"""
+
+import logging
+
+from fastapi import APIRouter, Depends, Request
+from libpvarki.middleware import MTLSHeader
+
+from ..db.user import User
+from ..schema.userdirect import UserCredentials
+
+LOGGER = logging.getLogger(__name__)
+
+userrouter = APIRouter(dependencies=[Depends(MTLSHeader(auto_error=True))])
+
+
+def get_callsign(request: Request) -> str:
+    """extract callsign from metadata"""
+    payload = request.state.mtlsdn
+    return str(payload.get("CN"))
+
+
+@userrouter.get("/credentials", response_model=UserCredentials)
+async def get_credentials(request: Request) -> UserCredentials:
+    """Get my MediaMTX credentials"""
+    user = await User.by_username(get_callsign(request))
+    return UserCredentials(username=user.username, password=user.mtxpassword)