114 lines
6.5 KiB
Python
Executable File
114 lines
6.5 KiB
Python
Executable File
import secrets
|
|
import time
|
|
from unittest.mock import patch
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
from authlib.integrations.django_client import OAuth
|
|
from django.conf import settings
|
|
from django.contrib.sessions.backends.db import SessionStore
|
|
from django.test import RequestFactory, TestCase, override_settings
|
|
from joserfc import jwt
|
|
from joserfc.jwk import RSAKey
|
|
|
|
|
|
class AuthlibFlowTests(TestCase):
|
|
"""Exercise real Authlib state, PKCE, nonce and signature validation offline."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super().setUpClass()
|
|
cls.key = RSAKey.generate_key(2048, parameters={"kid": "test-key"})
|
|
|
|
def flow(self, **changes):
|
|
issuer = settings.DEX_ISSUER
|
|
remote = OAuth().register("dex", client_id=settings.DEX_CLIENT_ID, client_secret="test-only", issuer=issuer, authorization_endpoint=issuer + "/auth", token_endpoint=issuer + "/token", jwks_uri=issuer + "/keys", id_token_signing_alg_values_supported=["RS256"], client_kwargs={"scope": "openid email profile federated:id", "code_challenge_method": "S256"})
|
|
request = RequestFactory().get("/members/login/", secure=True)
|
|
request.session = SessionStore()
|
|
nonce = secrets.token_urlsafe(32)
|
|
response = remote.authorize_redirect(request, "https://eom.dev/members/oidc/callback/", nonce=nonce)
|
|
auth_params = parse_qs(urlsplit(response.url).query)
|
|
self.assertEqual(auth_params["code_challenge_method"], ["S256"])
|
|
self.assertTrue(auth_params["code_challenge"][0])
|
|
callback = RequestFactory().get("/members/oidc/callback/", {"code": "test-code", "state": auth_params["state"][0]}, secure=True)
|
|
callback.session = request.session
|
|
claims = {"iss": issuer, "sub": "test-sub", "aud": settings.DEX_CLIENT_ID, "iat": int(time.time()), "exp": int(time.time()) + 300, "nonce": nonce, **changes}
|
|
token = {"id_token": jwt.encode({"alg": "RS256", "kid": "test-key"}, claims, self.key)}
|
|
return remote, callback, token
|
|
|
|
def exchange(self, remote, callback, token):
|
|
with patch.object(remote, "fetch_access_token", return_value=token) as exchange, patch.object(remote, "fetch_jwk_set", return_value={"keys": [self.key.as_dict(private=False)]}):
|
|
result = remote.authorize_access_token(callback, claims_options={"iss": {"essential": True, "value": settings.DEX_ISSUER}}, leeway=0)
|
|
self.assertTrue(exchange.call_args.kwargs["code_verifier"])
|
|
return result
|
|
|
|
def test_real_pkce_nonce_signature_and_one_use_state(self):
|
|
remote, callback, token = self.flow()
|
|
result = self.exchange(remote, callback, token)
|
|
self.assertEqual(result["userinfo"]["sub"], "test-sub")
|
|
with self.assertRaises(Exception):
|
|
self.exchange(remote, callback, token)
|
|
|
|
def test_verified_callback_creates_account_without_ldap(self):
|
|
from members import views
|
|
from members.models import Account
|
|
|
|
remote, callback, token = self.flow(preferred_username="alice", email="alice@example.org", email_verified=True, federated_claims={"connector_id": settings.DEX_CONNECTOR_ID, "user_id": "alice"})
|
|
with patch("members.views.dex", return_value=remote), patch.object(remote, "fetch_access_token", return_value=token), patch.object(remote, "fetch_jwk_set", return_value={"keys": [self.key.as_dict(private=False)]}), patch("members.directory.Connection", side_effect=AssertionError("Login must not bind to LDAP")) as ldap:
|
|
response = views.callback(callback)
|
|
self.assertEqual(response.status_code, 302)
|
|
member = Account.objects.get(oidc_issuer=settings.DEX_ISSUER, oidc_sub="test-sub")
|
|
self.assertIsNone(member.ldap_uuid)
|
|
self.assertEqual(dict(callback.session), {"account_id": str(member.pk)})
|
|
ldap.assert_not_called()
|
|
|
|
def test_wrong_nonce_issuer_audience_expiry_rejected(self):
|
|
for changes in ({"nonce": "attacker"}, {"iss": "https://evil.example"}, {"aud": "another-client"}, {"exp": int(time.time()) - 100}):
|
|
with self.subTest(changes=changes):
|
|
remote, callback, token = self.flow(**changes)
|
|
with self.assertRaises(Exception):
|
|
self.exchange(remote, callback, token)
|
|
|
|
def test_wrong_state_never_exchanges_code(self):
|
|
remote, callback, token = self.flow()
|
|
callback.GET = callback.GET.copy()
|
|
callback.GET["state"] = "unknown-state"
|
|
with patch.object(remote, "fetch_access_token") as exchange:
|
|
with self.assertRaises(Exception):
|
|
remote.authorize_access_token(callback)
|
|
exchange.assert_not_called()
|
|
|
|
def test_bad_signature_rejected(self):
|
|
remote, callback, token = self.flow()
|
|
other = RSAKey.generate_key(2048, parameters={"kid": "test-key"})
|
|
claims = {"iss": settings.DEX_ISSUER, "sub": "attacker", "aud": settings.DEX_CLIENT_ID, "iat": int(time.time()), "exp": int(time.time()) + 300, "nonce": "attacker"}
|
|
token["id_token"] = jwt.encode({"alg": "RS256", "kid": "test-key"}, claims, other)
|
|
with self.assertRaises(Exception):
|
|
self.exchange(remote, callback, token)
|
|
|
|
|
|
class DiscoveryIssuerTests(TestCase):
|
|
def test_settings_preserve_exact_configured_issuer(self):
|
|
import importlib
|
|
import os
|
|
|
|
settings_module = importlib.import_module("config.settings")
|
|
try:
|
|
with patch.dict(os.environ, {"DEX_ISSUER": "https://dex.example.org/"}):
|
|
importlib.reload(settings_module)
|
|
self.assertEqual(settings_module.DEX_ISSUER, "https://dex.example.org/")
|
|
finally:
|
|
importlib.reload(settings_module)
|
|
|
|
@override_settings(DEX_ISSUER="https://dex.example.org/")
|
|
@patch("members.views.OAuth")
|
|
def test_discovery_uses_single_slash_and_exact_issuer(self, oauth):
|
|
from members.views import dex
|
|
|
|
metadata = {"issuer": "https://dex.example.org/", "authorization_endpoint": "https://dex.example.org/auth", "token_endpoint": "https://dex.example.org/token", "jwks_uri": "https://dex.example.org/keys"}
|
|
oauth.return_value.register.return_value.load_server_metadata.return_value = metadata
|
|
dex()
|
|
self.assertEqual(oauth.return_value.register.call_args.kwargs["server_metadata_url"], "https://dex.example.org/.well-known/openid-configuration")
|
|
metadata["issuer"] = "https://dex.example.org"
|
|
with self.assertRaisesRegex(ValueError, "issuer mismatch"):
|
|
dex()
|