Merge pull request #46 from AustralianBioCommons/token-issuer-fix

AAI-268 fix token verification: need to allow for different issuers

Author: marius-mather

.env.example

@@ -1,8 +1,13 @@
+# Auth0 tenant domain
 AUTH0_DOMAIN=mytenant.auth0.com
 # ID and secret for an app authorized to use the management API
 AUTH0_MANAGEMENT_ID=management-app-id
 AUTH0_MANAGEMENT_SECRET=management-secret
 AUTH0_AUDIENCE=https://audience.com/api
+# Issuer for Auth0 tokens. Note that when you use a custom domain
+#  for login, the issuer will be the custom domain, but the
+#  audience will still be Auth0 tenant domain
+AUTH0_ISSUER=https://customdomain.org/
 # JWT secret key: used to provide some protection around registration
 # Generate with: python -c "import secrets; print(secrets.token_urlsafe(32))"
 JWT_SECRET_KEY=secret-key

.github/workflows/build-ecr.yml

@@ -43,6 +43,7 @@ jobs:
           AUTH0_MANAGEMENT_ID=${{ secrets.AUTH0_MANAGEMENT_ID }}
           AUTH0_MANAGEMENT_SECRET=${{ secrets.AUTH0_MANAGEMENT_SECRET }}
           AUTH0_AUDIENCE=${{ secrets.AUTH0_AUDIENCE }}
+          AUTH0_ISSUER=${{ secrets.AUTH0_ISSUER }}
           JWT_SECRET_KEY=${{ secrets.JWT_SECRET_KEY }}
           ADMIN_ROLES=${{ secrets.ADMIN_ROLES }}
           CORS_ALLOWED_ORIGINS=${{ secrets.CORS_ALLOWED_ORIGINS }}

auth/management.py

@@ -7,6 +7,8 @@
 
 
 def get_management_token(settings: Annotated[Settings, Depends(get_settings)]):
+    # Note: need to call the default auth0 domain here, not the custom
+    #  domain
     url = f"https://{settings.auth0_domain}/oauth/token"
     payload = {
         "grant_type": "client_credentials",

auth/validator.py

@@ -25,12 +25,17 @@ def verify_jwt(token: str, settings: Settings) -> AccessTokenPayload:
         )
 
     try:
+        # Issuer may be the Auth0 tenant domain, or the custom domain
+        #   used for the app. Allow for both
+        issuers = [f"https://{settings.auth0_domain}/"]
+        if settings.auth0_issuer is not None:
+            issuers.append(settings.auth0_issuer)
         payload = jwt.decode(
             token,
             rsa_key,
             algorithms=settings.auth0_algorithms,
             audience=settings.auth0_audience,
-            issuer=f"https://{settings.auth0_domain}/",
+            issuer=issuers,
         )
     except JWTError as e:
         raise HTTPException(status_code=401, detail=f"Invalid token: {e}")

config.py

@@ -1,5 +1,5 @@
 from functools import lru_cache
-from typing import Dict
+from typing import Dict, Optional
 
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
@@ -9,6 +9,9 @@ class Settings(BaseSettings):
     auth0_management_id: str
     auth0_management_secret: str
     auth0_audience: str
+    # Optional: issuer may be different to the auth0_domain if
+    #   a custom domain is used
+    auth0_issuer: Optional[str] = None
     jwt_secret_key: str
     auth0_algorithms: list[str] = ["RS256"]
     admin_roles: list[str] = []

tests/conftest.py

@@ -79,14 +79,15 @@ def mock_settings():
     """Fixture that returns mocked Settings object."""
     return Settings(
         auth0_domain="mock-domain",
+        auth0_issuer=None,
         auth0_management_id="mock-id",
         auth0_management_secret="mock-secret",
         auth0_audience="mock-audience",
         jwt_secret_key="mock-secret-key",
         cors_allowed_origins="https://test",
         send_email=False,
         admin_roles=["Admin"],
-        auth0_algorithms=["HS256"]
+        auth0_algorithms=["RS256"]
     )
 
 

tests/test_auth_validator.py

@@ -1,12 +1,113 @@
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from typing import Optional
 from unittest.mock import patch
 
+import pytest
+
+# Tools from hazmat should only be used for testing!
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.asymmetric.rsa import (
+    RSAPrivateKey,
+    RSAPublicKey,
+)
+from fastapi import HTTPException
 from jose import jwt
 from jose.backends.cryptography_backend import CryptographyRSAKey
 
-from auth.validator import get_rsa_key
+from auth.validator import get_rsa_key, verify_jwt
 from config import Settings
 
 
+def generate_public_private_key_pair():
+    # Code from https://fmpm.dev/mocking-auth0-tokens
+    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    public_key = private_key.public_key()
+    return public_key, private_key
+
+
+@dataclass
+class AuthTokenData:
+    """
+    Stores all the information needed to generate an access token and
+    test that it can be decoded.
+    """
+
+    private_key: RSAPrivateKey
+    public_key: RSAPublicKey
+    access_token_str: str
+    access_token_data: dict
+    key_id: str
+
+
+def create_access_token(
+    email: str = "user@example.com",
+    roles: Optional[list[str]] = None,
+    iss: str = "https://issuer.example.com",
+    sub: Optional[str] = None,
+    iat: Optional[int] = None,
+    exp: Optional[int] = None,
+    aud: str = "https://audience.example.com",
+    scope: Optional[list[str]] = None,
+    azp: Optional[str] = None,
+    permissions: Optional[list[str]] = None,
+    algorithm: str = "RS256",
+    public_key_id: str = "example-key",
+) -> AuthTokenData:
+    """
+    Create an OIDC access token along with a dummy private and public key
+    for signing it. Each field of the payload can be set, but otherwise
+    will get a sensible default (e.g. expiry time in the future).
+    """
+    if roles is None:
+        roles = []
+    # Generate a random alphanumeric ID
+    if sub is None:
+        sub = uuid.uuid4().hex
+    if iat is None:
+        iat = int(datetime.now().strftime("%s"))
+    if exp is None:
+        exp = int((datetime.now() + timedelta(hours=1)).strftime("%s"))
+    if azp is None:
+        azp = uuid.uuid4().hex
+    if permissions is None:
+        permissions = []
+
+    payload = {
+        "email": email,
+        "https://biocommons.org.au/roles": roles,
+        "iss": iss,
+        "sub": sub,
+        "aud": [aud],
+        "iat": iat,
+        "exp": exp,
+        "scope": scope,
+        "azp": azp,
+        "permissions": permissions,
+    }
+    public_key, private_key = generate_public_private_key_pair()
+    from cryptography.hazmat.primitives import serialization
+    pem_private_key = private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption(),
+    )
+    access_token_encoded = jwt.encode(
+        payload,
+        key=pem_private_key,
+        algorithm=algorithm,
+        headers={"kid": public_key_id},
+    )
+    return AuthTokenData(
+        private_key=private_key,
+        public_key=public_key,
+        access_token_str=access_token_encoded,
+        access_token_data=payload,
+        key_id=public_key_id,
+    )
+
+
 def test_get_rsa_key_returns_key(mock_settings: Settings):
     token = jwt.encode({"some": "payload"}, "secret", algorithm="HS256")
     unverified_header = {"kid": "testkey"}
@@ -27,3 +128,50 @@ def test_get_rsa_key_returns_key(mock_settings: Settings):
         key = get_rsa_key(token, settings=mock_settings)
         assert key is not None
         assert isinstance(key, CryptographyRSAKey)
+
+
+def test_verify_jwt(mock_settings: Settings, mocker):
+    """
+    Test we can verify a JWT based on issuer and audience.
+    """
+    mock_settings.auth0_audience = f"https://{mock_settings.auth0_domain}/api/"
+    token = create_access_token(
+        email="user@example.com",
+        # Our verify code assumes the issuer will be the auth0 domain
+        iss=f"https://{mock_settings.auth0_domain}/",
+        aud=f"https://{mock_settings.auth0_domain}/api/",
+    )
+    mocker.patch("auth.validator.get_rsa_key", return_value=token.public_key)
+    decoded = verify_jwt(token.access_token_str, settings=mock_settings)
+    assert decoded.email == "user@example.com"
+
+
+def test_verify_jwt_invalid_issuer(mock_settings: Settings, mocker):
+    """
+    Test invalid JWT issuer raises an error
+    """
+    mock_settings.auth0_audience = f"https://{mock_settings.auth0_domain}/api/"
+    token = create_access_token(
+        email="user@example.com",
+        iss="https://other.example.com/",
+        aud=f"https://{mock_settings.auth0_domain}/api/",
+    )
+    mocker.patch("auth.validator.get_rsa_key", return_value=token.public_key)
+    with pytest.raises(HTTPException, match="Invalid issuer"):
+        verify_jwt(token.access_token_str, settings=mock_settings)
+
+
+def test_verify_jwt_custom_domain_issuer(mock_settings: Settings, mocker):
+    """
+    Check that our verify code also works with the auth0_issuer setting
+    """
+    mock_settings.auth0_audience = f"https://{mock_settings.auth0_domain}/api/"
+    mock_settings.auth0_issuer = "https://mydomain.org/"
+    token = create_access_token(
+        email="user@example.com",
+        iss=mock_settings.auth0_issuer,
+        aud=mock_settings.auth0_audience,
+    )
+    mocker.patch("auth.validator.get_rsa_key", return_value=token.public_key)
+    decoded = verify_jwt(token.access_token_str, settings=mock_settings)
+    assert decoded.email == "user@example.com"

tests/test_galaxy.py

@@ -44,10 +44,11 @@ def test_get_registration_token(test_client, mock_settings):
     """
     Test get-registration-token endpoint returns a valid JWT token.
     """
+    from register.tokens import ALGORITHM
     response = test_client.get("/galaxy/register/get-registration-token")
     assert response.status_code == 200
     jwt.decode(response.json()["token"], mock_settings.jwt_secret_key,
-               algorithms=mock_settings.auth0_algorithms)
+               algorithms=ALGORITHM)
 
 
 def test_registration_token_invalid_purpose(mock_settings):