feat: add optional CSR field to enrollment schemas

Author: rambo

src/rasenmaeher_api/db/enrollments.py

@@ -168,6 +168,7 @@ class Enrollment(ORMBaseModel, table=True):  # type: ignore[call-arg,misc]
     )
     state: int = Field(nullable=False, index=False, unique=False, default=EnrollmentState.PENDING)
     extra: Dict[str, Any] = Field(sa_type=JSONB, nullable=False, sa_column_kwargs={"server_default": "{}"})
+    csr: Optional[str] = Field(default=None, nullable=True)
 
     @classmethod
     async def by_pk_or_callsign(cls, inval: Union[str, uuid.UUID]) -> "Enrollment":
@@ -273,11 +274,16 @@ async def _generate_unused_code(cls) -> str:
 
     @classmethod
     async def create_for_callsign(
-        cls, callsign: str, pool: Optional[EnrollmentPool] = None, extra: Optional[Dict[str, Any]] = None
+        cls,
+        callsign: str,
+        pool: Optional[EnrollmentPool] = None,
+        extra: Optional[Dict[str, Any]] = None,
+        csr: Optional[str] = None,
     ) -> "Enrollment":
         """Create a new one with random code for the callsign"""
         if callsign in RMSettings.singleton().valid_product_cns:
             raise CallsignReserved("Using product CNs as callsigns is forbidden")
+        # FIXME: Verify the CSR has the callsign as CN
         with EngineWrapper.get_session() as session:
             try:
                 await Enrollment.by_callsign(callsign)
@@ -294,6 +300,7 @@ async def create_for_callsign(
                 state=EnrollmentState.PENDING,
                 extra=extra,
                 pool=poolpk,
+                csr=csr,
             )
             session.add(obj)
             session.commit()

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

@@ -1,6 +1,6 @@
 """Schema for enrollment."""
 
-from typing import List, Dict, Any
+from typing import List, Dict, Any, Optional
 
 from pydantic import BaseModel, Extra, Field
 
@@ -105,6 +105,7 @@ class EnrollmentInitIn(BaseModel):  # pylint: disable=too-few-public-methods
     """Enrollment init in response schema"""
 
     callsign: str = Field(description="Callsign to create enrollment for")
+    csr: Optional[str] = Field(description="CSR for mTLS key in PEM format", default=None)
 
     class Config:  # pylint: disable=too-few-public-methods
         """Example values for schema"""

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

@@ -188,7 +188,9 @@ async def request_enrollment_init(
 
     # TODO ADD POOL NAME CHECK
 
-    new_enrollment = await Enrollment.create_for_callsign(callsign=request_in.callsign, pool=None, extra={})
+    new_enrollment = await Enrollment.create_for_callsign(
+        callsign=request_in.callsign, pool=None, extra={}, csr=request_in.csr
+    )
     # Create JWT token for user
     claims = {"sub": request_in.callsign}
     new_jwt = Issuer.singleton().issue(claims)

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

@@ -1,6 +1,8 @@
 """Schema for enrollment."""
 
-from pydantic import BaseModel, Extra
+from typing import Optional
+
+from pydantic import BaseModel, Extra, Field
 
 
 class FirstuserCheckCodeIn(BaseModel):  # pylint: disable=too-few-public-methods
@@ -51,6 +53,7 @@ class FirstuserAddAdminIn(BaseModel):  # pylint: disable=too-few-public-methods
 
     # temp_admin_code: str
     callsign: str
+    csr: Optional[str] = Field(default=None, description="CSR for mTLS key in PEM format")
 
     class Config:  # pylint: disable=too-few-public-methods
         """Example values for schema"""

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

@@ -78,7 +78,9 @@ async def post_admin_add(
         await anon_user.assign_role(role="anon_admin")
 
     # Create new admin user enrollment
-    enrollment = await Enrollment.create_for_callsign(callsign=request_in.callsign, pool=None, extra={})
+    enrollment = await Enrollment.create_for_callsign(
+        callsign=request_in.callsign, pool=None, extra={}, csr=request_in.csr
+    )
 
     # Get the anon_admin 'user' that will be used to approve the new admin user
     # and approve the user