Login
This commit is contained in:
@@ -5,11 +5,14 @@ import re
|
||||
import time
|
||||
from contextlib import contextmanager
|
||||
from datetime import timedelta
|
||||
from io import StringIO
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from django.conf import settings
|
||||
from django.core import mail
|
||||
from django.test import Client, SimpleTestCase, TestCase, override_settings
|
||||
from django.core.management import call_command
|
||||
from django.core.management.base import CommandError
|
||||
from django.test import Client, TestCase, TransactionTestCase, override_settings
|
||||
from django.utils import timezone
|
||||
from ldap3 import Connection, MOCK_SYNC, OFFLINE_SLAPD_2_4, Server
|
||||
|
||||
@@ -28,6 +31,8 @@ TIERS = {
|
||||
|
||||
|
||||
def account(**kwargs):
|
||||
if kwargs.get("oidc_sub"):
|
||||
kwargs.setdefault("oidc_issuer", settings.DEX_ISSUER)
|
||||
return Account.objects.create(uid="alice", email="alice@example.org", ldap_dn=DN, ldap_uuid="11111111-1111-1111-1111-111111111111", **kwargs)
|
||||
|
||||
|
||||
@@ -70,6 +75,7 @@ class RegistrationTests(TestCase):
|
||||
self.assertFalse(Subscription.objects.exists())
|
||||
self.assertFalse(Registration.objects.exists())
|
||||
self.assertNotIn("account_id", self.client.session)
|
||||
self.assertEqual(self.client.session["registration_identity"]["account_id"], str(Account.objects.get().pk))
|
||||
self.assertNotIn(password, str(list(Account.objects.values())))
|
||||
self.client.post("/members/register/finish/", {"password": password, "confirm_password": password})
|
||||
self.assertEqual(provision.call_count, 1)
|
||||
@@ -134,46 +140,118 @@ class IdentityTests(TestCase):
|
||||
def claims(self, **changes):
|
||||
return {"iss": settings.DEX_ISSUER, "sub": "dex-alice", "preferred_username": "alice", "email_verified": True, "email": "alice@example.org", "federated_claims": {"connector_id": settings.DEX_CONNECTOR_ID, "user_id": "alice"}, **changes}
|
||||
|
||||
@patch("members.views.directory.identity_for_login")
|
||||
def test_existing_directory_account_links_without_password(self, identity):
|
||||
identity.return_value = {"uid": "alice", "email": "alice@example.org", "ldap_dn": DN, "ldap_uuid": "uuid-alice"}
|
||||
@patch("members.directory.connection", side_effect=AssertionError("Login must not contact LDAP"))
|
||||
def test_first_and_returning_dex_login_need_no_ldap(self, connection):
|
||||
first = views.link_identity(self.claims())
|
||||
second = views.link_identity(self.claims(sub="ldap-alice"))
|
||||
second = views.link_identity(self.claims(email="updated@example.org"))
|
||||
self.assertEqual(first.pk, second.pk)
|
||||
first.refresh_from_db()
|
||||
self.assertEqual(first.oidc_sub, "ldap-alice")
|
||||
second.refresh_from_db()
|
||||
self.assertEqual(second.oidc_sub, "dex-alice")
|
||||
self.assertEqual(second.oidc_issuer, settings.DEX_ISSUER)
|
||||
self.assertEqual(second.email, "updated@example.org")
|
||||
self.assertIsNone(second.ldap_uuid)
|
||||
self.assertEqual(second.ldap_dn, "")
|
||||
self.assertEqual(second.membership_level, 0)
|
||||
self.assertEqual(ManagedGroup.objects.filter(account=second).count(), len(TIERS))
|
||||
self.assertFalse(Subscription.objects.exists())
|
||||
views.link_identity(self.claims(sub="next-verified-subject"))
|
||||
first.refresh_from_db()
|
||||
self.assertEqual(first.oidc_sub, "next-verified-subject")
|
||||
connection.assert_not_called()
|
||||
|
||||
@patch("members.views.directory.identity_for_login")
|
||||
def test_no_email_only_or_unverified_link(self, identity):
|
||||
for changes in ({"email_verified": False}, {"iss": "https://evil.example"}, {"sub": ""}):
|
||||
with self.assertRaises(ValueError):
|
||||
def test_invalid_claims_rejected(self):
|
||||
from django.core.exceptions import ValidationError
|
||||
|
||||
for changes in ({"email_verified": False}, {"iss": "https://evil.example"}, {"sub": ""}, {"sub": "x" * 513}, {"preferred_username": "*)(uid=*)"}, {"email": None}, {"email": "invalid"}):
|
||||
with self.subTest(changes=changes), self.assertRaises((ValueError, ValidationError)):
|
||||
views.link_identity(self.claims(**changes))
|
||||
identity.assert_not_called()
|
||||
identity.return_value = {"uid": "alice", "email": "other@example.org", "ldap_dn": DN, "ldap_uuid": "uuid-alice"}
|
||||
with self.assertRaises(ValueError):
|
||||
views.link_identity(self.claims())
|
||||
self.assertFalse(Account.objects.exists())
|
||||
|
||||
@patch("members.views.directory.identity_for_login")
|
||||
def test_other_connector_or_recreated_ldap_account_rejected(self, identity):
|
||||
identity.return_value = {"uid": "alice", "email": "alice@example.org", "ldap_dn": DN, "ldap_uuid": "uuid-alice"}
|
||||
for federated in ({"connector_id": "github", "user_id": "alice"}, {"connector_id": settings.DEX_CONNECTOR_ID, "user_id": "different-uid"}, {}):
|
||||
def test_wrong_or_missing_signed_connector_identity_rejected(self):
|
||||
for federated in ({"connector_id": "github", "user_id": "alice"}, {"connector_id": settings.DEX_CONNECTOR_ID, "user_id": "different-uid"}, {}, ["malformed"]):
|
||||
with self.assertRaises(ValueError):
|
||||
views.link_identity(self.claims(federated_claims=federated))
|
||||
self.assertFalse(Account.objects.exists())
|
||||
|
||||
@patch("members.views.directory.identity_for_login")
|
||||
def test_same_uid_different_uuid_cannot_take_over_account(self, identity):
|
||||
def test_existing_subject_and_directory_pin_preserved(self):
|
||||
original = account(oidc_sub="dex-alice", stripe_customer="cus_existing")
|
||||
linked = views.link_identity(self.claims())
|
||||
self.assertEqual(linked.pk, original.pk)
|
||||
self.assertEqual(linked.ldap_uuid, original.ldap_uuid)
|
||||
self.assertEqual(linked.ldap_dn, original.ldap_dn)
|
||||
self.assertEqual(linked.stripe_customer, "cus_existing")
|
||||
|
||||
def test_different_subject_cannot_take_over_existing_account(self):
|
||||
original = account(oidc_sub="original")
|
||||
identity.return_value = {"uid": "alice", "email": "alice@example.org", "ldap_dn": DN, "ldap_uuid": "new-directory-entry"}
|
||||
with self.assertRaises(ValueError):
|
||||
views.link_identity(self.claims(), registration_account_id=str(original.pk))
|
||||
original.refresh_from_db()
|
||||
self.assertEqual(original.oidc_sub, "original")
|
||||
|
||||
def test_issuer_change_does_not_reuse_subject(self):
|
||||
original = account(oidc_sub="dex-alice", oidc_issuer="https://previous.example/")
|
||||
with self.assertRaises(ValueError):
|
||||
views.link_identity(self.claims())
|
||||
original.refresh_from_db()
|
||||
self.assertEqual(original.oidc_sub, "original")
|
||||
self.assertEqual(original.oidc_issuer, "https://previous.example/")
|
||||
|
||||
def test_email_match_does_not_link_different_identity(self):
|
||||
original = account(oidc_sub="original")
|
||||
other = views.link_identity(self.claims(sub="dex-bob", preferred_username="bob", federated_claims={"connector_id": settings.DEX_CONNECTOR_ID, "user_id": "bob"}))
|
||||
self.assertNotEqual(other.pk, original.pk)
|
||||
self.assertIsNone(other.stripe_customer)
|
||||
|
||||
def test_subject_cannot_be_reused_with_another_uid(self):
|
||||
account(oidc_sub="dex-alice")
|
||||
with self.assertRaises(ValueError):
|
||||
views.link_identity(self.claims(preferred_username="bob", federated_claims={"connector_id": settings.DEX_CONNECTOR_ID, "user_id": "bob"}))
|
||||
self.assertEqual(Account.objects.count(), 1)
|
||||
|
||||
def test_unbound_account_requires_registration_proof(self):
|
||||
original = account()
|
||||
for proof in (None, "another-account"):
|
||||
with self.assertRaises(ValueError):
|
||||
views.link_identity(self.claims(), registration_account_id=proof)
|
||||
linked = views.link_identity(self.claims(), registration_account_id=str(original.pk))
|
||||
self.assertEqual(linked.pk, original.pk)
|
||||
self.assertEqual(linked.ldap_uuid, original.ldap_uuid)
|
||||
self.assertEqual(linked.oidc_sub, "dex-alice")
|
||||
|
||||
@patch("members.views.dex")
|
||||
def test_callback_consumes_registration_proof_and_rejects_expired_proof(self, dex):
|
||||
original = account()
|
||||
dex.return_value.authorize_access_token.return_value = {"id_token": "private-token", "userinfo": self.claims()}
|
||||
session = self.client.session
|
||||
session["registration_identity"] = {"account_id": str(original.pk), "expires_at": time.time() - 10}
|
||||
session.save()
|
||||
self.assertContains(self.client.get("/members/oidc/callback/"), "could not be verified")
|
||||
original.refresh_from_db()
|
||||
self.assertIsNone(original.oidc_sub)
|
||||
session = self.client.session
|
||||
session["registration_identity"] = {"account_id": str(original.pk), "expires_at": time.time() + 300}
|
||||
session.save()
|
||||
self.assertRedirects(self.client.get("/members/oidc/callback/"), "/members/")
|
||||
self.assertEqual(dict(self.client.session), {"account_id": str(original.pk)})
|
||||
|
||||
def test_operator_binding_is_explicit_idempotent_and_cannot_rebind(self):
|
||||
original = account()
|
||||
options = {"account": str(original.pk), "uid": "alice", "subject": "dex-alice", "stdout": StringIO()}
|
||||
with self.assertRaises(CommandError):
|
||||
call_command("bind_dex_identity", **{**options, "uid": "bob"})
|
||||
call_command("bind_dex_identity", **options)
|
||||
call_command("bind_dex_identity", **options)
|
||||
self.assertEqual(views.link_identity(self.claims()).pk, original.pk)
|
||||
with self.assertRaises(CommandError):
|
||||
call_command("bind_dex_identity", **{**options, "subject": "different-sub"})
|
||||
|
||||
def test_operator_cannot_bind_subject_owned_by_another_account(self):
|
||||
original = account()
|
||||
Account.objects.create(uid="bob", email="bob@example.org", oidc_sub="dex-alice", oidc_issuer=settings.DEX_ISSUER)
|
||||
with self.assertRaises(CommandError):
|
||||
call_command("bind_dex_identity", account=str(original.pk), uid="alice", subject="dex-alice", stdout=StringIO())
|
||||
original.refresh_from_db()
|
||||
self.assertIsNone(original.oidc_sub)
|
||||
|
||||
def test_operator_binding_rejects_invalid_account_ids(self):
|
||||
with self.assertRaises(CommandError):
|
||||
call_command("bind_dex_identity", account="not-a-uuid", uid="alice", subject="dex-alice", stdout=StringIO())
|
||||
|
||||
@patch("members.views.dex")
|
||||
@patch("members.views.link_identity")
|
||||
@@ -196,6 +274,18 @@ class IdentityTests(TestCase):
|
||||
self.assertContains(response, "could not be verified")
|
||||
self.assertNotIn("account_id", self.client.session)
|
||||
|
||||
@patch("members.views.dex")
|
||||
@patch("members.directory.Connection")
|
||||
@override_settings(LDAP_URL="", LDAP_BIND_DN="", LDAP_BIND_PASSWORD="", LDAP_PASSWORD_SCHEME="{CRYPT}")
|
||||
def test_callback_works_without_ldap_configuration(self, conn_factory, dex):
|
||||
from ldap3.core.exceptions import LDAPInvalidCredentialsResult
|
||||
|
||||
dex.return_value.authorize_access_token.return_value = {"id_token": "private-token", "userinfo": self.claims()}
|
||||
conn_factory.return_value.bind.side_effect = LDAPInvalidCredentialsResult(description="private LDAP diagnostic")
|
||||
self.assertRedirects(self.client.get("/members/oidc/callback/?code=private-code"), "/members/")
|
||||
self.assertEqual(dict(self.client.session), {"account_id": str(Account.objects.get().pk)})
|
||||
conn_factory.assert_not_called()
|
||||
|
||||
|
||||
@override_settings(MEMBERSHIP_TIERS=TIERS, STRIPE_WEBHOOK_SECRET="whsec_test_only", STRIPE_PORTAL_CONFIGURATION="bpc_test")
|
||||
class BillingViewsTests(TestCase):
|
||||
@@ -438,7 +528,7 @@ class ReconciliationTests(TestCase):
|
||||
|
||||
|
||||
@override_settings(LDAP_SEARCH_BASE_DN=BASE, LDAP_GROUP_PLACEHOLDER_DN=SENTINEL)
|
||||
class DirectoryTests(SimpleTestCase):
|
||||
class DirectoryTests(TestCase):
|
||||
def setUp(self):
|
||||
# ldap3 supports offline schemas; its third-party annotation omits them.
|
||||
server = Server("mock", get_info=OFFLINE_SLAPD_2_4) # pyright: ignore[reportArgumentType]
|
||||
@@ -476,9 +566,54 @@ class DirectoryTests(SimpleTestCase):
|
||||
with self.assertRaises(ValueError):
|
||||
directory.reconcile_groups(self.member, {GROUP}, {GROUP})
|
||||
|
||||
def test_login_uid_rejects_filter_injection(self):
|
||||
with self.assertRaises(ValueError):
|
||||
directory.identity_for_login("*)(uid=*)")
|
||||
def test_worker_resolves_and_pins_first_time_dex_identity(self):
|
||||
self.member.ldap_uuid = None
|
||||
self.member.ldap_dn = ""
|
||||
self.member.oidc_issuer = settings.DEX_ISSUER
|
||||
self.member.oidc_sub = "dex-alice"
|
||||
with patch("members.directory.connection", self.mock_connection), patch.object(self.conn, "search", wraps=self.conn.search) as search:
|
||||
self.assertEqual(directory.reconcile_groups(self.member, {GROUP}, {GROUP}), DN)
|
||||
self.assertEqual(search.call_args_list[0].kwargs["attributes"], ["uid", "mail", "entryUUID"])
|
||||
self.assertIn("(uid=alice)", search.call_args_list[0].args[1])
|
||||
self.assertEqual(self.member.ldap_uuid, "11111111-1111-1111-1111-111111111111")
|
||||
# A recreated uid must never substitute for the now-pinned LDAP entry.
|
||||
self.conn.delete(DN)
|
||||
self.conn.strategy.add_entry(DN, {"objectClass": ["inetOrgPerson"], "uid": "alice", "cn": "alice", "sn": "alice", "mail": "alice@example.org", "entryUUID": "22222222-2222-2222-2222-222222222222"})
|
||||
with patch("members.directory.connection", self.mock_connection), patch.object(self.conn, "modify", wraps=self.conn.modify) as modify:
|
||||
with self.assertRaises(ValueError):
|
||||
directory.reconcile_groups(self.member, {GROUP}, {GROUP})
|
||||
modify.assert_not_called()
|
||||
|
||||
def test_worker_unpinned_identity_requires_trusted_mapping(self):
|
||||
self.member.ldap_uuid = None
|
||||
for uid, issuer, subject in (("alice", "", None), ("alice", "https://other.example", "sub"), ("*)(uid=*)", settings.DEX_ISSUER, "sub")):
|
||||
self.member.uid, self.member.oidc_issuer, self.member.oidc_sub = uid, issuer, subject
|
||||
with patch("members.directory.connection", self.mock_connection):
|
||||
with self.assertRaises(ValueError):
|
||||
directory.reconcile_groups(self.member, {GROUP}, {GROUP})
|
||||
|
||||
def test_worker_refuses_directory_uuid_already_owned_by_another_account(self):
|
||||
account()
|
||||
self.member.ldap_uuid = None
|
||||
self.member.oidc_issuer = settings.DEX_ISSUER
|
||||
self.member.oidc_sub = "dex-alice"
|
||||
with patch("members.directory.connection", self.mock_connection), patch.object(self.conn, "modify", wraps=self.conn.modify) as modify:
|
||||
with self.assertRaises(ValueError):
|
||||
directory.reconcile_groups(self.member, {GROUP}, {GROUP})
|
||||
modify.assert_not_called()
|
||||
|
||||
@override_settings(LDAP_URL="ldaps://ldap.example.org", LDAP_BIND_DN="cn=service", LDAP_BIND_PASSWORD="test-only", LDAP_ACCOUNT_BASE_DN=BASE)
|
||||
@patch("members.directory.Connection")
|
||||
def test_service_bind_failure_is_redacted_and_unbound(self, conn_factory):
|
||||
from ldap3.core.exceptions import LDAPInvalidCredentialsResult
|
||||
|
||||
conn_factory.return_value.bind.side_effect = LDAPInvalidCredentialsResult(description="private diagnostic")
|
||||
with self.assertRaises(directory.LDAPServiceBindError) as raised:
|
||||
with directory.connection():
|
||||
self.fail("Rejected credentials must not reach LDAP operations")
|
||||
self.assertIn("LDAP_BIND_PASSWORD", str(raised.exception))
|
||||
self.assertNotIn("private diagnostic", str(raised.exception))
|
||||
conn_factory.return_value.unbind.assert_called_once()
|
||||
|
||||
@override_settings(LDAP_URL="ldap://ldap.example.org", LDAP_ALLOW_INSECURE=False, LDAP_BIND_DN="cn=service", LDAP_BIND_PASSWORD="test-only", LDAP_ACCOUNT_BASE_DN=BASE)
|
||||
@patch("members.directory.Connection")
|
||||
@@ -515,3 +650,58 @@ class DirectoryTests(SimpleTestCase):
|
||||
self.assertEqual(options["use_ssl"], scheme == "ldaps")
|
||||
self.assertEqual(options["port"], port)
|
||||
self.assertEqual(options["tls"].validate, ssl.CERT_REQUIRED)
|
||||
|
||||
|
||||
@override_settings(LDAP_SEARCH_BASE_DN=BASE, LDAP_GROUP_PLACEHOLDER_DN=SENTINEL, MEMBERSHIP_TIERS=TIERS)
|
||||
class DeferredDirectoryPersistenceTests(TestCase):
|
||||
def test_worker_persists_uuid_and_dn_after_real_offline_ldap_sync(self):
|
||||
member = views.link_identity({"iss": settings.DEX_ISSUER, "sub": "dex-alice", "preferred_username": "alice", "email_verified": True, "email": "alice@example.org", "federated_claims": {"connector_id": settings.DEX_CONNECTOR_ID, "user_id": "alice"}})
|
||||
server = Server("mock", get_info=OFFLINE_SLAPD_2_4) # pyright: ignore[reportArgumentType]
|
||||
conn = Connection(server, client_strategy=MOCK_SYNC)
|
||||
conn.bind()
|
||||
conn.strategy.add_entry(DN, {"objectClass": ["inetOrgPerson"], "uid": "alice", "cn": "alice", "sn": "alice", "mail": "alice@example.org", "entryUUID": "11111111-1111-1111-1111-111111111111"})
|
||||
conn.strategy.add_entry(SENTINEL, {"objectClass": ["organizationalRole"], "cn": "sentinel"})
|
||||
for group in (GROUP, OTHER_GROUP):
|
||||
conn.strategy.add_entry(group, {"objectClass": ["groupOfUniqueNames"], "cn": "paid", "uniqueMember": [SENTINEL]})
|
||||
|
||||
@contextmanager
|
||||
def mock_connection():
|
||||
yield conn
|
||||
|
||||
with patch("members.directory.connection", mock_connection):
|
||||
self.assertTrue(billing.sync_account(member.pk, refresh_stripe=False))
|
||||
member.refresh_from_db()
|
||||
self.assertEqual(member.ldap_uuid, "11111111-1111-1111-1111-111111111111")
|
||||
self.assertEqual(member.ldap_dn, DN)
|
||||
self.assertIsNotNone(member.last_ldap_sync)
|
||||
self.assertEqual(member.membership_level, 0)
|
||||
|
||||
|
||||
class DexIdentityMigrationTests(TransactionTestCase):
|
||||
@override_settings(DEX_ISSUER="https://original-issuer.example/")
|
||||
def test_upgrade_preserves_existing_bindings_and_allows_unresolved_users(self):
|
||||
from django.db import connection
|
||||
from django.db.migrations.executor import MigrationExecutor
|
||||
|
||||
previous = [("members", "0003_membership_levels")]
|
||||
current = [("members", "0004_dex_login_identity")]
|
||||
executor = MigrationExecutor(connection)
|
||||
executor.migrate(previous)
|
||||
try:
|
||||
OldAccount = executor.loader.project_state(previous).apps.get_model("members", "Account")
|
||||
linked = OldAccount.objects.create(uid="alice", email="alice@example.org", ldap_dn=DN, ldap_uuid="uuid-alice", oidc_sub="existing-sub", stripe_customer="cus_existing")
|
||||
unlinked = OldAccount.objects.create(uid="bob", email="bob@example.org", ldap_dn="uid=bob," + BASE, ldap_uuid="uuid-bob")
|
||||
executor = MigrationExecutor(connection)
|
||||
executor.migrate(current)
|
||||
NewAccount = executor.loader.project_state(current).apps.get_model("members", "Account")
|
||||
restored = NewAccount.objects.values("oidc_issuer", "oidc_sub", "ldap_uuid", "stripe_customer").get(pk=linked.pk)
|
||||
self.assertEqual(restored["oidc_issuer"], "https://original-issuer.example/")
|
||||
self.assertEqual(restored["oidc_sub"], "existing-sub")
|
||||
self.assertEqual(restored["ldap_uuid"], "uuid-alice")
|
||||
self.assertEqual(restored["stripe_customer"], "cus_existing")
|
||||
self.assertEqual(NewAccount.objects.values_list("oidc_issuer", flat=True).get(pk=unlinked.pk), "")
|
||||
for uid in ("charlie", "david"):
|
||||
new = NewAccount.objects.create(uid=uid, email=f"{uid}@example.org", oidc_issuer=settings.DEX_ISSUER, oidc_sub=f"dex-{uid}")
|
||||
self.assertIsNone(NewAccount.objects.values_list("ldap_uuid", flat=True).get(pk=new.pk))
|
||||
finally:
|
||||
MigrationExecutor(connection).migrate(current)
|
||||
|
||||
@@ -48,6 +48,19 @@ class AuthlibFlowTests(TestCase):
|
||||
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):
|
||||
|
||||
Reference in New Issue
Block a user