Feat: Sync Streams from Mediamtx to TAK Server

src/rasenmaeher_api/web/api/product/schema.py

@@ -107,3 +107,51 @@ class ProductAddRequest(BaseModel):  # pylint: disable=too-few-public-methods
 
     certcn: str = Field(description="CN of the certificate")
     x509cert: str = Field(description="Certificate encoded with CFSSL conventions (newlines escaped)")
+
+
+class ProductAuthzRequest(BaseModel):  # pylint: disable=too-few-public-methods
+    """Request authz for a specific source product."""
+
+    model_config = ConfigDict(
+        extra="forbid",
+        json_schema_extra={
+            "examples": [
+                {
+                    "certcn": "product.deployment.tld",
+                },
+            ],
+        },
+    )
+
+    certcn: str = Field(description="CN of the source product certificate")
+
+
+class ProductAuthzResponse(BaseModel):  # pylint: disable=too-few-public-methods
+    """Authz info for a product integration."""
+
+    model_config = ConfigDict(
+        extra="forbid",
+        json_schema_extra={
+            "examples": [
+                {
+                    "type": "mtls",
+                },
+                {
+                    "type": "bearer-token",
+                    "token": "<JWT>",
+                },
+                {
+                    "type": "basic",
+                    "username": "product.deployment.tld",
+                    "password": "<PASSWORD>",
+                    "ro_password": "<PASSWORD>",
+                },
+            ],
+        },
+    )
+
+    type: str = Field(description="type of authz: bearer-token, basic, mtls")
+    token: str | None = Field(description="Bearer token", default=None)
+    username: str | None = Field(description="Username for basic auth", default=None)
+    password: str | None = Field(description="Password for basic auth", default=None)
+    ro_password: str | None = Field(description="Password for read-only streaming", default=None)

src/rasenmaeher_api/web/api/product/views.py

@@ -14,7 +14,15 @@
 from OpenSSL.crypto import load_certificate_request, FILETYPE_PEM  # FIXME: use cryptography instead of pyOpenSSL
 
 
-from .schema import CertificatesResponse, CertificatesRequest, RevokeRequest, KCClientToken, ProductAddRequest
+from .schema import (
+    CertificatesResponse,
+    CertificatesRequest,
+    RevokeRequest,
+    KCClientToken,
+    ProductAddRequest,
+    ProductAuthzRequest,
+    ProductAuthzResponse,
+)
 from ....db.nonces import SeenToken
 from ....db.errors import NotFound
 from ....db import Person
@@ -170,6 +178,35 @@ async def add_interop(
     return resp
 
 
+@router.get("/interop/{tgtproduct}/authz", dependencies=[Depends(MTLSHeader(auto_error=True))])
+async def get_interop_authz(
+    tgtproduct: str,
+    request: Request,
+) -> ProductAuthzResponse:
+    """Broker authz for another product integration on behalf of the caller."""
+    payload = request.state.mtlsdn
+    srcproduct = payload.get("CN")
+    if srcproduct not in RMSettings.singleton().valid_product_cns:
+        raise HTTPException(status_code=403)
+
+    manifest = RMSettings.singleton().kraftwerk_manifest_dict
+    if "products" not in manifest:
+        LOGGER.error("Manifest does not have products key")
+        raise HTTPException(status_code=500, detail="Manifest does not have products key")
+    if tgtproduct not in manifest["products"]:
+        raise HTTPException(status_code=404, detail=f"Unknown product {tgtproduct}")
+
+    resp = await post_to_product(
+        tgtproduct,
+        "/api/v1/interop/authz",
+        ProductAuthzRequest(certcn=str(srcproduct)).model_dump(),
+        ProductAuthzResponse,
+    )
+    if resp is None:
+        raise HTTPException(status_code=502, detail="Target integration did not return authz")
+    return cast(ProductAuthzResponse, resp)
+
+
 @router.get("/proxy/{tgtproduct}/{tgtpath:path}", dependencies=[Depends(ValidUser(auto_error=True))])
 async def get_product_proxy(
     tgtproduct: str,