Personal access tokens (#150)

* Personal access tokens * fix and add test * fix schema * add fixture * fix test * - --------- Co-authored-by: David Ruiz Falco <[email protected]>
nuclia · Jan 30, 2025 · e23a6aa · e23a6aa
1 parent 18087e3
commit e23a6aa
Show file tree

Hide file tree

Showing 6 changed files with 108 additions and 4 deletions.
diff --git a/.vscode/settings.json b/.vscode/settings.json
@@ -9,5 +9,10 @@
     "labelsets",
     "Serviceaccount",
     "tqdm"
-  ]
+  ],
+  "python.testing.pytestArgs": [
+    "nuclia"
+  ],
+  "python.testing.unittestEnabled": false,
+  "python.testing.pytestEnabled": true
 }
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,9 +1,10 @@
 # Changelog
 
-## 4.4.10 (unreleased)
+## 4.5.0 (unreleased)
 
 
 - Add AI agents documentation
+- Support Personal Access Tokens
 
 
 ## 4.4.9 (2025-01-29)

diff --git a/docs/02-auth.md b/docs/02-auth.md
@@ -20,6 +20,47 @@ It is a good fit for interactive use, but not for long running scripts.
   sdk.NucliaAuth().set_user_token(USER_TOKEN)
   ```
 
+## Personal access token
+
+A personal access token is a long-lived user token that can be used to authenticate to the API. It is a good fit for long running scripts. It grants the same access as user authentication.
+
+You define its expiration date (default is 90 days), and you can revoke it at any time.
+
+- CLI:
+
+  To generate a token, run:
+
+  ```sh
+  nuclia auth create_personal_token --description="My token" --days=30
+  ```
+
+  To generate a token and use it as authentication in the CLI, run:
+
+  ```sh
+  nuclia auth create_personal_token --description="My token" --days=30 --login
+  ```
+
+  To list the tokens, run:
+
+  ```sh
+  nuclia auth list_personal_tokens
+  ```
+
+  To delete a token, run:
+
+  ```sh
+  nuclia auth delete_personal_token [TOKEN_ID]
+  ```
+
+- SDK:
+
+  ```python
+  from nuclia import sdk
+  sdk.NucliaAuth().create_personal_token(description="My token", days=30)
+  sdk.NucliaAuth().list_personal_tokens()
+  sdk.NucliaAuth().delete_personal_token(token_id=TOKEN_ID)
+  ```
+
 ## API key
 
 An API key can be generated from the Nuclia Dashboard, see [Get an API key](https://docs.nuclia.dev/docs/guides/getting-started/quick-start/push#get-an-api-key).

diff --git a/nuclia/config.py b/nuclia/config.py
@@ -1,7 +1,7 @@
 from enum import Enum
 import os
 from typing import List, Optional
-
+from datetime import datetime
 from pydantic import BaseModel
 
 from nuclia import CLOUD_ID
@@ -42,6 +42,18 @@ def __str__(self):
         return f"{self.client_id} {self.account} {self.account_type:30}"
 
 
+class PersonalTokenCreate(BaseModel):
+    id: str
+    token: str
+    expires: Optional[datetime]
+
+
+class PersonalTokenItem(BaseModel):
+    id: str
+    description: str
+    expires: Optional[datetime]
+
+
 class Zone(BaseModel):
     id: str
     title: str

diff --git a/nuclia/sdk/auth.py b/nuclia/sdk/auth.py
@@ -1,7 +1,9 @@
 import base64
+import datetime
 import json
 import webbrowser
 from typing import Any, Dict, List, Optional, Tuple
+from pydantic import TypeAdapter
 
 from httpx import AsyncClient, Client, ConnectError
 from prompt_toolkit import prompt
@@ -13,6 +15,8 @@
     Account,
     Config,
     KnowledgeBox,
+    PersonalTokenCreate,
+    PersonalTokenItem,
     User,
     Zone,
     retrieve_account,
@@ -26,6 +30,8 @@
 ZONES = "/api/v1/zones"
 LIST_KBS = "/api/v1/account/{account}/kbs"
 VERIFY_NUA = "/api/authorizer/info"
+PERSONAL_TOKENS = "/api/v1/user/pa_tokens"
+PERSONAL_TOKEN = "/api/v1/user/pa_token/{token_id}"
 
 
 class BaseNucliaAuth:
@@ -345,6 +351,33 @@ def post_login(self):
         self.accounts()
         self.zones()
 
+    def create_personal_token(
+        self, description: str, days: int = 90, login: bool = False
+    ) -> PersonalTokenCreate:
+        expiration_date = datetime.datetime.now() + datetime.timedelta(days=days)
+        resp = self._request(
+            "POST",
+            get_global_url(PERSONAL_TOKENS),
+            {
+                "description": description,
+                "expiration_date": expiration_date.isoformat(),
+            },
+        )
+        token = PersonalTokenCreate.model_validate(resp)
+        if login:
+            self.set_user_token(token.token)
+        return token
+
+    def delete_personal_token(self, token_id: str) -> None:
+        self._request(
+            "DELETE", get_global_url(PERSONAL_TOKEN.format(token_id=token_id))
+        )
+
+    def list_personal_tokens(self) -> List[PersonalTokenItem]:
+        resp = self._request("GET", get_global_url(PERSONAL_TOKENS))
+        ta = TypeAdapter(List[PersonalTokenItem])
+        return ta.validate_python(resp)
+
     def _request(
         self, method: str, path: str, data: Optional[Any] = None, remove_null=True
     ):
@@ -357,6 +390,7 @@ def _request(
             if remove_null:
                 data = {k: v for k, v in data.items() if v is not None}
             kwargs["content"] = json.dumps(data)
+
         resp = self.client.request(
             method,
             path,
@@ -617,7 +651,7 @@ async def _request(
             return resp.json()
         elif resp.status_code >= 300 and resp.status_code < 400:
             return None
-        elif resp.status_code == 403:
+        elif resp.status_code == 403 or resp.status_code == 401:
             raise UserTokenExpired()
         else:
             raise Exception({"status": resp.status_code, "message": resp.text})

diff --git a/nuclia/tests/test_manage/test_auth.py b/nuclia/tests/test_manage/test_auth.py
@@ -22,3 +22,14 @@ def test_auth_nua(testing_nua: str):
     assert client
     assert account_type
     assert account
+
+
+def test_auth_pat(testing_config):
+    na = NucliaAuth()
+    token = na.create_personal_token(description="sdk test token", days=1, login=False)
+    assert token
+    tokens = na.list_personal_tokens()
+    assert len([t.id for t in tokens if t.id == token.id]) == 1
+    na.delete_personal_token(token_id=token.id)
+    tokens = na.list_personal_tokens()
+    assert len([t.id for t in tokens if t.id == token.id]) == 0