pykanidm: Add retrieving credential reset token for a person. (#3279)

pykanidm/kanidm/__init__.py

@@ -17,7 +17,13 @@ import yarl
 
 from kanidm.models.group import Group, GroupList, IGroup, RawGroup
 from kanidm.models.oauth2_rs import IOauth2Rs, OAuth2Rs, Oauth2RsList, RawOAuth2Rs
-from kanidm.models.person import IPerson, Person, PersonList, RawPerson
+from kanidm.models.person import (
+    IPerson,
+    Person,
+    PersonList,
+    RawPerson,
+    PersonCredentialResetToken,
+)
 from kanidm.models.service_account import (
     IServiceAccount,
     ServiceAccount,
@@ -765,6 +771,21 @@ class KanidmClient:
         endpoint = f"{Endpoints.PERSON}/{id}"
         return await self.call_delete(endpoint)
 
+    async def person_account_credential_update_token(self, id: str, ttl: Optional[int] = None) -> PersonCredentialResetToken:
+        """Create a password reset token for person with an optional time to live in seconds"""
+        endpoint = f"{Endpoints.PERSON}/{id}/_credential/_update_intent"
+        if ttl:
+            endpoint = f"{endpoint}/{ttl}"
+
+        response: ClientResponse[Any] = await self.call_get(endpoint)
+        if response.content is None:
+            raise ValueError(f"Failed to get token: {response.content}")
+        if response.status_code != 200:
+            raise ValueError(f"Failed to get token: {response.content}")
+        token = PersonCredentialResetToken.model_validate(json_lib.loads(response.content))
+
+        return token
+
     async def person_account_post_ssh_key(self, id: str, tag: str, pubkey: str) -> ClientResponse[None]:
         """Create an SSH key for a user"""
         endpoint = f"{Endpoints.PERSON}/{id}/_ssh_pubkeys"

pykanidm/kanidm/models/person.py

@@ -38,8 +38,15 @@ class RawPerson(BaseModel):
             uuid=UUID(self.attrs["uuid"][0]),
         )
 
+
 PersonList = RootModel[List[RawPerson]]
 
 
 class IPerson(TypedDict):
     attrs: Dict[str, List[str]]
+
+
+class PersonCredentialResetToken(BaseModel):
+    token: str
+    expiry_time: int
+    model_config = ConfigDict(arbitrary_types_allowed=True)

pykanidm/kanidm/types.py

@@ -1,4 +1,5 @@
 """ type objects """
+
 # pylint: disable=too-few-public-methods
 # ^ disabling this because pydantic models don't have public methods
 
@@ -31,7 +32,7 @@ class ClientResponse(BaseModel, Generic[T]):
 
 
 class AuthInitResponse(BaseModel):
-    """Aelps parse the response from the Auth 'init' stage"""
+    """Helps parse the response from the Auth 'init' stage"""
 
     class _AuthInitState(BaseModel):
         """sub-class for the AuthInitResponse model"""
@@ -146,9 +147,7 @@ class RadiusClient(BaseModel):
             socket.gethostbyname(value)
             return value
         except socket.gaierror as error:
-            raise ValueError(
-                f"ipaddr value ({value}) wasn't an IP Address, Network or valid hostname: {error}"
-            )
+            raise ValueError(f"ipaddr value ({value}) wasn't an IP Address, Network or valid hostname: {error}")
 
 
 class KanidmClientConfig(BaseModel):
@@ -196,9 +195,7 @@ class KanidmClientConfig(BaseModel):
             uri = urlparse(value)
             valid_schemes = ["http", "https"]
             if uri.scheme not in valid_schemes:
-                raise ValueError(
-                    f"Invalid URL Scheme for uri='{value}': '{uri.scheme}' - expected one of {valid_schemes}"
-                )
+                raise ValueError(f"Invalid URL Scheme for uri='{value}': '{uri.scheme}' - expected one of {valid_schemes}")
 
             # make sure the URI ends with a /
             if not value.endswith("/"):