DRAFT: Make it possible for product integration APIs to enable admin privileges for themselves to another product

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 1.6.0
+current_version = 1.7.0
 commit = False
 tag = False
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "takrmapi"
-version = "1.6.0"
+version = "1.7.0"
 description = "RASENMAEHER integration API for TAK server"
 authors = ["Eero af Heurlin <rambo@iki.fi>", "Ari Karhunen <FIXME@example.com>"]
 homepage = "https://github.com/pvarki/python-tak-rmapi"

src/takrmapi/__init__.py

@@ -1,3 +1,3 @@
 """ RASENMAEHER integration API for TAK server """
 
-__version__ = "1.6.0"  # NOTE Use `bump2version --config-file patch` to bump versions correctly
+__version__ = "1.7.0"  # NOTE Use `bump2version --config-file patch` to bump versions correctly

src/takrmapi/api/__init__.py

@@ -3,7 +3,6 @@
 from fastapi.routing import APIRouter
 
 from .usercrud import router as usercrud_router
-from .usercrud import test_router as testing_router  # REMOVE ME
 from .clientinfo import router as clientinfo_router
 from .admininfo import router as admininfo_router
 from .healthcheck import router as healthcheck_router
@@ -12,7 +11,6 @@
 from .tak_datapackage import router as takdatapackage_router
 
 all_routers = APIRouter()
-all_routers.include_router(testing_router, prefix="/users", tags=["users"])  # REMOVE ME
 all_routers.include_router(usercrud_router, prefix="/users", tags=["users"])
 all_routers.include_router(clientinfo_router, prefix="/clients", tags=["clients"])
 all_routers.include_router(admininfo_router, prefix="/admins", tags=["admins"])

src/takrmapi/api/helpers.py

@@ -0,0 +1,12 @@
+"""General helpers"""
+
+from fastapi import Request, HTTPException
+
+from .. import config
+
+
+def comes_from_rm(request: Request) -> None:
+    """Check the CN, raises 403 if not"""
+    payload = request.state.mtlsdn
+    if payload.get("CN") != config.RMCN:
+        raise HTTPException(status_code=403)

src/takrmapi/api/interop.py

@@ -0,0 +1,148 @@
+"""Routes for interoperation between products"""
+
+from typing import Sequence, Optional
+import logging
+from pathlib import Path
+
+from fastapi import APIRouter, Depends, Request, HTTPException
+from libpvarki.middleware import MTLSHeader
+from libpvarki.schemas.generic import OperationResultResponse
+from libpvarki.schemas.product import UserCRUDRequest
+from pydantic import Field, BaseModel, ConfigDict, Extra  # pylint: disable=E0611  # False positive
+
+
+from .helpers import comes_from_rm
+from ..tak_helpers import UserCRUD as TAKBaseCRUD
+from ..tak_helpers import Helpers as TAKBaseHelper
+from .. import config
+
+
+LOGGER = logging.getLogger(__name__)
+
+interoprouter = APIRouter(dependencies=[Depends(MTLSHeader(auto_error=True))])
+
+
+# FIXME: Move to libpvarki
+class ProductAddRequest(BaseModel):  # pylint: disable=too-few-public-methods
+    """Request to add product interoperability."""
+
+    certcn: str = Field(description="CN of the certificate")
+    x509cert: str = Field(description="Certificate encoded with CFSSL conventions (newlines escaped)")
+
+    model_config = ConfigDict(
+        extra=Extra.forbid,
+        schema_extra={
+            "examples": [
+                {
+                    "certcn": "product.deployment.tld",
+                    "x509cert": "-----BEGIN CERTIFICATE-----\\nMIIEwjCC...\\n-----END CERTIFICATE-----\\n",
+                },
+            ],
+        },
+    )
+
+
+# FIXME: Move to libpvarki
+class ProductAuthzResponse(BaseModel):  # pylint: disable=too-few-public-methods
+    """Authz info"""
+
+    type: str = Field(description="type of authz: bearer-token, basic, mtls")
+    token: Optional[str] = Field(description="Bearer token", default=None)
+    username: Optional[str] = Field(description="Username for basic auth", default=None)
+    password: Optional[str] = Field(description="Password for basic auth", default=None)
+
+    model_config = ConfigDict(
+        extra=Extra.forbid,
+        schema_extra={
+            "examples": [
+                {
+                    "type": "mtls",
+                },
+                {
+                    "type": "bearer-token",
+                    "token": "<JWT>",
+                },
+                {
+                    "type": "basic",
+                    "username": "product.deployment.tld",
+                    "password": "<PASSWORD>",
+                },
+            ],
+        },
+    )
+
+
+def cn_to_callsign(certcn: str) -> str:
+    """Convert CN to something that is valid as callsign filename"""
+    return certcn.replace(".", "_")
+
+
+class ProductCRUD(TAKBaseCRUD):
+    """Pretend to be user"""
+
+    def __init__(self, user: UserCRUDRequest):
+        super().__init__(user)
+        self.helpers = TakProductHelper(self)
+        self.write_cert_conditional()
+
+    @property
+    def certpath(self) -> Path:
+        """Path to local cert"""
+        return config.TAK_CERTS_FOLDER / f"{self.certcn}_rm.pem"
+
+    def write_cert_conditional(self) -> None:
+        """write the X509 cert if it did not exist"""
+        if self.certpath.exists():
+            return
+        if self.user.x509cert == "NOTHERE":
+            raise HTTPException(status_code=404, detail="Certificate not set")
+        self.certpath.parent.mkdir(parents=True, exist_ok=True)
+        self.certpath.write_text(self.rm_certpem)
+
+
+class TakProductHelper(TAKBaseHelper):  # pylint: disable=too-few-public-methods
+    """Do TAK things"""
+
+    @property
+    def enable_user_cert_names(self) -> Sequence[str]:
+        """Return the stems for cert PEM files"""
+        return (f"{self.user.callsign}_rm",)
+
+
+@interoprouter.post("/add")
+async def add_product(
+    product: ProductAddRequest,
+    request: Request,
+) -> OperationResultResponse:
+    """Product needs interop privileges. This can only be called by RASENMAEHER"""
+    comes_from_rm(request)
+    callsign = cn_to_callsign(product.certcn)
+    product_user = UserCRUDRequest(
+        uuid=callsign,
+        callsign=callsign,
+        x509cert=product.x509cert,
+    )
+    crud = ProductCRUD(product_user)
+    res = await crud.helpers.add_admin_to_tak_with_cert()
+    if not res:
+        return OperationResultResponse(success=False, error="TAK admin script failed")
+    return OperationResultResponse(success=True)
+
+
+@interoprouter.get("/authz")
+async def get_authz(
+    request: Request,
+) -> ProductAuthzResponse:
+    """Get authz info for the product, for tak it's always mtls"""
+    payload = request.state.mtlsdn
+    callsign = cn_to_callsign(payload.get("CN"))
+    product_user = UserCRUDRequest(
+        uuid=callsign,
+        callsign=callsign,
+        x509cert="NOTHERE",
+    )
+    crud = ProductCRUD(product_user)
+    if not crud.certpath.exists():
+        raise HTTPException(status_code=404, detail="Certificate does not exist")
+    result = ProductAuthzResponse(type="mtls")
+    return result

src/takrmapi/api/usercrud.py

@@ -1,23 +1,23 @@
 """"User actions"""
 
 import logging
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, Request
 from libpvarki.middleware import MTLSHeader
 from libpvarki.schemas.product import UserCRUDRequest
 from libpvarki.schemas.generic import OperationResultResponse
 
-from takrmapi import tak_helpers
+from .. import tak_helpers
+from .helpers import comes_from_rm
 
 LOGGER = logging.getLogger(__name__)
 
 router = APIRouter(dependencies=[Depends(MTLSHeader(auto_error=True))])
 
-test_router = APIRouter()
-
 
 @router.post("/created")
-async def user_created(user: UserCRUDRequest) -> OperationResultResponse:
+async def user_created(user: UserCRUDRequest, request: Request) -> OperationResultResponse:
     """New device cert was created"""
+    comes_from_rm(request)
     tak_usercrud = tak_helpers.UserCRUD(user)
     LOGGER.info("Adding new user '{}' to TAK".format(user.callsign))
     await tak_usercrud.add_new_user()
@@ -29,8 +29,9 @@ async def user_created(user: UserCRUDRequest) -> OperationResultResponse:
 # While delete would be semantically better it takes no body and definitely forces the
 # integration layer to keep track of UUIDs
 @router.post("/revoked")
-async def user_revoked(user: UserCRUDRequest) -> OperationResultResponse:
+async def user_revoked(user: UserCRUDRequest, request: Request) -> OperationResultResponse:
     """Device cert was revoked"""
+    comes_from_rm(request)
     tak_usercrud = tak_helpers.UserCRUD(user)
     LOGGER.info("Removing user '{}' from TAK".format(user.callsign))
     await tak_usercrud.revoke_user()
@@ -39,8 +40,9 @@ async def user_revoked(user: UserCRUDRequest) -> OperationResultResponse:
 
 
 @router.post("/promoted")
-async def user_promoted(user: UserCRUDRequest) -> OperationResultResponse:
+async def user_promoted(user: UserCRUDRequest, request: Request) -> OperationResultResponse:
     """Device cert was promoted to admin privileges"""
+    comes_from_rm(request)
     tak_usercrud = tak_helpers.UserCRUD(user)
     LOGGER.info("Promoting user '{}' to admin".format(user.callsign))
     await tak_usercrud.promote_user()
@@ -49,8 +51,9 @@ async def user_promoted(user: UserCRUDRequest) -> OperationResultResponse:
 
 
 @router.post("/demoted")
-async def user_demoted(user: UserCRUDRequest) -> OperationResultResponse:
+async def user_demoted(user: UserCRUDRequest, request: Request) -> OperationResultResponse:
     """Device cert was demoted to standard privileges"""
+    comes_from_rm(request)
     tak_usercrud = tak_helpers.UserCRUD(user)
     LOGGER.info("Demoting user '{}' to normal user".format(user.callsign))
     await tak_usercrud.demote_user()
@@ -59,8 +62,9 @@ async def user_demoted(user: UserCRUDRequest) -> OperationResultResponse:
 
 
 @router.put("/updated")
-async def user_updated(user: UserCRUDRequest) -> OperationResultResponse:
+async def user_updated(user: UserCRUDRequest, request: Request) -> OperationResultResponse:
     """Device callsign updated"""
+    comes_from_rm(request)
     tak_usercrud = tak_helpers.UserCRUD(user)
     await tak_usercrud.update_user()
     result = OperationResultResponse(success=True)

src/takrmapi/config.py

@@ -75,3 +75,4 @@ def read_deployment_name() -> str:
     "TAK_SERVER_NETWORKMESH_KEY_FILE", cast=Path, default=Path("/opt/tak/data/tak_server_networkmesh")
 )
 TAK_SERVER_NETWORKMESH_KEY_STR: str = ""
+RMCN: str = cfg("RMCN", cast=str, default="rasenmaeher")  # expected CN for RASENMAEHERs mTLS cert

tests/conftest.py

@@ -24,7 +24,7 @@ def mtlsclient() -> Generator[TestClient, None, None]:
     client = TestClient(
         APP,
         headers={
-            "X-ClientCert-DN": "CN=harjoitus1.pvarki.fi,O=harjoitus1.pvarki.fi,L=KeskiSuomi,ST=Jyvaskyla,C=FI",
+            "X-ClientCert-DN": "CN=rasenmaeher,O=harjoitus1.pvarki.fi,L=KeskiSuomi,ST=Jyvaskyla,C=FI",
         },
     )
     yield client

tests/test_takrmapi.py

@@ -5,4 +5,4 @@
 
 def test_version() -> None:
     """Make sure version matches expected"""
-    assert __version__ == "1.6.0"
+    assert __version__ == "1.7.0"