From 79c0c6711671ef3d69d9d86b42abd6be2a2e6650 Mon Sep 17 00:00:00 2001 From: nuwang <2070605+nuwang@users.noreply.github.com> Date: Fri, 3 Nov 2023 22:26:15 +0530 Subject: [PATCH] Add test for unauthorized audience --- lib/galaxy/authnz/managers.py | 5 +---- test/integration/oidc/test_auth_oidc.py | 7 +++++++ 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/lib/galaxy/authnz/managers.py b/lib/galaxy/authnz/managers.py index 41e82359d9cf..5abf55e47e6c 100644 --- a/lib/galaxy/authnz/managers.py +++ b/lib/galaxy/authnz/managers.py @@ -421,14 +421,11 @@ def find_user_by_access_token_in_provider(self, sa_session, provider, access_tok return user return None except Exception as e: - msg = f"An error occurred when finding user by token: {e}" + msg = f"An error occurred with provider: {provider} when finding user by token: {e}" log.error(msg) return None def find_user_by_access_token(self, sa_session, access_token): - # decoded_token = jwt.decode(access_token, options={"verify_signature": False}) - # issuer = decoded_token["iss"] - # audience = decoded_token["aud"] for provider in self.oidc_backends_config: user = self.find_user_by_access_token_in_provider(sa_session, provider, access_token) if user: diff --git a/test/integration/oidc/test_auth_oidc.py b/test/integration/oidc/test_auth_oidc.py index a1c6ffeec93a..e4fc398dde23 100644 --- a/test/integration/oidc/test_auth_oidc.py +++ b/test/integration/oidc/test_auth_oidc.py @@ -242,6 +242,13 @@ def test_auth_with_another_authorized_client(self): access_token = self._get_keycloak_access_token(client_id="bpaclient", scopes=["gx:*"]) response = self._get("users/current", headers={"Authorization": f"Bearer {access_token}"}) self._assert_status_code_is(response, 200) + assert response.json()["email"] == "gxyuser@galaxy.org" + + def test_auth_with_authorized_client_but_unauthorized_audience(self): + _, response = self._login_via_keycloak("bpaonlyuser", KEYCLOAK_TEST_PASSWORD) + access_token = self._get_keycloak_access_token(client_id="bpaclient", username="bpaonlyuser") + response = self._get("users/current", headers={"Authorization": f"Bearer {access_token}"}) + self._assert_status_code_is(response, 400) def test_auth_with_unauthorized_client(self): _, response = self._login_via_keycloak(KEYCLOAK_TEST_USERNAME, KEYCLOAK_TEST_PASSWORD)