feat(api): adds automatic keycloak verification

.devcontainer/devcontainer.json

@@ -51,9 +51,11 @@
         // Make sure SELinux does not disable with access to host filesystems like tmp
         "--security-opt=label=disable"
     ],
-    "forwardPorts": [
-        5173
-    ],
+    "portsAttributes": {
+        "5173": {
+            "onAutoForward": "ignore"
+        }
+    },
     "mounts": [
         // Mount in the user terminal config folder so it can be edited
         {

notes

@@ -0,0 +1,12 @@
+# refresh_token: dict[str, str] = (
+# # pyright: ignore[reportUnknownVariableType]
+#     keycloak_openid.refresh_token(  # pyright: ignore[reportUnknownMemberType]
+#         token["refresh_token"]
+#     )
+# )
+# print(refresh_token)
+
+# save token and refresh token to file
+# if they're present, check if access token is valid, if no but refresh token is
+# , just refresh token
+# if invalid, run whole thing

pyproject.toml

@@ -19,6 +19,7 @@ dependencies = [
     "hera",
     "requests",
     "PyYAML",
+    "python-keycloak",
 ] # Add project dependencies here, e.g. ["click", "numpy"]
 dynamic = ["version"]
 license.file = "LICENSE"

src/python_interface_to_workflows/auth/keycloak_checker.py

@@ -0,0 +1,47 @@
+import os
+
+from keycloak import KeycloakOpenID
+from keycloak.pkce_utils import generate_code_challenge, generate_code_verifier
+
+from python_interface_to_workflows.auth.open_auth_url import open_auth_url
+
+
+def return_key(dev: bool) -> str:
+    match dev:
+        case True:
+            keycloak_openid = KeycloakOpenID(
+                server_url="https://identity-test.diamond.ac.uk/",
+                client_id="workflows-ui-dev",
+                realm_name="dls",
+                client_secret_key="",
+                pool_maxsize=1,
+            )
+        case False:
+            keycloak_openid = KeycloakOpenID(
+                client_id="workflows-dashboard",
+                server_url="https://identity-test.diamond.ac.uk/",
+                realm_name="dls",
+                client_secret_key="",
+                pool_maxsize=1,
+            )
+    code_verifier = generate_code_verifier()
+    code_challenge, code_challenge_method = generate_code_challenge(code_verifier)
+    auth_url = keycloak_openid.auth_url(
+        redirect_uri="http://localhost:5173/",
+        scope="openid posix-uid profile email fedid",
+        state="",
+        code_challenge=code_challenge,
+        code_challenge_method=code_challenge_method,
+    )
+    open_auth_url(auth_url)
+    token: dict[str, str] = (  # pyright: ignore[reportUnknownVariableType]
+        keycloak_openid.token(  # pyright: ignore[reportUnknownMemberType]
+            grant_type="authorization_code",
+            code=os.environ["AUTH"],
+            redirect_uri="http://localhost:5173/",
+            code_verifier=code_verifier,
+        )
+    )
+    os.environ["REFRESH"] = token["refresh_token"]
+    os.environ["TOKEN"] = token["access_token"]
+    return token["access_token"]  # pyright: ignore[reportUnknownArgumentType]

src/python_interface_to_workflows/auth/open_auth_url.py

@@ -0,0 +1,41 @@
+import os
+import socket
+import urllib.parse
+import webbrowser
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from typing import cast
+
+
+class _ReusingHTTPServer(HTTPServer):
+    allow_reuse_address = True
+    auth_code: str
+
+
+class CallbackHandler(BaseHTTPRequestHandler):
+    def do_GET(self):
+        query = urllib.parse.urlparse(self.path).query
+        params = urllib.parse.parse_qs(query)
+        if "code" in params:
+            cast(_ReusingHTTPServer, self.server).auth_code = params["code"][0]
+            self.send_response(200)
+            self.end_headers()
+            self.wfile.write(b"Authorization successful. You can close this window.")
+        else:
+            self.send_response(400)
+            self.end_headers()
+            self.wfile.write(b"Missing authorization code.")
+
+
+def open_auth_url(auth_url: str):
+    httpd = _ReusingHTTPServer(("localhost", 5173), CallbackHandler)
+    webbrowser.open(auth_url)
+    try:
+        httpd.handle_request()
+        os.environ["AUTH"] = httpd.auth_code
+    except OSError:
+        os.environ["AUTH"] = ""
+        print("ERROR: Port in use. Please restart your terminal.")
+        exit(1)
+    finally:
+        httpd.socket.shutdown(socket.SHUT_RDWR)
+        httpd.server_close()

src/python_interface_to_workflows/submit_to_graphql.py

@@ -1,11 +1,14 @@
 from gql import Client, gql
 from gql.transport.aiohttp import AIOHTTPTransport
 
+from python_interface_to_workflows.auth.keycloak_checker import return_key
+
 
 def submit_to_graphql():
+    token = return_key(dev=True)
     transport = AIOHTTPTransport(
         url="https://staging.workflows.diamond.ac.uk/graphql",
-        headers={"Authorization": "Bearer "},  # after Bearer = access token
+        headers={"Authorization": f"Bearer {token}"},
     )
     client = Client(
         transport=transport,

tests/test_keycloak_checker.py

@@ -0,0 +1,39 @@
+import os
+from unittest.mock import MagicMock, patch
+
+from pytest import mark
+
+from python_interface_to_workflows.auth.keycloak_checker import return_key
+
+
+@mark.parametrize("dev", [True, False])
+@patch("python_interface_to_workflows.auth.keycloak_checker.KeycloakOpenID")
+@patch("python_interface_to_workflows.auth.keycloak_checker.generate_code_verifier")
+@patch("python_interface_to_workflows.auth.keycloak_checker.generate_code_challenge")
+@patch("python_interface_to_workflows.auth.keycloak_checker.open_auth_url")
+def test_return_key(
+    mock_open_auth_url: MagicMock,
+    mock_gen_code_challenge: MagicMock,
+    mock_gen_code_verifier: MagicMock,
+    mock_gen_keycloak_id: MagicMock,
+    dev: bool,
+):
+    mock_gen_code_verifier.return_value = "verifier"
+    mock_gen_code_challenge.return_value = ("challenge", "S256")
+    os.environ["AUTH"] = "auth_url_code"
+    keycloak = MagicMock()
+    mock_gen_keycloak_id.return_value = keycloak
+
+    keycloak.auth_url.return_value = "https://mock.site"
+    keycloak.token.return_value = {
+        "access_token": "fake_token",
+        "refresh_token": "fake_refresh",
+    }
+    assert return_key(dev) == "fake_token"
+    mock_open_auth_url.assert_called_once_with("https://mock.site")
+    keycloak.token.assert_called_once_with(
+        grant_type="authorization_code",
+        code="auth_url_code",
+        redirect_uri="http://localhost:5173/",
+        code_verifier="verifier",
+    )