Split UserInfo so it can be used in diracx.core

src/diracx/core/models.py

@@ -105,3 +105,10 @@ class SetJobStatusReturn(BaseModel):
     start_exec_time: datetime | None = Field(alias="StartExecTime")
     end_exec_time: datetime | None = Field(alias="EndExecTime")
     last_update_time: datetime | None = Field(alias="LastUpdateTime")
+
+
+class UserInfo(BaseModel):
+    sub: str  # dirac generated vo:sub
+    preferred_username: str
+    dirac_group: str
+    vo: str

src/diracx/routers/auth.py

@@ -33,7 +33,7 @@
     ExpiredFlowError,
     PendingAuthorizationError,
 )
-from diracx.core.models import TokenResponse
+from diracx.core.models import TokenResponse, UserInfo
 from diracx.core.properties import (
     PROXY_MANAGEMENT,
     SecurityProperty,
@@ -82,7 +82,7 @@ def has_properties(expression: UnevaluatedProperty | SecurityProperty):
     )
 
     async def require_property(
-        user: Annotated[UserInfo, Depends(verify_dirac_access_token)]
+        user: Annotated[AuthorizedUserInfo, Depends(verify_dirac_access_token)]
     ):
         if not evaluator(user.properties):
             raise HTTPException(status.HTTP_403_FORBIDDEN)
@@ -164,18 +164,14 @@ class AuthInfo(BaseModel):
     properties: list[SecurityProperty]
 
 
-class UserInfo(AuthInfo):
-    # dirac generated vo:sub
-    sub: str
-    preferred_username: str
-    dirac_group: str
-    vo: str
+class AuthorizedUserInfo(AuthInfo, UserInfo):
+    pass
 
 
 async def verify_dirac_access_token(
     authorization: Annotated[str, Depends(oidc_scheme)],
     settings: AuthSettings,
-) -> UserInfo:
+) -> AuthorizedUserInfo:
     """Verify dirac user token and return a UserInfo class
     Used for each API endpoint
     """
@@ -204,7 +200,7 @@ async def verify_dirac_access_token(
             detail="Invalid JWT",
         ) from None
 
-    return UserInfo(
+    return AuthorizedUserInfo(
         bearer_token=raw_token,
         token_id=token["jti"],
         properties=token["dirac_properties"],
@@ -876,7 +872,7 @@ async def get_oidc_token_info_from_refresh_flow(
 @router.get("/refresh-tokens")
 async def get_refresh_tokens(
     auth_db: AuthDB,
-    user_info: Annotated[UserInfo, Depends(verify_dirac_access_token)],
+    user_info: Annotated[AuthorizedUserInfo, Depends(verify_dirac_access_token)],
 ) -> list:
     subject: str | None = user_info.sub
     if PROXY_MANAGEMENT in user_info.properties:
@@ -889,7 +885,7 @@ async def get_refresh_tokens(
 @router.delete("/refresh-tokens/{jti}")
 async def revoke_refresh_token(
     auth_db: AuthDB,
-    user_info: Annotated[UserInfo, Depends(verify_dirac_access_token)],
+    user_info: Annotated[AuthorizedUserInfo, Depends(verify_dirac_access_token)],
     jti: str,
 ) -> str:
     res = await auth_db.get_refresh_token(jti)
@@ -1006,7 +1002,7 @@ class UserInfoResponse(TypedDict):
 
 @router.get("/userinfo")
 async def userinfo(
-    user_info: Annotated[UserInfo, Depends(verify_dirac_access_token)]
+    user_info: Annotated[AuthorizedUserInfo, Depends(verify_dirac_access_token)]
 ) -> UserInfoResponse:
     return {
         "sub": user_info.sub,

src/diracx/routers/job_manager/__init__.py

@@ -26,7 +26,7 @@
     set_job_status,
 )
 
-from ..auth import UserInfo, has_properties, verify_dirac_access_token
+from ..auth import AuthorizedUserInfo, has_properties, verify_dirac_access_token
 from ..dependencies import JobDB, JobLoggingDB
 from ..fastapi_classes import DiracxRouter
 
@@ -105,7 +105,7 @@ async def submit_bulk_jobs(
     job_definitions: Annotated[list[str], Body(example=EXAMPLE_JDLS["Simple JDL"])],
     job_db: JobDB,
     job_logging_db: JobLoggingDB,
-    user_info: Annotated[UserInfo, Depends(verify_dirac_access_token)],
+    user_info: Annotated[AuthorizedUserInfo, Depends(verify_dirac_access_token)],
 ) -> list[InsertedJob]:
     from DIRAC.Core.Utilities.ClassAd.ClassAdLight import ClassAd
     from DIRAC.Core.Utilities.ReturnValues import returnValueOrRaise
@@ -116,7 +116,7 @@ async def submit_bulk_jobs(
     )
 
     class DiracxJobPolicy(JobPolicy):
-        def __init__(self, user_info: UserInfo, allInfo: bool = True):
+        def __init__(self, user_info: AuthorizedUserInfo, allInfo: bool = True):
             self.userName = user_info.preferred_username
             self.userGroup = user_info.dirac_group
             self.userProperties = user_info.properties
@@ -353,7 +353,7 @@ async def get_job_status_history_bulk(
 async def search(
     config: Annotated[Config, Depends(ConfigSource.create)],
     job_db: JobDB,
-    user_info: Annotated[UserInfo, Depends(verify_dirac_access_token)],
+    user_info: Annotated[AuthorizedUserInfo, Depends(verify_dirac_access_token)],
     page: int = 0,
     per_page: int = 100,
     body: Annotated[
@@ -385,7 +385,7 @@ async def search(
 async def summary(
     config: Annotated[Config, Depends(ConfigSource.create)],
     job_db: JobDB,
-    user_info: Annotated[UserInfo, Depends(verify_dirac_access_token)],
+    user_info: Annotated[AuthorizedUserInfo, Depends(verify_dirac_access_token)],
     body: JobSummaryParams,
 ):
     """Show information suitable for plotting"""