move method to issue token inside CoreAuthHandler

Author: pieroit

core/cat/auth/headers.py

@@ -113,8 +113,8 @@ async def get_user_stray(user: AuthUserInfo):
             else:
                 stray = StrayCat(
                     ws=websocket,
-                    user_id=user.id,
-                    user_data=user.extra,
+                    user_id=user.name, # TODOV2: user_id should be the user.id
+                    user_data=user,
                     main_loop=asyncio.get_running_loop(),
                 )
                 strays[user.id] = stray
@@ -167,7 +167,8 @@ async def get_user_stray(user: AuthUserInfo):
 
             if user.id not in strays.keys():
                 strays[user.id] = StrayCat(
-                    user_id=user.id, user_data=user.extra, main_loop=event_loop
+                     # TODOV2: user_id should be the user.id
+                    user_id=user.name, user_data=user, main_loop=event_loop
                 )
             return strays[user.id]
 
@@ -227,7 +228,8 @@ async def frontend_auth(request: Request) -> None | StrayCat:
 
             if user.id not in strays.keys():
                 strays[user.id] = StrayCat(
-                    user_id=user.id, user_data=user.extra, main_loop=event_loop
+                     # TODOV2: user_id should be the user.id
+                    user_id=user.name, user_data=user, main_loop=event_loop
                 )
             return strays[user.id]
 

core/cat/factory/custom_auth_handler.py

@@ -1,9 +1,11 @@
 from abc import ABC, abstractmethod
+from datetime import datetime, timedelta
+from pytz import utc
 import jwt
 
 from cat.db.crud import get_users
 from cat.auth.utils import (
-    AuthPermission, AuthResource, AuthUserInfo, get_base_permissions, get_full_permissions, is_jwt
+    AuthPermission, AuthResource, AuthUserInfo, get_base_permissions, get_full_permissions, is_jwt, check_password
 )
 from cat.env import get_env
 from cat.log import log
@@ -56,6 +58,7 @@ async def authorize_user_from_key(
 
 # Core auth handler, verify token on local idp
 class CoreAuthHandler(BaseAuthHandler):
+
     async def authorize_user_from_jwt(
         self, token: str, auth_resource: AuthResource, auth_permission: AuthPermission
     ) -> AuthUserInfo | None:
@@ -118,6 +121,34 @@ async def authorize_user_from_key(
 
         # do not pass
         return None
+    
+    async def issue_jwt(self, username: str, password: str) -> str | None:
+        # authenticate local user credentials and return a JWT token
+
+        # brutal search over users, which are stored in a simple dictionary.
+        # waiting to have graph in core to store them properly
+        # TODOAUTH: get rid of this shameful loop
+        users = get_users()
+        for user_id, user in users.items():
+            if user["username"] == username and check_password(password, user["password"]):
+                # TODOAUTH: expiration with timezone needs to be tested
+                # using seconds for easier testing
+                expire_delta_in_seconds = float(get_env("CCAT_JWT_EXPIRE_MINUTES")) * 60
+                expires = datetime.now(utc) + timedelta(seconds=expire_delta_in_seconds)
+                # TODOAUTH: add issuer and redirect_uri (and verify them when a token is validated)
+
+                jwt_content = {
+                    "sub": user_id,                      # Subject (the user ID)
+                    "username": username,                # Username
+                    "permissions": user["permissions"],  # User permissions
+                    "exp": expires                       # Expiry date as a Unix timestamp
+                }
+                return jwt.encode(
+                    jwt_content,
+                    get_env("CCAT_JWT_SECRET"),
+                    algorithm=get_env("CCAT_JWT_ALGORITHM"),
+                )
+        return None
 
 
 # Default Auth, always deny auth by default (only core auth decides).

core/cat/routes/auth.py

@@ -1,21 +1,15 @@
-from pytz import utc
 import asyncio
 from typing import Dict, List
 from urllib.parse import urlencode
-from datetime import datetime, timedelta
-import jwt
 from pydantic import BaseModel
 
 from fastapi import APIRouter, Request, HTTPException, Response, status, Query
 from fastapi.responses import RedirectResponse
 
 
-from cat.auth.utils import AuthPermission, AuthResource, get_full_permissions, check_password
-from cat.db.crud import get_users
-from cat.env import get_env
+from cat.auth.utils import AuthPermission, AuthResource, get_full_permissions
 from cat.routes.static.templates import get_jinja_templates
 
-
 router = APIRouter()
 
 class UserCredentials(BaseModel):
@@ -34,7 +28,8 @@ async def core_login_token(request: Request, response: Response):
     form_data = await request.form()
 
     # use username and password to authenticate user from local identity provider and get token
-    access_token = await authenticate_local_user(
+    auth_handler = request.app.state.ccat.core_auth_handler
+    access_token = await auth_handler.issue_jwt(
         form_data["username"], form_data["password"]
     )
 
@@ -86,13 +81,14 @@ async def get_available_permissions():
     return get_full_permissions()
 
 @router.post("/token", response_model=JWTResponse)
-async def auth_token(credentials: UserCredentials):
+async def auth_token(request: Request, credentials: UserCredentials):
     """Endpoint called from client to get a JWT from local identity provider.
     This endpoint receives username and password as form-data, validates credentials and issues a JWT.
     """
 
     # use username and password to authenticate user from local identity provider and get token
-    access_token = await authenticate_local_user(
+    auth_handler = request.app.state.ccat.core_auth_handler
+    access_token = await auth_handler.issue_jwt(
         credentials.username, credentials.password
     )
 
@@ -103,33 +99,3 @@ async def auth_token(credentials: UserCredentials):
     # wait a little to avoid brute force attacks
     await asyncio.sleep(1)
     raise HTTPException(status_code=403, detail={"error": "Invalid Credentials"})
-
-
-async def authenticate_local_user(username: str, password: str) -> str | None:
-    # authenticate local user credentials and return a JWT token
-
-    # brutal search over users, which are stored in a simple dictionary.
-    # waiting to have graph in core to store them properly
-    # TODOAUTH: get rid of this shameful loop
-    users = get_users()
-    for user_id, user in users.items():
-        if user["username"] == username and check_password(password, user["password"]):
-            # TODOAUTH: expiration with timezone needs to be tested
-            # using seconds for easier testing
-            expire_delta_in_seconds = float(get_env("CCAT_JWT_EXPIRE_MINUTES")) * 60
-            expires = datetime.now(utc) + timedelta(seconds=expire_delta_in_seconds)
-            # TODOAUTH: add issuer and redirect_uri (and verify them when a token is validated)
-            # TODOAUTH: move this method in auth.utils
-
-            jwt_content = {
-                "sub": user_id,                      # Subject (the user ID)
-                "username": username,                # Username
-                "permissions": user["permissions"],  # User permissions
-                "exp": expires                       # Expiry date as a Unix timestamp
-            }
-            return jwt.encode(
-                jwt_content,
-                get_env("CCAT_JWT_SECRET"),
-                algorithm=get_env("CCAT_JWT_ALGORITHM"),
-            )
-    return None

core/tests/routes/auth/test_jwt.py

@@ -39,7 +39,6 @@ async def test_issue_jwt(client):
         "password": "admin"
     }
     res = client.post("/auth/token", json=creds)
-
     assert res.status_code == 200
 
     # did we obtain a JWT?
@@ -52,11 +51,8 @@ async def test_issue_jwt(client):
     user_info = await auth_handler.authorize_user_from_jwt(
         received_token, AuthResource.ADMIN, AuthPermission.WRITE
     )
-    assert user_info.user_id == "admin"
-    assert user_info.user_data["username"] == "admin"
-    assert (
-        user_info.user_data["exp"] - time.time() < 60 * 60 * 24
-    )  # expires in less than 24 hours
+    assert len(user_info.id) == 36 and len(user_info.id.split("-")) == 5 # uuid4
+    assert user_info.name == "admin"
 
     # manual JWT verification
     try: