Remove CAS-specific logic

Author: gfrn

src/scaup/auth/micro.py

@@ -1,8 +1,9 @@
-from typing import TypeVar
+from typing import List, TypeVar
 
 import requests
 from fastapi import Depends, HTTPException, Request, status
 from fastapi.security import HTTPAuthorizationCredentials
+from jwt import InvalidAlgorithmError, InvalidAudienceError
 from lims_utils.auth import GenericUser
 from lims_utils.logging import app_logger
 from lims_utils.models import parse_proposal
@@ -23,21 +24,8 @@
 T = TypeVar("T")
 
 
-def _is_cas_token(token: str):
-    return len(token) > 2 and token[:2] == "AT"
-
-
 def _get_user(token: str):
-    if _is_cas_token(token):
-        response = requests.get(
-            Config.auth.endpoint + "/user",
-            headers={"Authorization": f"Bearer {token}"},
-        )
-
-        if response.status_code != 200:
-            raise HTTPException(status_code=response.status_code, detail=response.json().get("detail"))
-        return response.json()
-    else:
+    try:
         # TODO: replace this once something more permanent becomes available
         user = decode_jwt(token, "scaup_general")
         app_id = user.get("sub")
@@ -51,6 +39,15 @@ def _get_user(token: str):
             "permissions": user.get("permissions"),
             "email": "",
         }
+    except (InvalidAudienceError, InvalidAlgorithmError):
+        response = requests.get(
+            Config.auth.endpoint + "/user",
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        if response.status_code != 200:
+            raise HTTPException(status_code=response.status_code, detail=response.json().get("detail"))
+        return response.json()
 
 
 class User(GenericUser):
@@ -67,33 +64,33 @@ def __init__(
 
 
 def _check_perms(data_id: T, endpoint: str, token: str) -> T:
-    if not _is_cas_token(token):
-        user = _get_user(token)
-        if is_admin(user["permissions"]):
+    try:
+        permissions: List[str] = decode_jwt(token, "scaup_general").get("permissions", [])
+        if is_admin(permissions):
             return data_id
 
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Provided JWT lacks permissions")
+    except (InvalidAudienceError, InvalidAlgorithmError):
+        response = requests.get(
+            "".join(
+                [
+                    Config.auth.endpoint,
+                    "/permission/",
+                    endpoint,
+                    "/",
+                    str(data_id) if endpoint != "proposal" else str(data_id) + "/inSessions",
+                ]
+            ),
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        if response.status_code != 200:
+            detail = response.json().get("detail")
+            app_logger.error(f"Microauth returned {response.status_code}: {detail}")
+
+            raise HTTPException(status_code=response.status_code, detail=detail)
 
-    response = requests.get(
-        "".join(
-            [
-                Config.auth.endpoint,
-                "/permission/",
-                endpoint,
-                "/",
-                str(data_id) if endpoint != "proposal" else str(data_id) + "/inSessions",
-            ]
-        ),
-        headers={"Authorization": f"Bearer {token}"},
-    )
-
-    if response.status_code != 200:
-        detail = response.json().get("detail")
-        app_logger.error(f"Microauth returned {response.status_code}: {detail}")
-
-        raise HTTPException(status_code=response.status_code, detail=detail)
-
-    return data_id
+        return data_id
 
 
 def _generic_table_check(

src/scaup/utils/auth.py

@@ -1,14 +1,14 @@
 from typing import Any
 
 from fastapi import HTTPException, status
-from jwt import DecodeError, ExpiredSignatureError, InvalidAudienceError, decode
+from jwt import DecodeError, decode
 from lims_utils.auth import GenericUser
 from lims_utils.logging import app_logger
 
 from .config import Config
 
 
-def is_admin(perms: list[int]):
+def is_admin(perms: list[str]):
     return bool(set(Config.auth.read_all_perms) & set(perms))
 
 
@@ -30,8 +30,8 @@ def decode_jwt(token: str, aud: str = Config.shipping_service.callback_url) -> d
         )
 
         return decoded_body
-    except (DecodeError, ExpiredSignatureError, InvalidAudienceError) as e:
-        app_logger.warning(f"Error while parsing token {token}: {e}")
+    except DecodeError as e:
+        app_logger.warning(f"Error while parsing token: {e}")
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token provided")
 
 

tests/utils/test_auth.py

@@ -1,6 +1,7 @@
 import jwt
 import pytest
 from fastapi import HTTPException
+from jwt.exceptions import ExpiredSignatureError, InvalidAudienceError
 
 from scaup.utils.auth import check_em_staff, check_jwt
 from scaup.utils.config import Config
@@ -33,7 +34,7 @@ def test_jwt_invalid_aud():
     """Should not allow unmatched audiences"""
     token = jwt.encode({"id": 1, "exp": 9e9, "aud": "invalid-aud"}, Config.auth.jwt_private, algorithm="ES256")
 
-    with pytest.raises(HTTPException, match="401: Invalid token provided"):
+    with pytest.raises(InvalidAudienceError):
         check_jwt(token, 1)
 
 
@@ -43,7 +44,7 @@ def test_jwt_invalid_exp():
         {"id": 1, "exp": 0, "aud": Config.shipping_service.callback_url}, Config.auth.jwt_private, algorithm="ES256"
     )
 
-    with pytest.raises(HTTPException, match="401: Invalid token provided"):
+    with pytest.raises(ExpiredSignatureError):
         check_jwt(token, 1)