[WIP] OpenID connect client library - eyeing toward deploying keycloak

.gitignore

@@ -128,4 +128,6 @@ dist/
 .sass-cache/
 .pytest_cache/
 
-fastly_hourly_stats.ini
+fastly_hourly_stats.ini
+
+*.db-journal

arxiv/auth/openid/oidc_idp.py

@@ -0,0 +1,91 @@
+"""
+OpenID connect IdP client
+"""
+from typing import List
+import requests
+import jwt
+from jwt.algorithms import RSAAlgorithm, RSAPublicKey
+
+
+class ArxivOidcIdpClient:
+    """arXiv OpenID Connect IdP client
+    This is implemented for Keycloak. If the APIs different, you may want to subclass/adjust.
+    """
+
+    server_url: str
+    client_id: str
+    client_secret: str | None
+    realm: str
+    redirect_uri: str  #
+    scope: List[str]  # it's okay to be empty. Keycloak should be configured to provide scopes.
+    _server_certs: dict  # Cache for the IdP certs
+
+    def __init__(self, redirect_uri: str,
+                 server_url: str = "https://openid.arxiv.org",
+                 realm: str ="arxiv",
+                 client_id: str = "arxiv-user",
+                 scope: List[str] = [],
+                 client_secret: str | None = None):
+        self.server_url = server_url
+        self.realm = realm
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.redirect_uri = redirect_uri
+        self.scope = scope if scope else ["openid", "profile", "email"]
+        self._server_certs = {}
+        pass
+
+    @property
+    def oidc(self) -> str:
+        return f'{self.server_url}/realms/{self.realm}/protocol/openid-connect'
+
+
+    @property
+    def auth_url(self) -> str:
+        return self.oidc + '/auth'
+
+    @property
+    def token_url(self) -> str:
+        return self.oidc + '/token'
+
+    @property
+    def certs_url(self) -> str:
+        return self.oidc + '/certs'
+
+    @property
+    def login_url(self) -> str:
+        scope = "&" + ",".join(self.scope) if self.scope else ""
+        return f'{self.auth_url}?client_id={self.client_id}&redirect_uri={self.redirect_uri}&response_type=code{scope}'
+
+
+    @property
+    def server_certs(self) -> dict:
+        if not self._server_certs:
+            certs_response = requests.get(self.certs_url)
+            self._server_certs = certs_response.json()
+        return self._server_certs
+
+
+    def get_public_key(self, kid: str) -> RSAPublicKey | None:
+        for key in self.server_certs['keys']:
+            if key['kid'] == kid:
+                pkey = RSAAlgorithm.from_jwk(key)
+                if isinstance(pkey, RSAPublicKey):
+                    return pkey
+        return None
+
+    def validate_token(self, token: str) -> dict | None:
+        try:
+            unverified_header = jwt.get_unverified_header(token)
+            kid = unverified_header['kid'] # key id
+            algorithm = unverified_header['alg'] # key algo
+            public_key = self.get_public_key(kid)
+            if public_key is None:
+                return None
+            decoded_token: dict = jwt.decode(token, public_key, algorithms=[algorithm])
+            return dict(decoded_token)
+        except jwt.ExpiredSignatureError:
+            return None
+        except jwt.InvalidTokenError:
+            return None
+        # not reached

arxiv/auth/openid/tests/test_keycloak.py

@@ -0,0 +1,44 @@
+import subprocess
+import pytest
+from selenium import webdriver
+from selenium.webdriver.common.by import By
+import time
+
+@pytest.fixture(scope="module")
+def web_driver() -> webdriver.Chrome:
+    # Set up the Selenium WebDriver
+    _web_driver = webdriver.Chrome()  # Ensure you have ChromeDriver installed and in PATH
+    _web_driver.implicitly_wait(10)  # Wait for elements to be ready
+    yield _web_driver
+    _web_driver.quit()  # Close the browser window after tests
+
+@pytest.fixture(scope="module")
+def toy_flask():
+    flask_app = subprocess.Popen(["python3", "-m", "arxiv.auth.openid.tests.toy_flask"])
+    time.sleep(5)  # Give the server time to start
+    yield flask_app
+    # Stop the Flask app
+    flask_app.terminate()
+    flask_app.wait()
+
+def test_login(web_driver, toy_flask):
+    web_driver.get("http://localhost:5000/login")  # URL of your Flask app's login route
+
+    # Simulate user login on the IdP login page
+    # Replace the following selectors with the actual ones from your IdP login form
+    username_field = web_driver.find_element(By.ID, "username")  # Example selector
+    password_field = web_driver.find_element(By.ID, "password")  # Example selector
+    login_button = web_driver.find_element(By.ID, "kc-login")    # Example selector
+
+    # Enter credentials
+    username_field.send_keys("testuser")
+    password_field.send_keys("testpassword")
+    login_button.click()
+
+    # Wait for the redirect back to the Flask app
+    time.sleep(5)
+
+    # Check if the login was successful by verifying the presence of a specific element or text
+    web_driver.get("http://localhost:5000/protected")  # URL of your protected route
+    body_text = web_driver.find_element(By.TAG_NAME, "body").text
+    assert "Token is valid" in body_text

arxiv/auth/openid/tests/toy_flask.py

@@ -0,0 +1,91 @@
+import os
+
+from flask import Flask, redirect, request, session, url_for, jsonify
+from ..oidc_idp import ArxivOidcIdpClient
+
+import requests
+
+KEYCLOAK_SERVER_URL = os.environ.get('KEYCLOAK_SERVER_URL', 'localhost')
+#REALM_NAME = 'arxiv'
+#CLIENT_ID = 'arxiv-user'
+#CLIENT_SECRET = 'your-client-secret'
+#REDIRECT_URI = 'http://localhost:5000/callback'
+
+class ToyFlask(Flask):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.secret_key = 'secret'  # Replace with a secure secret key
+        self.idp = ArxivOidcIdpClient("http://localhost:5000/callback",
+                                      server_url=KEYCLOAK_SERVER_URL)
+
+app = ToyFlask(__name__)
+
+
+@app.route('/')
+def home():
+    session.clear()
+    return redirect('/login')
+
+@app.route('/login')
+def login():
+    return redirect(app.idp.login_url)
+
+
+@app.route('/callback')
+def callback():
+    # Get the authorization code from the callback URL
+    code = request.args.get('code')
+
+    # Exchange the authorization code for an access token
+    token_response = requests.post(
+        app.idp.token_url,
+        data={
+            'grant_type': 'authorization_code',
+            'code': code,
+            'redirect_uri': app.idp.redirect_uri,
+            'client_id': app.idp.client_id,
+        }
+    )
+
+    if token_response.status_code != 200:
+        session.clear()
+        return 'Something is wrong'
+
+    # Parse the token response
+    token_json = token_response.json()
+    access_token = token_json.get('access_token')
+    refresh_token = token_json.get('refresh_token')
+
+    # Store tokens in session (for demonstration purposes)
+    session['access_token'] = access_token
+    session['refresh_token'] = refresh_token
+
+    print(app.idp.validate_token(access_token))
+    return 'Login successful!'
+
+
+@app.route('/logout')
+def logout():
+    # Clear the session and redirect to home
+    session.clear()
+    return redirect(url_for('home'))
+
+
+@app.route('/protected')
+def protected():
+    access_token = session.get('access_token')
+    if not access_token:
+        return redirect(app.idp.login_url)
+
+    decoded_token = app.idp.validate_token(access_token)
+    if not decoded_token:
+        return jsonify({'error': 'Invalid token'}), 401
+
+    return jsonify({'message': 'Token is valid', 'token': decoded_token})
+
+
+def create_app():
+    return app
+
+if __name__ == '__main__':
+    app.run(debug=True, host='0.0.0.0', port=5000)