483 lines
27 KiB
Python
Executable File
483 lines
27 KiB
Python
Executable File
import hashlib
|
|
import hmac
|
|
import json
|
|
import re
|
|
import time
|
|
from contextlib import contextmanager
|
|
from datetime import timedelta
|
|
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.utils import timezone
|
|
from ldap3 import Connection, MOCK_SYNC, OFFLINE_SLAPD_2_4, Server
|
|
|
|
from members import billing, directory, views
|
|
from members.models import Account, ManagedGroup, Registration, Subscription, WebhookEvent
|
|
|
|
GROUP = "cn=supporter,ou=groups,dc=example,dc=org"
|
|
OTHER_GROUP = "cn=patron,ou=groups,dc=example,dc=org"
|
|
BASE = "ou=people,dc=example,dc=org"
|
|
DN = "uid=alice," + BASE
|
|
SENTINEL = "cn=sentinel,dc=example,dc=org"
|
|
TIERS = {
|
|
"supporter": {"label": "Supporter", "price_id": "price_supporter", "price_label": "$5 / month", "group_dn": GROUP},
|
|
"patron": {"label": "Patron", "price_id": "price_patron", "price_label": "$15 / month", "group_dn": OTHER_GROUP},
|
|
}
|
|
|
|
|
|
def account(**kwargs):
|
|
return Account.objects.create(uid="alice", email="alice@example.org", ldap_dn=DN, ldap_uuid="11111111-1111-1111-1111-111111111111", **kwargs)
|
|
|
|
|
|
def paged(values):
|
|
response = MagicMock()
|
|
response.auto_paging_iter.side_effect = lambda: iter(values)
|
|
return response
|
|
|
|
|
|
@override_settings(MEMBERSHIP_TIERS=TIERS)
|
|
class RegistrationTests(TestCase):
|
|
def test_public_home_and_registration(self):
|
|
response = self.client.get("/members/")
|
|
self.assertContains(response, "Create a free account")
|
|
self.assertContains(response, "Supporter")
|
|
self.assertEqual(response.headers["Cache-Control"], "no-store")
|
|
self.assertContains(self.client.get("/members/register/"), "Send verification email")
|
|
|
|
def start_signup(self):
|
|
self.client.post("/members/register/", {"uid": "alice", "email": "ALICE@example.org"})
|
|
match = re.search(r"/verify/([^/]+)/", str(mail.outbox[-1].body))
|
|
assert match is not None
|
|
token = match.group(1)
|
|
self.assertNotEqual(Registration.objects.get().pk, token)
|
|
return token
|
|
|
|
@patch("members.views.directory.create_account")
|
|
def test_verified_one_use_signup_no_paid_access(self, provision):
|
|
provision.return_value = {"uid": "alice", "email": "alice@example.org", "ldap_dn": DN, "ldap_uuid": "uuid-alice"}
|
|
token = self.start_signup()
|
|
self.assertFalse(Account.objects.exists())
|
|
self.assertFalse(provision.called)
|
|
self.assertRedirects(self.client.get(f"/members/verify/{token}/"), "/members/register/finish/")
|
|
password = "correct-horse-purple-planet-943!"
|
|
response = self.client.post("/members/register/finish/", {"password": password, "confirm_password": password})
|
|
self.assertContains(response, "Account created")
|
|
provision.assert_called_once_with("alice", "alice@example.org", password)
|
|
self.assertEqual(Account.objects.count(), 1)
|
|
self.assertFalse(Subscription.objects.exists())
|
|
self.assertFalse(Registration.objects.exists())
|
|
self.assertNotIn("account_id", self.client.session)
|
|
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)
|
|
|
|
@patch("members.views.directory.create_account")
|
|
def test_expired_token_cannot_provision(self, provision):
|
|
token = self.start_signup()
|
|
Registration.objects.update(expires_at=timezone.now() - timedelta(seconds=1))
|
|
self.assertContains(self.client.get(f"/members/verify/{token}/"), "Link expired")
|
|
self.client.post("/members/register/finish/", {"password": "not-used"})
|
|
provision.assert_not_called()
|
|
|
|
@patch("members.views.directory.create_account")
|
|
def test_password_validation(self, provision):
|
|
token = self.start_signup()
|
|
self.client.get(f"/members/verify/{token}/")
|
|
response = self.client.post("/members/register/finish/", {"password": "123", "confirm_password": "123"})
|
|
self.assertContains(response, "too short")
|
|
provision.assert_not_called()
|
|
|
|
def test_signup_validation_and_throttle(self):
|
|
self.client.post("/members/register/", {"uid": "*)(uid=*)", "email": "not-an-email"})
|
|
self.assertFalse(Registration.objects.exists())
|
|
for _ in range(3):
|
|
self.assertEqual(self.client.post("/members/register/", {"uid": "alice", "email": "alice@example.org"}).status_code, 200)
|
|
self.assertEqual(self.client.post("/members/register/", {"uid": "alice", "email": "alice@example.org"}).status_code, 429)
|
|
|
|
@patch("members.views.send_mail", side_effect=OSError("SMTP unavailable"))
|
|
def test_mail_failure_invalidates_pending_token(self, send):
|
|
response = self.client.post("/members/register/", {"uid": "alice", "email": "alice@example.org"})
|
|
self.assertContains(response, "Email unavailable")
|
|
self.assertFalse(Registration.objects.exists())
|
|
|
|
def test_csrf_and_method_safety(self):
|
|
strict = Client(enforce_csrf_checks=True)
|
|
for path in ("register/", "login/", "logout/", "billing/", "checkout/supporter/"):
|
|
self.assertEqual(strict.post("/members/" + path).status_code, 403)
|
|
self.assertEqual(self.client.get("/members/login/").status_code, 405)
|
|
self.assertEqual(self.client.get("/members/checkout/supporter/").status_code, 405)
|
|
|
|
@patch("members.views.dex")
|
|
def test_same_origin_login_preserves_csrf_protection(self, dex):
|
|
from django.http import HttpResponseRedirect
|
|
|
|
strict = Client(enforce_csrf_checks=True)
|
|
response = strict.get("/members/", secure=True)
|
|
self.assertEqual(response.headers["Referrer-Policy"], "same-origin")
|
|
token = strict.cookies[settings.CSRF_COOKIE_NAME].value
|
|
dex.return_value.authorize_redirect.return_value = HttpResponseRedirect(settings.DEX_ISSUER + "/auth")
|
|
response = strict.post("/members/login/", {"csrfmiddlewaretoken": token}, secure=True, HTTP_ORIGIN="https://testserver")
|
|
self.assertEqual(response.status_code, 302)
|
|
dex.return_value.authorize_redirect.assert_called_once()
|
|
for origin in ("null", "https://attacker.example"):
|
|
response = strict.post("/members/login/", {"csrfmiddlewaretoken": token}, secure=True, HTTP_ORIGIN=origin)
|
|
self.assertEqual(response.status_code, 403)
|
|
response = strict.post("/members/login/", {"csrfmiddlewaretoken": token}, secure=True, HTTP_REFERER="https://testserver/members/")
|
|
self.assertEqual(response.status_code, 302)
|
|
|
|
|
|
@override_settings(MEMBERSHIP_TIERS=TIERS)
|
|
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": "uuid-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"}
|
|
first = views.link_identity(self.claims())
|
|
second = views.link_identity(self.claims())
|
|
self.assertEqual(first.pk, second.pk)
|
|
self.assertEqual(first.oidc_sub, "dex-alice")
|
|
self.assertFalse(Subscription.objects.exists())
|
|
with self.assertRaises(ValueError):
|
|
views.link_identity(self.claims(sub="different-sub"))
|
|
|
|
@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):
|
|
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": "uuid-alice"}, {"connector_id": settings.DEX_CONNECTOR_ID, "user_id": "different-uuid"}, {}):
|
|
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):
|
|
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())
|
|
original.refresh_from_db()
|
|
self.assertEqual(original.oidc_sub, "original")
|
|
|
|
@patch("members.views.dex")
|
|
@patch("members.views.link_identity")
|
|
def test_callback_rotates_session_and_does_not_store_tokens(self, link, dex):
|
|
member = account()
|
|
link.return_value = member
|
|
session = self.client.session
|
|
session["old_state"] = "old"
|
|
session.save()
|
|
old_key = session.session_key
|
|
dex.return_value.authorize_access_token.return_value = {"id_token": "never-persist-this", "userinfo": self.claims(), "access_token": "also-private"}
|
|
self.assertRedirects(self.client.get("/members/oidc/callback/?state=valid&code=example"), "/members/")
|
|
self.assertNotEqual(self.client.session.session_key, old_key)
|
|
self.assertEqual(dict(self.client.session), {"account_id": str(member.pk)})
|
|
|
|
@patch("members.views.dex")
|
|
def test_bad_callback_has_no_login(self, dex):
|
|
dex.return_value.authorize_access_token.side_effect = ValueError("bad state / nonce")
|
|
response = self.client.get("/members/oidc/callback/?code=bad")
|
|
self.assertContains(response, "could not be verified")
|
|
self.assertNotIn("account_id", self.client.session)
|
|
|
|
|
|
@override_settings(MEMBERSHIP_TIERS=TIERS, STRIPE_WEBHOOK_SECRET="whsec_test_only", STRIPE_PORTAL_CONFIGURATION="bpc_test")
|
|
class BillingViewsTests(TestCase):
|
|
def setUp(self):
|
|
self.member = account(stripe_customer="cus_alice")
|
|
session = self.client.session
|
|
session["account_id"] = str(self.member.pk)
|
|
session.save()
|
|
|
|
def signed_event(self, event, age=0):
|
|
payload = json.dumps(event)
|
|
timestamp = int(time.time()) - age
|
|
signature = hmac.new(settings.STRIPE_WEBHOOK_SECRET.encode(), f"{timestamp}.{payload}".encode(), hashlib.sha256).hexdigest()
|
|
return self.client.post("/members/stripe/webhook/", payload, content_type="application/json", HTTP_STRIPE_SIGNATURE=f"t={timestamp},v1={signature}")
|
|
|
|
def event(self, **changes):
|
|
return {"id": "evt_example", "object": "event", "livemode": False, "type": "invoice.paid", "data": {"object": {"customer": "cus_alice"}}, **changes}
|
|
|
|
def test_signed_webhook_durable_and_duplicate_safe(self):
|
|
self.assertEqual(self.signed_event(self.event()).status_code, 200)
|
|
self.assertEqual(self.signed_event(self.event()).status_code, 200)
|
|
self.assertEqual(WebhookEvent.objects.count(), 1)
|
|
self.assertEqual(WebhookEvent.objects.get().account, self.member)
|
|
self.assertIsNone(WebhookEvent.objects.get().processed_at)
|
|
self.assertFalse(Subscription.objects.exists())
|
|
|
|
def test_rejects_bad_expired_wrong_mode_and_connect_events(self):
|
|
self.assertEqual(self.client.post("/members/stripe/webhook/", "{}", content_type="application/json").status_code, 400)
|
|
self.assertEqual(self.signed_event(self.event(), age=301).status_code, 400)
|
|
self.assertEqual(self.signed_event(self.event(livemode=True)).status_code, 400)
|
|
self.assertEqual(self.signed_event(self.event(account="acct_other")).status_code, 400)
|
|
self.assertFalse(WebhookEvent.objects.exists())
|
|
|
|
@patch("members.views.billing.client")
|
|
def test_checkout_server_allowlist_and_metadata(self, api_factory):
|
|
api = api_factory.return_value
|
|
api.subscriptions.list.return_value = paged([])
|
|
api.checkout.sessions.create.return_value = {"id": "cs_first", "url": "https://checkout.stripe.com/c/pay/first"}
|
|
response = self.client.post("/members/checkout/supporter/", {"price_id": "price_attacker", "customer": "cus_other"})
|
|
self.assertEqual(response.status_code, 302)
|
|
params = api.checkout.sessions.create.call_args.args[0]
|
|
self.assertEqual(params["customer"], "cus_alice")
|
|
self.assertEqual(params["line_items"], [{"price": "price_supporter", "quantity": 1}])
|
|
self.assertEqual(params["subscription_data"]["metadata"]["membership_account"], str(self.member.pk))
|
|
self.assertFalse(Subscription.objects.exists())
|
|
self.assertEqual(self.client.post("/members/checkout/attacker/").status_code, 404)
|
|
|
|
@patch("members.views.billing.client")
|
|
def test_open_checkout_reused_and_duplicate_subscription_blocked(self, api_factory):
|
|
api = api_factory.return_value
|
|
api.subscriptions.list.return_value = paged([])
|
|
Account.objects.filter(pk=self.member.pk).update(checkout_id="cs_open")
|
|
api.checkout.sessions.retrieve.return_value = {"status": "open", "url": "https://checkout.stripe.com/c/pay/open"}
|
|
self.assertEqual(self.client.post("/members/checkout/patron/").status_code, 302)
|
|
api.checkout.sessions.create.assert_not_called()
|
|
api.subscriptions.list.return_value = paged([{"status": "active"}])
|
|
self.assertContains(self.client.post("/members/checkout/patron/"), "already have a subscription")
|
|
|
|
@patch("members.views.billing.client")
|
|
def test_portal_scoped_to_session_account(self, api_factory):
|
|
api = api_factory.return_value
|
|
api.billing_portal.sessions.create.return_value = {"url": "https://billing.stripe.com/p/session/test"}
|
|
response = self.client.post("/members/billing/", {"customer": "cus_victim"})
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertEqual(api.billing_portal.sessions.create.call_args.args[0]["customer"], "cus_alice")
|
|
|
|
def test_billing_requires_session(self):
|
|
self.client.post("/members/logout/")
|
|
self.assertRedirects(self.client.post("/members/checkout/supporter/"), "/members/")
|
|
|
|
|
|
@override_settings(MEMBERSHIP_TIERS=TIERS)
|
|
class ReconciliationTests(TestCase):
|
|
def setUp(self):
|
|
self.member = account(stripe_customer="cus_alice")
|
|
self.now = timezone.now()
|
|
self.end = int((self.now + timedelta(days=30)).timestamp())
|
|
self.sub = {"id": "sub_alice", "customer": "cus_alice", "livemode": False, "metadata": {"membership_account": str(self.member.pk)}, "status": "active", "items": {"data": [{"price": {"id": "price_supporter"}}], "has_more": False}, "cancel_at_period_end": False}
|
|
self.invoice = {"id": "in_paid", "customer": "cus_alice", "status": "paid", "livemode": False, "parent": {"subscription_details": {"subscription": "sub_alice"}}}
|
|
self.line = {"amount": 500, "parent": {"subscription_item_details": {"subscription": "sub_alice"}}, "pricing": {"price_details": {"price": "price_supporter"}}, "period": {"start": int(self.now.timestamp()) - 60, "end": self.end}}
|
|
|
|
def api(self):
|
|
api = MagicMock()
|
|
api.subscriptions.list.return_value = paged([self.sub])
|
|
api.invoices.list.return_value = paged([self.invoice])
|
|
api.invoices.line_items.list.return_value = paged([self.line])
|
|
return api
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.billing.client")
|
|
def test_paid_invoice_grants_and_records_service_expiry(self, factory, ldap):
|
|
factory.return_value = self.api()
|
|
self.assertTrue(billing.sync_account(self.member.pk))
|
|
sub = Subscription.objects.get()
|
|
assert sub.paid_until is not None
|
|
self.assertEqual(int(sub.paid_until.timestamp()), self.end)
|
|
self.assertEqual(sub.tier, "supporter")
|
|
self.assertEqual(ldap.call_args.args[1], {GROUP})
|
|
self.assertEqual(ManagedGroup.objects.count(), 2)
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.billing.client")
|
|
def test_cancellation_at_period_end_retains_paid_time(self, factory, ldap):
|
|
self.sub["cancel_at_period_end"] = True
|
|
factory.return_value = self.api()
|
|
billing.sync_account(self.member.pk)
|
|
self.assertEqual(ldap.call_args.args[1], {GROUP})
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.billing.client")
|
|
def test_canceled_trial_unpaid_paused_unknown_never_grant(self, factory, ldap):
|
|
for status in ("canceled", "trialing", "unpaid", "incomplete", "incomplete_expired", "paused"):
|
|
self.sub["status"] = status
|
|
factory.return_value = self.api()
|
|
billing.sync_account(self.member.pk)
|
|
self.assertEqual(ldap.call_args.args[1], set(), status)
|
|
self.sub["status"] = "active"
|
|
self.sub["items"]["data"][0]["price"]["id"] = "price_unknown"
|
|
factory.return_value = self.api()
|
|
billing.sync_account(self.member.pk)
|
|
self.assertEqual(ldap.call_args.args[1], set())
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.billing.client")
|
|
def test_no_invoice_no_grant_even_if_active(self, factory, ldap):
|
|
api = self.api()
|
|
api.invoices.list.return_value = paged([])
|
|
factory.return_value = api
|
|
billing.sync_account(self.member.pk)
|
|
self.assertEqual(ldap.call_args.args[1], set())
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.billing.client")
|
|
def test_wrong_invoice_price_negative_and_future_lines_rejected(self, factory, ldap):
|
|
for change in ({"amount": -500}, {"pricing": {"price_details": {"price": "price_other"}}}, {"period": {"start": self.end, "end": self.end + 300}}):
|
|
api = self.api()
|
|
api.invoices.line_items.list.return_value = paged([{**self.line, **change}])
|
|
factory.return_value = api
|
|
billing.sync_account(self.member.pk)
|
|
self.assertEqual(ldap.call_args.args[1], set())
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.billing.client")
|
|
def test_past_due_only_retains_previously_paid_time(self, factory, ldap):
|
|
self.sub["status"] = "past_due"
|
|
factory.return_value = self.api()
|
|
billing.sync_account(self.member.pk)
|
|
self.assertEqual(ldap.call_args.args[1], {GROUP})
|
|
self.line["period"]["end"] = int((self.now - timedelta(seconds=1)).timestamp())
|
|
factory.return_value = self.api()
|
|
billing.sync_account(self.member.pk)
|
|
self.assertEqual(ldap.call_args.args[1], set())
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.billing.client", side_effect=OSError("offline"))
|
|
def test_local_expiry_during_stripe_outage(self, factory, ldap):
|
|
Subscription.objects.create(stripe_id="sub_alice", account=self.member, tier="supporter", status="active", paid_until=self.now - timedelta(seconds=1))
|
|
self.assertFalse(billing.sync_account(self.member.pk))
|
|
self.assertEqual(ldap.call_args.args[1], set())
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.billing.client")
|
|
def test_local_only_sweep_never_calls_stripe(self, factory, ldap):
|
|
Subscription.objects.create(stripe_id="sub_expired", account=self.member, tier="supporter", status="active", paid_until=self.now - timedelta(seconds=1))
|
|
self.assertTrue(billing.sync_account(self.member.pk, refresh_stripe=False))
|
|
factory.assert_not_called()
|
|
self.assertEqual(ldap.call_args.args[1], set())
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.billing.client")
|
|
def test_snapshot_is_not_partially_applied_on_error(self, factory, ldap):
|
|
old_end = self.now + timedelta(days=1)
|
|
Subscription.objects.create(stripe_id="sub_alice", account=self.member, tier="supporter", status="active", paid_until=old_end)
|
|
api = self.api()
|
|
api.invoices.line_items.list.side_effect = OSError("network unavailable")
|
|
factory.return_value = api
|
|
self.assertFalse(billing.sync_account(self.member.pk))
|
|
self.assertEqual(Subscription.objects.get().paid_until, old_end)
|
|
self.assertEqual(ldap.call_args.args[1], {GROUP})
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.billing.client")
|
|
def test_removed_tier_still_revoked(self, factory, ldap):
|
|
factory.return_value = self.api()
|
|
billing.sync_account(self.member.pk)
|
|
with override_settings(MEMBERSHIP_TIERS={}):
|
|
billing.sync_account(self.member.pk)
|
|
self.assertEqual(ldap.call_args.args[1], set())
|
|
self.assertIn(GROUP, ldap.call_args.args[2])
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.billing.client")
|
|
def test_wrong_customer_mode_metadata_cannot_grant(self, factory, ldap):
|
|
for changes in ({"customer": "cus_victim"}, {"livemode": True}, {"metadata": {"membership_account": "attacker"}}):
|
|
api = self.api()
|
|
api.subscriptions.list.return_value = paged([{**self.sub, **changes}])
|
|
factory.return_value = api
|
|
billing.sync_account(self.member.pk)
|
|
self.assertEqual(ldap.call_args.args[1], set())
|
|
|
|
@patch("members.billing.reconcile_groups", side_effect=OSError("LDAP offline"))
|
|
@patch("members.billing.client")
|
|
def test_ldap_failure_keeps_retry_event_and_management_intent(self, factory, ldap):
|
|
factory.return_value = self.api()
|
|
WebhookEvent.objects.create(stripe_id="evt_pending", account=self.member)
|
|
with self.assertRaises(OSError):
|
|
billing.sync_account(self.member.pk)
|
|
self.assertIsNone(WebhookEvent.objects.get().processed_at)
|
|
self.assertEqual(ManagedGroup.objects.count(), 2)
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.billing.client")
|
|
def test_reordered_webhook_never_restores_canceled_access(self, factory, ldap):
|
|
self.sub["status"] = "canceled"
|
|
factory.return_value = self.api()
|
|
WebhookEvent.objects.create(stripe_id="evt_old_paid", account=self.member)
|
|
billing.sync_account(self.member.pk)
|
|
self.assertEqual(ldap.call_args.args[1], set())
|
|
self.assertIsNotNone(WebhookEvent.objects.get().processed_at)
|
|
|
|
@patch("members.billing.reconcile_groups", return_value=DN)
|
|
@patch("members.billing.client")
|
|
def test_multiple_subscriptions_aggregate_groups(self, factory, ldap):
|
|
api = self.api()
|
|
other_sub = {**self.sub, "id": "sub_other", "items": {"data": [{"price": {"id": "price_patron"}}]}}
|
|
other_invoice = {**self.invoice, "id": "in_other", "parent": {"subscription_details": {"subscription": "sub_other"}}}
|
|
other_line = {**self.line, "parent": {"subscription_item_details": {"subscription": "sub_other"}}, "pricing": {"price_details": {"price": "price_patron"}}}
|
|
api.subscriptions.list.return_value = paged([self.sub, other_sub])
|
|
api.invoices.list.side_effect = lambda params: paged([self.invoice] if params["subscription"] == "sub_alice" else [other_invoice])
|
|
api.invoices.line_items.list.side_effect = lambda inv, params: paged([self.line] if inv == "in_paid" else [other_line])
|
|
factory.return_value = api
|
|
billing.sync_account(self.member.pk)
|
|
self.assertEqual(ldap.call_args.args[1], {GROUP, OTHER_GROUP})
|
|
|
|
|
|
@override_settings(LDAP_SEARCH_BASE_DN=BASE, LDAP_GROUP_PLACEHOLDER_DN=SENTINEL)
|
|
class DirectoryTests(SimpleTestCase):
|
|
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]
|
|
self.conn = Connection(server, client_strategy=MOCK_SYNC)
|
|
self.conn.bind()
|
|
self.conn.strategy.add_entry(DN, {"objectClass": ["inetOrgPerson"], "uid": "alice", "cn": "alice", "sn": "alice", "mail": "alice@example.org", "entryUUID": "11111111-1111-1111-1111-111111111111"})
|
|
self.conn.strategy.add_entry(SENTINEL, {"objectClass": ["organizationalRole"], "cn": "sentinel"})
|
|
self.conn.strategy.add_entry(GROUP, {"objectClass": ["groupOfUniqueNames"], "cn": "supporter", "uniqueMember": [DN]})
|
|
self.member = Account(uid="alice", ldap_dn=DN, ldap_uuid="11111111-1111-1111-1111-111111111111")
|
|
|
|
@contextmanager
|
|
def mock_connection(self):
|
|
yield self.conn
|
|
|
|
def members(self):
|
|
self.conn.search(GROUP, "(objectClass=*)", attributes=["uniqueMember"])
|
|
return set(self.conn.entries[0].uniqueMember.values)
|
|
|
|
def test_final_member_revoked_with_sentinel_and_idempotency(self):
|
|
with patch("members.directory.connection", self.mock_connection):
|
|
directory.reconcile_groups(self.member, set(), {GROUP})
|
|
self.assertEqual(self.members(), {SENTINEL})
|
|
directory.reconcile_groups(self.member, set(), {GROUP})
|
|
directory.reconcile_groups(self.member, {GROUP}, {GROUP})
|
|
directory.reconcile_groups(self.member, {GROUP}, {GROUP})
|
|
self.assertEqual(self.members(), {SENTINEL, DN})
|
|
|
|
def test_missing_sentinel_and_missing_identity_fail_closed(self):
|
|
with patch("members.directory.connection", self.mock_connection):
|
|
self.conn.delete(SENTINEL)
|
|
with self.assertRaises(Exception):
|
|
directory.reconcile_groups(self.member, set(), {GROUP})
|
|
self.assertEqual(self.members(), {DN})
|
|
self.member.ldap_uuid = "missing"
|
|
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=*)")
|
|
|
|
@override_settings(LDAP_URL="ldap://ldap.example.org", LDAP_BIND_DN="cn=service", LDAP_BIND_PASSWORD="test-only", LDAP_ACCOUNT_BASE_DN=BASE)
|
|
@patch("members.directory.Connection")
|
|
def test_starttls_failure_never_binds_credentials(self, conn_factory):
|
|
conn_factory.return_value.start_tls.return_value = False
|
|
with self.assertRaises(Exception):
|
|
with directory.connection():
|
|
pass
|
|
conn_factory.return_value.bind.assert_not_called()
|
|
conn_factory.return_value.unbind.assert_called_once()
|