Update TokenConfig and add JWT models

neon_data_models/enum.py

@@ -38,9 +38,11 @@ class AccessRoles(IntEnum):
     NONE = 0
     GUEST = 1
     USER = 2
-    ADMIN = 3
-    OWNER = 4
-
+    # 3-5 Reserved for "premium users"
+    ADMIN = 6
+    # 7-8 Reserved for "restricted owners"
+    OWNER = 9
+    # 10 Reserved for "unlimited access"
     NODE = -1
 
 

neon_data_models/models/api/jwt.py

@@ -0,0 +1,71 @@
+# NEON AI (TM) SOFTWARE, Software Development Kit & Application Development System
+# All trademark and other rights reserved by their respective owners
+# Copyright 2008-2024 Neongecko.com Inc.
+# BSD-3
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+# 1. Redistributions of source code must retain the above copyright notice,
+#    this list of conditions and the following disclaimer.
+# 2. Redistributions in binary form must reproduce the above copyright notice,
+#    this list of conditions and the following disclaimer in the documentation
+#    and/or other materials provided with the distribution.
+# 3. Neither the name of the copyright holder nor the names of its
+#    contributors may be used to endorse or promote products derived from this
+#    software without specific prior written permission.
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
+# THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
+# CONTRIBUTORS  BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA,
+# OR PROFITS;  OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+# SOFTWARE,  EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+from typing import Optional, List, Literal
+from uuid import uuid4
+
+from pydantic import Field
+from neon_data_models.enum import AccessRoles
+from neon_data_models.models.base import BaseModel
+
+
+class JWT(BaseModel):
+    iss: Optional[str] = Field(None, description="Token issuer")
+    sub: Optional[str] = Field(None,
+                               description="Unique token subject, ie a user ID")
+    exp: int = Field(None,
+                     description="Expiration time in epoch seconds")
+    iat: int = Field(None,
+                     description="Token creation time in epoch seconds")
+    jti: str = Field(description="Unique token identifier",
+                     default_factory=lambda: str(uuid4()))
+
+    client_id: str = Field(None, description="Client identifier")
+    roles: List[str] = Field(None,
+                             description="List of roles, "
+                                         "formatted as `<name> <AccessRole>`. "
+                                         "See PermissionsConfig for role names")
+
+
+class HanaToken(JWT):
+    def __init__(self, **kwargs):
+        permissions = kwargs.get("permissions")
+        if permissions and isinstance(permissions, dict):
+            core_permissions = AccessRoles.GUEST if \
+                permissions.get("assist") else AccessRoles.NONE
+            diana_permissions = AccessRoles.GUEST if \
+                permissions.get("backend") else AccessRoles.NONE
+            node_permissions = AccessRoles.USER if \
+                permissions.get("node") else AccessRoles.NONE
+            kwargs["roles"] = [f"core {core_permissions.value}",
+                               f"diana {diana_permissions.value}",
+                               f"node {node_permissions.value}"]
+        if kwargs.get("expire") and isinstance(kwargs["expire"], float):
+            kwargs["expire"] = round(kwargs["expire"])
+        BaseModel.__init__(self, **kwargs)
+
+    # Private parameters
+    purpose: Literal["access", "refresh"] = "access"

neon_data_models/models/api/node_v1/__init__.py

@@ -80,8 +80,8 @@ class TextInputData(BaseModel):
 
 class NodeKlatResponse(BaseMessage):
     msg_type: Literal["klat.response"] = "klat.response"
-    data: Dict[str, KlatResponse] = Field(type=Dict[str, KlatResponse],
-                       description="dict of BCP-47 language: KlatResponse")
+    data: Dict[str, KlatResponse] = Field(
+        description="dict of BCP-47 language: KlatResponse")
 
 
 class NodeAudioInputResponse(BaseMessage):
@@ -97,8 +97,7 @@ class NodeGetSttResponse(BaseMessage):
 class NodeGetTtsResponse(BaseMessage):
     msg_type: Literal["neon.get_tts.response"] = "neon.get_tts.response"
     data: Dict[str, KlatResponse] = (
-        Field(type=Dict[str, KlatResponse],
-              description="dict of BCP-47 language: KlatResponse"))
+        Field(description="dict of BCP-47 language: KlatResponse"))
 
 
 class CoreWWDetected(BaseMessage):

neon_data_models/models/user/database.py

@@ -124,22 +124,55 @@ class PermissionsConfig(BaseModel):
     class Config:
         use_enum_values = True
 
+    @classmethod
+    def from_roles(cls, roles: List[str]):
+        """
+        Parse PermissionsConfig from standard JWT roles configuration.
+        """
+        kwargs = {}
+        for role in roles:
+            name, value = role.split(' ')
+            kwargs[name] = AccessRoles[value.upper()]
+        return cls(**kwargs)
+
+    def to_roles(self):
+        """
+        Dump a PermissionsConfig to standard JWT roles to be included in a JWT.
+        """
+        roles = []
+        for key, val in self.model_dump().items():
+            roles.append(f"{key} {AccessRoles(val).name}")
+        return roles
+
 
 class TokenConfig(BaseModel):
-    username: str
-    client_id: str
-    permissions: Dict[str, bool]
-    refresh_token: str
-    expiration: int = Field(
-        description="Unix timestamp of auth token expiration")
-    refresh_expiration: int = Field(
+    """
+    Data model for storing token data in the user database. Note that the actual
+    tokens are not included here, only metadata used to validate or invalidate a
+    token and present a list of issued tokens to the user.
+    """
+    def __init__(self, **kwargs):
+        # The JWT standard uses "jti" and "sub" for encoded values. Outside of
+        # that context, those keys are not very descriptive, so we use our own
+        if jti := kwargs.get("jti"):
+            kwargs.setdefault("token_id", jti)
+        if sub := kwargs.get("sub"):
+            kwargs.setdefault("user_id", sub)
+        BaseModel.__init__(self, **kwargs)
+
+    token_name: str = Field(description="Human-readable token identifier")
+    token_id: str = Field(description="Unique token identifier")
+    user_id: str = Field(description="User ID the token is associated with")
+    client_id: str = Field(description="Client ID the token is associated with")
+    permissions: PermissionsConfig = Field(
+        description="Permissions for this token "
+                    "(overrides user-level permissions)")
+    refresh_expiration_timestamp: int = Field(
         description="Unix timestamp of refresh token expiration")
-    token_name: str
     creation_timestamp: int = Field(
-        description="Unix timestamp of auth token creation")
+        description="Unix timestamp of refresh token creation")
     last_refresh_timestamp: int = Field(
         description="Unix timestamp of last auth token refresh")
-    access_token: Optional[str] = None
 
 
 class User(BaseModel):

tests/models/test_user.py

@@ -26,6 +26,8 @@
 
 from time import time
 from unittest import TestCase
+from uuid import uuid4
+
 from pydantic import ValidationError
 from datetime import date
 from neon_data_models.models.user.database import NeonUserConfig, TokenConfig, User
@@ -83,13 +85,12 @@ def test_neon_user_config(self):
     def test_user(self):
         user_kwargs = dict(username="test",
                            password_hash="test",
-                           tokens=[{"username": "test",
-                                    "client_id": "test_id",
+                           tokens=[{"token_name": "test_token",
+                                    "token_id": str(uuid4()),
+                                    "user_id": str(uuid4()),
+                                    "client_id": str(uuid4()),
                                     "permissions": {},
-                                    "refresh_token": "",
-                                    "expiration": round(time()),
-                                    "refresh_expiration": round(time()),
-                                    "token_name": "test_token",
+                                    "refresh_expiration_timestamp": round(time()),
                                     "creation_timestamp": round(time()),
                                     "last_refresh_timestamp": round(time())}])
         default_user = User(**user_kwargs)
@@ -105,6 +106,36 @@ def test_user(self):
         self.assertNotEqual(default_user, duplicate_user)
         self.assertEqual(default_user.tokens, duplicate_user.tokens)
 
+    def test_permissions_config(self):
+        from neon_data_models.models.user.database import PermissionsConfig
+        from neon_data_models.enum import AccessRoles
+
+        # Test Default
+        default_config = PermissionsConfig()
+        for _, value in default_config.model_dump().items():
+            self.assertEqual(value, AccessRoles.NONE)
+
+        test_config = PermissionsConfig(klat=AccessRoles.USER,
+                                        core=AccessRoles.GUEST,
+                                        diana=AccessRoles.GUEST,
+                                        node=AccessRoles.NODE,
+                                        hub=AccessRoles.NODE,
+                                        llm=AccessRoles.NONE)
+        # Test dump/load
+        self.assertEqual(PermissionsConfig(**test_config.model_dump()),
+                         test_config)
+
+        # Test to/from roles
+        roles = test_config.to_roles()
+        self.assertIsInstance(roles, list)
+        for role in roles:
+            self.assertEqual(len(role.split()), 2)
+        self.assertEqual(PermissionsConfig.from_roles(roles), test_config)
+
+    def test_token_config(self):
+        from neon_data_models.models.user.database import TokenConfig
+        # TODO
+
 
 class TestNeonProfile(TestCase):
     def test_create(self):