Add randkey for krb api 1225

app/api/main/adapters/kerberos.py

@@ -12,7 +12,12 @@
 from starlette.background import BackgroundTask
 
 from api.base_adapter import BaseAdapter
-from api.main.schema import KerberosSetupRequest
+from api.main.schema import (
+    KerberosSetupRequest,
+    KtaddRequest,
+    PrincipalAddRequest,
+    PrincipalPutRequest,
+)
 from ldap_protocol.dialogue import LDAPSession, UserSchema
 from ldap_protocol.kerberos import KerberosState
 from ldap_protocol.kerberos.service import KerberosService
@@ -66,31 +71,29 @@ async def setup_kdc(
         )
         return Response(background=task)
 
-    async def add_principal(
-        self,
-        primary: str,
-        instance: str,
-    ) -> None:
+    async def add_principal(self, request: PrincipalAddRequest) -> None:
         """Create principal in Kerberos with given name.
 
         :raises HTTPException: on Kerberos errors
         :return: None
         """
-        return await self._service.add_principal(primary, instance)
+        return await self._service.add_principal(
+            request.principal_name,
+            password=request.password,
+            algorithms=request.algorithms,
+        )
 
-    async def rename_principal(
-        self,
-        principal_name: str,
-        principal_new_name: str,
-    ) -> None:
-        """Rename principal in Kerberos.
+    async def rename_principal(self, request: PrincipalPutRequest) -> None:
+        """Modify principal (rename, password, algorithms).
 
         :raises HTTPException: on Kerberos errors
         :return: None
         """
         return await self._service.rename_principal(
-            principal_name,
-            principal_new_name,
+            principal_name=request.principal_name,
+            principal_new_name=request.new_principal_name,
+            algorithms=request.algorithms,
+            password=request.password,
         )
 
     async def reset_principal_pw(
@@ -121,14 +124,17 @@ async def delete_principal(
 
     async def ktadd(
         self,
-        names: list[str],
+        data: KtaddRequest,
     ) -> StreamingResponse:
         """Generate keytab and return as streaming response.
 
         :raises HTTPException: on Kerberos errors
         :return: StreamingResponse
         """
-        aiter_bytes, task_struct = await self._service.ktadd(names)
+        aiter_bytes, task_struct = await self._service.ktadd(
+            data.names,
+            is_rand_key=data.is_rand_key,
+        )
         task = BackgroundTask(
             task_struct.func,
             *task_struct.args,

app/api/main/krb5_router.py

@@ -23,7 +23,12 @@
     DomainErrorTranslator,
 )
 from api.main.adapters.kerberos import KerberosFastAPIAdapter
-from api.main.schema import KerberosSetupRequest
+from api.main.schema import (
+    KerberosSetupRequest,
+    KtaddRequest,
+    PrincipalAddRequest,
+    PrincipalPutRequest,
+)
 from api.utils import require_master_db
 from enums import DomainCodes
 from ldap_protocol.dialogue import LDAPSession
@@ -149,15 +154,15 @@ async def setup_kdc(
     error_map=error_map,
 )
 async def ktadd(
-    names: Annotated[LIMITED_LIST, Body()],
     kerberos_adapter: FromDishka[KerberosFastAPIAdapter],
+    request: KtaddRequest,
 ) -> StreamingResponse:
     """Create keytab from kadmin server.
 
     :param Annotated[LDAPSession, Depends ldap_session: ldap
     :return bytes: file
     """
-    return await kerberos_adapter.ktadd(names)
+    return await kerberos_adapter.ktadd(request)
 
 
 @krb5_router.get(
@@ -178,13 +183,12 @@ async def get_krb_status(
 
 
 @krb5_router.post(
-    "/principal/add",
+    "/principal",
     dependencies=[Depends(verify_auth), Depends(require_master_db)],
     error_map=error_map,
 )
 async def add_principal(
-    primary: Annotated[LIMITED_STR, Body()],
-    instance: Annotated[LIMITED_STR, Body()],
+    request: PrincipalAddRequest,
     kerberos_adapter: FromDishka[KerberosFastAPIAdapter],
 ) -> None:
     """Create principal in kerberos with given name.
@@ -194,31 +198,19 @@ async def add_principal(
     :param Annotated[LDAPSession, Depends ldap_session: ldap
     :raises HTTPException: on failed kamin request.
     """
-    await kerberos_adapter.add_principal(primary, instance)
+    await kerberos_adapter.add_principal(request)
 
 
-@krb5_router.patch(
-    "/principal/rename",
+@krb5_router.put(
+    "/principal",
     dependencies=[Depends(verify_auth), Depends(require_master_db)],
     error_map=error_map,
 )
 async def rename_principal(
-    principal_name: Annotated[LIMITED_STR, Body()],
-    principal_new_name: Annotated[LIMITED_STR, Body()],
+    request: PrincipalPutRequest,
     kerberos_adapter: FromDishka[KerberosFastAPIAdapter],
 ) -> None:
-    """Rename principal in kerberos with given name.
-
-    \f
-    :param Annotated[str, Body principal_name: upn
-    :param Annotated[LIMITED_STR, Body principal_new_name: _description_
-    :param Annotated[LDAPSession, Depends ldap_session: ldap
-    :raises HTTPException: on failed kamin request.
-    """
-    await kerberos_adapter.rename_principal(
-        principal_name,
-        principal_new_name,
-    )
+    await kerberos_adapter.rename_principal(request)
 
 
 @krb5_router.patch(

app/api/main/schema.py

@@ -70,6 +70,30 @@ class KerberosSetupRequest(BaseModel):
     stash_password: SecretStr
 
 
+class PrincipalAddRequest(BaseModel):
+    """Request schema for POST /principal/add."""
+
+    principal_name: str
+    algorithms: list[str] | None = None
+    password: str | None = None
+
+
+class KtaddRequest(BaseModel):
+    """Request schema for POST /ktadd."""
+
+    names: list[str]
+    is_rand_key: bool = False
+
+
+class PrincipalPutRequest(BaseModel):
+    """Request schema for PUT /principal (full modify)."""
+
+    principal_name: str
+    new_principal_name: str
+    algorithms: list[str] | None = None
+    password: str | None = None
+
+
 class DNSServiceSetupRequest(BaseModel):
     """DNS setup request schema."""
 

app/ldap_protocol/kerberos/base.py

@@ -153,8 +153,9 @@ async def setup(
     @abstractmethod
     async def add_principal(
         self,
-        name: str,
-        password: str | None,
+        principal_name: str,
+        password: str | None = None,
+        algorithms: list[str] | None = None,
         timeout: int | float = 1,
     ) -> None: ...
 
@@ -179,7 +180,13 @@ async def create_or_update_principal_pw(
     ) -> None: ...
 
     @abstractmethod
-    async def rename_princ(self, name: str, new_name: str) -> None: ...
+    async def rename_princ(
+        self,
+        name: str,
+        new_name: str,
+        algorithms: list[str] | None = None,
+        password: str | None = None,
+    ) -> None: ...
 
     @backoff.on_exception(
         backoff.constant,
@@ -202,7 +209,11 @@ async def get_status(self, wait_for_positive: bool = False) -> bool:
         return status
 
     @abstractmethod
-    async def ktadd(self, names: list[str]) -> httpx.Response: ...
+    async def ktadd(
+        self,
+        names: list[str],
+        is_rand_key: bool,
+    ) -> httpx.Response: ...
 
     @abstractmethod
     async def lock_principal(self, name: str) -> None: ...
@@ -221,14 +232,17 @@ async def ldap_principal_setup(self, name: str, path: str) -> None:
         if response.status_code == 200:
             return
 
-        response = await self.client.post("/principal", json={"name": name})
+        response = await self.client.post(
+            "/principal",
+            json={"principal_name": name},
+        )
         if response.status_code != 201:
             log.error(f"Error creating ldap principal: {response.text}")
             return
 
         response = await self.client.post(
             "/principal/ktadd",
-            json=[name],
+            json={"names": [name], "is_rand_key": True},
         )
         if response.status_code != 200:
             log.error(f"Error getting keytab: {response.text}")

app/ldap_protocol/kerberos/client.py

@@ -20,12 +20,17 @@ async def add_principal(
         self,
         name: str,
         password: str | None,
-        timeout: int = 1,
+        algorithms: list[str] | None = None,
+        timeout: int | float = 1,
     ) -> None:
         """Add request."""
         response = await self.client.post(
             "principal",
-            json={"name": name, "password": password},
+            json={
+                "name": name,
+                "password": password,
+                "algorithms": algorithms,
+            },
             timeout=timeout,
         )
 
@@ -89,17 +94,32 @@ async def create_or_update_principal_pw(
             raise krb_exc.KRBAPIChangePasswordError(response.text)
 
     @logger_wraps()
-    async def rename_princ(self, name: str, new_name: str) -> None:
+    async def rename_princ(
+        self,
+        name: str,
+        new_name: str,
+        algorithms: list[str] | None,
+        password: str | None,
+    ) -> None:
         """Rename request."""
         response = await self.client.put(
             "principal",
-            json={"name": name, "new_name": new_name},
+            json={
+                "name": name,
+                "new_name": new_name,
+                "algorithms": algorithms,
+                "password": password,
+            },
         )
         if response.status_code != 202:
             raise krb_exc.KRBAPIRenamePrincipalError(response.text)
 
     @logger_wraps()
-    async def ktadd(self, names: list[str]) -> httpx.Response:
+    async def ktadd(
+        self,
+        names: list[str],
+        is_rand_key: bool,
+    ) -> httpx.Response:
         """Ktadd build request for stream and return response.
 
         :param list[str] names: principals
@@ -108,7 +128,7 @@ async def ktadd(self, names: list[str]) -> httpx.Response:
         request = self.client.build_request(
             "POST",
             "/principal/ktadd",
-            json=names,
+            json={"names": names, "is_rand_key": is_rand_key},
         )
 
         response = await self.client.send(request, stream=True)

app/ldap_protocol/kerberos/service.py

@@ -358,7 +358,12 @@ async def _schedule_principal_task(
         )
         return TaskStruct(func=func, args=args)
 
-    async def add_principal(self, primary: str, instance: str) -> None:
+    async def add_principal(
+        self,
+        principal_name: str,
+        password: str | None,
+        algorithms: list[str] | None,
+    ) -> None:
         """Create principal in Kerberos with given name.
 
         :param str primary: Principal primary name.
@@ -367,8 +372,11 @@ async def add_principal(self, primary: str, instance: str) -> None:
         :return None: None.
         """
         try:
-            principal_name = f"{primary}/{instance}"
-            await self._kadmin.add_principal(principal_name, None)
+            await self._kadmin.add_principal(
+                principal_name,
+                password,
+                algorithms,
+            )
         except KRBAPIAddPrincipalError as exc:
             raise KerberosDependencyError(
                 f"Error adding principal: {exc}",
@@ -378,16 +386,25 @@ async def rename_principal(
         self,
         principal_name: str,
         principal_new_name: str,
+        algorithms: list[str] | None,
+        password: str | None,
     ) -> None:
         """Rename principal in Kerberos with given name.
 
         :param str principal_name: Current principal name.
         :param str principal_new_name: New principal name.
+        :param list[str] | None algorithms: Algorithms.
+        :param str | None password: Password.
         :raises KerberosDependencyError: On failed kadmin request.
         :return None: None.
         """
         try:
-            await self._kadmin.rename_princ(principal_name, principal_new_name)
+            await self._kadmin.rename_princ(
+                principal_name,
+                principal_new_name,
+                algorithms,
+                password,
+            )
         except KRBAPIRenamePrincipalError as exc:
             raise KerberosDependencyError(
                 f"Error renaming principal: {exc}",
@@ -432,15 +449,17 @@ async def delete_principal(self, principal_name: str) -> None:
     async def ktadd(
         self,
         names: list[str],
+        is_rand_key: bool,
     ) -> tuple[AsyncIterator[bytes], TaskStruct]:
         """Generate keytab and return (aiter_bytes, TaskStruct).
 
         :param list[str] names: List of principal names.
+        :param bool is_rand_key: If True, generate random key.
         :raises KerberosNotFoundError: If principal not found.
         :return tuple: (aiter_bytes, (func, args, kwargs)).
         """
         try:
-            response = await self._kadmin.ktadd(names)
+            response = await self._kadmin.ktadd(names, is_rand_key)
         except KRBAPIPrincipalNotFoundError:
             raise KerberosNotFoundError("Principal not found")
         aiter_bytes = response.aiter_bytes()