feat: add a few more tests

diracx-routers/src/diracx/routers/auth/management.py

@@ -47,8 +47,7 @@ async def get_refresh_tokens(
     if PROXY_MANAGEMENT in user_info.properties:
         subject = None
 
-    res = await auth_db.get_user_refresh_tokens(subject)
-    return res
+    return await auth_db.get_user_refresh_tokens(subject)
 
 
 @router.delete("/refresh-tokens/{jti}")

diracx-routers/src/diracx/routers/auth/token.py

@@ -84,7 +84,7 @@ async def token(
 
     elif grant_type == GrantType.authorization_code:
         oidc_token_info, scope = await get_oidc_token_info_from_authorization_flow(
-            code, redirect_uri, code_verifier, auth_db, settings
+            code, client_id, redirect_uri, code_verifier, auth_db, settings
         )
 
     elif grant_type == GrantType.refresh_token:
@@ -148,6 +148,7 @@ async def get_oidc_token_info_from_device_flow(
 
 async def get_oidc_token_info_from_authorization_flow(
     code: str | None,
+    client_id: str | None,
     redirect_uri: str | None,
     code_verifier: str | None,
     auth_db: AuthDB,
@@ -163,7 +164,13 @@ async def get_oidc_token_info_from_authorization_flow(
             status_code=status.HTTP_400_BAD_REQUEST,
             detail="Invalid redirect_uri",
         )
+    if client_id != info["client_id"]:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Bad client_id",
+        )
 
+    # Check the code_verifier
     try:
         assert code_verifier is not None
         code_challenge = (

diracx-routers/tests/auth/test_standard.py

@@ -115,6 +115,41 @@ async def test_authorization_flow(test_client, auth_httpx_mock: HTTPXMock):
         .replace("=", "")
     )
 
+    # Initiate the authorization flow with a wrong client ID
+    # Check that the client ID is not recognized
+    r = test_client.get(
+        "/api/auth/authorize",
+        params={
+            "response_type": "code",
+            "code_challenge": code_challenge,
+            "code_challenge_method": "S256",
+            "client_id": "Unknown client ID",
+            "redirect_uri": "http://diracx.test.invalid:8000/api/docs/oauth2-redirect",
+            "scope": "vo:lhcb property:NormalUser",
+            "state": "external-state",
+        },
+        follow_redirects=False,
+    )
+    assert r.status_code == 400, r.text
+
+    # Initiate the authorization flow with an unrecognized redirect URI
+    # Check that the redirect URI is not recognized
+    r = test_client.get(
+        "/api/auth/authorize",
+        params={
+            "response_type": "code",
+            "code_challenge": code_challenge,
+            "code_challenge_method": "S256",
+            "client_id": DIRAC_CLIENT_ID,
+            "redirect_uri": "http://diracx.test.unrecognized:8000/api/docs/oauth2-redirect",
+            "scope": "vo:lhcb property:NormalUser",
+            "state": "external-state",
+        },
+        follow_redirects=False,
+    )
+    assert r.status_code == 400, r.text
+
+    # Correctly initiate the authorization flow
     r = test_client.get(
         "/api/auth/authorize",
         params={
@@ -157,6 +192,30 @@ async def test_authorization_flow(test_client, auth_httpx_mock: HTTPXMock):
     assert query_parameters["state"][0] == "external-state"
     code = query_parameters["code"][0]
 
+    # Try to get token with the wrong client ID
+    request_data = {
+        "grant_type": "authorization_code",
+        "code": code,
+        "state": state,
+        "client_id": "Unknown client ID",
+        "redirect_uri": "http://diracx.test.invalid:8000/api/docs/oauth2-redirect",
+        "code_verifier": code_verifier,
+    }
+    r = test_client.post("/api/auth/token", data=request_data)
+    assert r.status_code == 400, r.json()
+
+    # Try to get token with the wrong redirect URI
+    request_data = {
+        "grant_type": "authorization_code",
+        "code": code,
+        "state": state,
+        "client_id": "Unknown client ID",
+        "redirect_uri": "http://diracx.test.unrecognized:8000/api/docs/oauth2-redirect",
+        "code_verifier": code_verifier,
+    }
+    r = test_client.post("/api/auth/token", data=request_data)
+    assert r.status_code == 400, r.json()
+
     # Get and check token
     request_data = {
         "grant_type": "authorization_code",
@@ -178,6 +237,17 @@ async def test_authorization_flow(test_client, auth_httpx_mock: HTTPXMock):
 
 
 async def test_device_flow(test_client, auth_httpx_mock: HTTPXMock):
+    # Initiate the device flow with a wrong client ID
+    # Check that the client ID is not recognized
+    r = test_client.post(
+        "/api/auth/device",
+        params={
+            "client_id": "Unknown client ID",
+            "scope": "vo:lhcb property:NormalUser",
+        },
+    )
+    assert r.status_code == 400, r.json()
+
     # Initiate the device flow (would normally be done from CLI)
     r = test_client.post(
         "/api/auth/device",
@@ -235,6 +305,15 @@ async def test_device_flow(test_client, auth_httpx_mock: HTTPXMock):
     r = test_client.get(redirect_uri, params={"code": "valid-code", "state": state})
     assert r.status_code == 400, r.text
 
+    # Try to get token with the wrong redirect URI
+    request_data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
+        "device_code": data["device_code"],
+        "client_id": "Unknown client ID",
+    }
+    r = test_client.post("/api/auth/token", data=request_data)
+    assert r.status_code == 400, r.json()
+
     # Get and check token
     request_data = {
         "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
@@ -737,12 +816,30 @@ def test_parse_scopes(vos, groups, scope, expected):
 @pytest.mark.parametrize(
     "vos, groups, scope, expected_error",
     [
+        [
+            ["lhcb"],
+            ["lhcb_user"],
+            "group:lhcb_user undefinedscope:undefined",
+            "Unrecognised scopes",
+        ],
         [
             ["lhcb"],
             ["lhcb_user"],
             "group:lhcb_user",
             "No vo scope requested",
         ],
+        [
+            ["lhcb", "gridpp"],
+            ["lhcb_user", "lhcb_admin"],
+            "vo:lhcb vo:gridpp group:lhcb_user group:lhcb_admin",
+            "Only one vo is allowed",
+        ],
+        [
+            ["lhcb"],
+            ["lhcb_user", "lhcb_admin"],
+            "vo:lhcb group:lhcb_user group:lhcb_admin",
+            "Only one DIRAC group allowed",
+        ],
     ],
 )
 def test_parse_scopes_invalid(vos, groups, scope, expected_error):