Added functional testing for SCIM authentication

A new CI job has been added to test the OIDC authentication for the
API endpoints. This CI starts Keycloak in a container, installs coldfront
then runs the API functional tests.

A few code decisions to note:
- The functional tests subclassed the keycloak client provided by `CCI-MOC/onboarding-tools`.
Subclassing was needed because the original client class contained code which is out-of-date
with the current version of Keycloak
- The `is_user_superuser()` function in `utils.py` now also checks if `user.is_superuser`,
to account for edge case where a superuser is created but their `is_staff` attribute is
not set.

Author: QuanMPhm

.github/workflows/test-py39-functional-scim-auth.yaml

@@ -0,0 +1,32 @@
+name: test-py39-functional-scim-auth
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+  test-scim-auth:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v3
+
+      - name: Set up python
+        uses: actions/setup-python@v4
+        with:
+          python-version: 3.9
+
+      - name: Run keycloak container
+        run: |
+          ./ci/run_keycloak.sh
+
+      - name: Install Coldfront and plugin
+        run: |
+          ./ci/setup.sh
+
+      - name: Run SCIM authentication functional test
+        run: |
+          ./ci/run_functional_test_scim_auth.sh

ci/run_functional_test_scim_auth.sh

@@ -0,0 +1,20 @@
+# For onbarding-tools
+export KEYCLOAK_USERNAME=admin
+export KEYCLOAK_PASSWORD=nomoresecret
+export KEYCLOAK_URL="http://localhost:8080"
+export OIDC_CLIENT_ID=coldfront
+export OIDC_CLIENT_SECRET=nomoresecret
+export HORIZON_URL="http://foo"
+
+# For coldfront oidc plugin
+export DJANGO_SETTINGS_MODULE="local_settings"
+export PLUGIN_AUTH_OIDC=True
+export OIDC_RP_CLIENT_ID="coldfront"
+export OIDC_RP_CLIENT_SECRET='nomoresecret'
+export OIDC_OP_AUTHORIZATION_ENDPOINT="http://localhost:8080/realms/master/protocol/openid-connect/auth"
+export OIDC_OP_TOKEN_ENDPOINT="http://localhost:8080/realms/master/protocol/openid-connect/token"
+export OIDC_OP_USER_ENDPOINT="http://localhost:8080/realms/master/protocol/openid-connect/userinfo"
+export OIDC_RP_SIGN_ALGO='RS256'
+export OIDC_OP_JWKS_ENDPOINT="http://localhost:8080/realms/master/protocol/openid-connect/certs"
+
+coldfront test coldfront_plugin_api.tests.functional.test_scim_auth

ci/run_keycloak.sh

@@ -0,0 +1,8 @@
+set -xe
+
+sudo docker run -d --name keycloak \
+    -e KEYCLOAK_ADMIN=admin \
+    -e KEYCLOAK_ADMIN_PASSWORD=nomoresecret \
+    -p 8080:8080 \
+    -p 8443:8443 \
+    quay.io/keycloak/keycloak:latest start-dev

src/coldfront_plugin_api/scim_v2/auth_middleware.py

@@ -10,8 +10,9 @@
 class SCIMColdfrontAuthCheckMiddleware(SCIMAuthCheckMiddleware):
     def process_request(self, request):
         if not request.user or not request.user.is_authenticated:
-            # PLUGIN_AUTH_OIDC implies DRF OIDC backend is configured
+            # Check if OIDC authentication is enabled to see if we should authenticate user using OIDC
+            # PLUGIN_AUTH_OIDC implies OIDC is enabled and DRF OIDC backend is configured
             if os.getenv("PLUGIN_AUTH_OIDC") == "True":
-                oidc_auth_obj = OIDCAuthentication()
-                request.user = oidc_auth_obj.authenticate(request)
+                if user_tuple := OIDCAuthentication().authenticate(request):
+                    request.user = user_tuple[0]
         return super().process_request(request)

src/coldfront_plugin_api/tests/unit/base.py renamed to src/coldfront_plugin_api/tests/base.py

@@ -0,0 +1,65 @@
+import os
+
+from django.test import Client
+
+
+from coldfront_plugin_api.tests.functional import utils
+from coldfront_plugin_api.tests.base import TestBase
+
+
+class TestAuthOIDC(TestBase):
+    @classmethod
+    def setUpClass(self):
+        super().setUpClass()
+        self.kc_client = utils.UpdatedKeycloakClient()
+        # Only initialize Coldfront client if running in Github action
+        if os.getenv("CI") == "true":
+            self.kc_client.create_client(
+                "master",
+                "coldfront",
+                "nomoresecret",
+                ["http://foo/signup/oidc_redirect_uri"],
+            )
+            self.kc_client.create_user("master", "[email protected]", "staff", "staff")
+            self.kc_client.create_user("master", "[email protected]", "user", "user")
+
+    def setUp(self):
+        super().setUp()
+        self.staff_user = self.new_user("[email protected]")
+        self.normal_user = self.new_user("[email protected]")
+
+        self.staff_user.is_staff = True
+        self.staff_user.save()
+
+    def test_oidc_authenticated(self):
+        # Test for both staff and normal authenticated users
+        def impersonate_and_get_endpoint(
+            user_to_impersonate, endpoint_url, expected_status_code
+        ):
+            user_token = self.kc_client.impersonate_access_token(user_to_impersonate)
+
+            cf_client = Client()
+            r = cf_client.get(
+                endpoint_url,
+                HTTP_AUTHORIZATION=f"Bearer {user_token}",
+            )
+            self.assertEqual(r.status_code, expected_status_code)
+
+        for endpoint_url in [
+            "/api/scim/v2/Users",
+            "/api/scim/v2/Groups",
+            "/api/allocations",
+        ]:
+            impersonate_and_get_endpoint(self.staff_user.username, endpoint_url, 200)
+            impersonate_and_get_endpoint(self.normal_user.username, endpoint_url, 403)
+
+    def test_oidc_unauthenticated(self):
+        # Test for unauthenticated user case
+        cf_client = Client()
+        for endpoint_url in [
+            "/api/scim/v2/Users",
+            "/api/scim/v2/Groups",
+            "/api/allocations",
+        ]:
+            r = cf_client.get(endpoint_url)
+            self.assertEqual(r.status_code, 401)

src/coldfront_plugin_api/tests/functional/__init__.py

@@ -0,0 +1,41 @@
+import urllib
+import requests
+
+from onboarding_tools.keycloak import KeycloakClient
+from onboarding_tools import settings
+
+
+class UpdatedKeycloakClient(KeycloakClient):
+    @staticmethod
+    def construct_url(realm, path):
+        return f"{settings.KEYCLOAK_URL}/admin/realms/{realm}/{path}"
+
+    @property
+    def url_base(self):
+        return f"{settings.KEYCLOAK_URL}/admin/realms"
+
+    @staticmethod
+    def auth_endpoint(realm):
+        return f"{settings.KEYCLOAK_URL}/realms/{realm}/protocol/openid-connect/auth"
+
+    @staticmethod
+    def token_endpoint(realm):
+        return f"{settings.KEYCLOAK_URL}/realms/{realm}/protocol/openid-connect/token"
+
+    def impersonate_access_token(self, user):
+        user_session = requests.session()
+        user_session.cookies.update(self.impersonate(user).cookies)
+        params = {
+            "response_mode": "fragment",
+            "response_type": "token",
+            "client_id": settings.OIDC_CLIENT_ID,
+            "client_secret": settings.OIDC_CLIENT_SECRET,
+            "redirect_uri": f"{settings.HORIZON_URL}/signup/oidc_redirect_uri",
+            "scope": "openid profile email",
+        }
+        response = user_session.get(
+            self.auth_endpoint("master"), params=params, allow_redirects=False
+        )
+        redirect = response.headers["Location"]
+        token = urllib.parse.parse_qs(redirect)["access_token"][0]
+        return token

src/coldfront_plugin_api/tests/functional/test_scim_auth.py

@@ -6,7 +6,7 @@
     AllocationUserStatusChoice,
 )
 from coldfront.core.project.models import ProjectUser
-from coldfront_plugin_api.tests.unit import base
+from coldfront_plugin_api.tests import base
 
 from unittest.mock import patch
 

src/coldfront_plugin_api/tests/functional/utils.py

@@ -2,9 +2,10 @@
 import uuid
 
 from coldfront.core.resource import models as resource_models
+from coldfront_plugin_api.tests import base
 from rest_framework.test import APIClient
 
-from coldfront_plugin_api.tests.unit import base, fakes
+from coldfront_plugin_api.tests.unit import fakes
 
 
 def get_payload_for_single_operation(operation, username):