893 lines
50 KiB
Python
Executable File
893 lines
50 KiB
Python
Executable File
"""Offline self-service profile tests; never bind LDAP or deliver real email.
|
|
|
|
Run only this module with config.test_settings. Credentials and SSH keys are
|
|
ephemeral test data, not service credentials. Directory writes use mocks; LDAP
|
|
reads additionally exercise ldap3's in-memory strategy.
|
|
"""
|
|
|
|
import copy
|
|
import hashlib
|
|
import secrets
|
|
from datetime import timedelta
|
|
from io import BytesIO
|
|
from unittest.mock import MagicMock, patch
|
|
from urllib.parse import urlsplit
|
|
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric import ec, ed25519, rsa
|
|
from django.conf import settings
|
|
from django.core import mail, signing
|
|
from django.core.exceptions import ValidationError
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from django.test import Client, SimpleTestCase, TestCase, override_settings
|
|
from django.urls import Resolver404, resolve, reverse
|
|
from django.utils import timezone
|
|
from ldap3 import BASE, MOCK_SYNC, MODIFY_REPLACE, Connection, Server
|
|
from ldap3.core.exceptions import LDAPException
|
|
from PIL import Image
|
|
|
|
from members import profiles
|
|
from members.models import Account, ProfileEmailChange
|
|
from members.profile_forms import (
|
|
MAX_INPUT_LENGTH,
|
|
MAX_PHOTO_BYTES,
|
|
CurrentPasswordForm,
|
|
ProfileEmailForm,
|
|
ProfileForm,
|
|
ProfilePasswordForm,
|
|
ProfileValuesField,
|
|
)
|
|
|
|
|
|
ALICE_DN = "uid=alice,ou=people,dc=example,dc=org"
|
|
SEARCH_DN = "ou=people,dc=example,dc=org"
|
|
|
|
|
|
def profile_values(**overrides):
|
|
"""The complete READ_ATTRIBUTES shape returned by the real reader."""
|
|
values = {name: [] for name in profiles.READ_ATTRIBUTES}
|
|
values.update({
|
|
"description": ["Community member"],
|
|
"displayName": ["AliceInGame"],
|
|
"givenName": ["Alice"],
|
|
"sn": ["Example"],
|
|
"mail": ["alice@example.org", "alice.other@example.org"],
|
|
"postalAddress": ["1 Example Street$Example City", "2 Other Road$Other City"],
|
|
"labeledURI": ["https://example.org/ Personal website"],
|
|
"entryUUID": ["uuid-alice"],
|
|
"uid": ["alice"],
|
|
"modifyTimestamp": ["20261029120000Z"],
|
|
"objectClass": ["top", "person", "organizationalPerson", "inetOrgPerson"],
|
|
})
|
|
values.update(overrides)
|
|
return values
|
|
|
|
|
|
def jpeg_bytes(size=(800, 400), *, metadata=True):
|
|
output = BytesIO()
|
|
image = Image.new("RGB", size, (50, 80, 120))
|
|
options = {}
|
|
if metadata:
|
|
exif = Image.Exif()
|
|
exif[0x010E] = "private image description"
|
|
options = {"exif": exif, "icc_profile": b"private-test-icc"}
|
|
image.save(output, format="JPEG", **options)
|
|
return output.getvalue()
|
|
|
|
|
|
def photo_upload(raw=None):
|
|
return SimpleUploadedFile(
|
|
"photo.jpg", jpeg_bytes() if raw is None else raw,
|
|
content_type="image/jpeg",
|
|
)
|
|
|
|
|
|
class ProfileFormTests(SimpleTestCase):
|
|
def setUp(self):
|
|
self.password = " " + secrets.token_urlsafe(24) + " "
|
|
|
|
def form(self, data=None, files=None):
|
|
return ProfileForm(
|
|
data={"sn": "Example", "current_password": self.password, **(data or {})},
|
|
files=files,
|
|
)
|
|
|
|
def test_friendly_labels_and_single_gamer_tag(self):
|
|
form = self.form({"displayName": "AliceInGame"})
|
|
self.assertTrue(form.is_valid(), form.errors)
|
|
self.assertEqual(form.fields["displayName"].label, "Gamer tag")
|
|
self.assertEqual(form.fields["sn"].label, "Last name")
|
|
self.assertEqual(form.fields["l"].label, "City/town")
|
|
self.assertEqual(form.cleaned_data["displayName"], ["AliceInGame"])
|
|
invalid = self.form({"displayName": "FirstTag\nSecondTag"})
|
|
self.assertFalse(invalid.is_valid())
|
|
self.assertIn("displayName", invalid.errors)
|
|
|
|
def test_all_editable_text_attributes_have_fields_but_identity_is_read_only(self):
|
|
self.assertTrue(set(profiles.TEXT_ATTRIBUTES) <= set(ProfileForm.base_fields))
|
|
for name in ("uid", "dn", "ldap_dn", "entryUUID", "mail", "userPassword", "objectClass"):
|
|
self.assertNotIn(name, ProfileForm.base_fields)
|
|
|
|
def test_field_groups_and_unique_password_ids(self):
|
|
form = self.form()
|
|
grouped = [
|
|
*form.identity_fields(), *form.contact_fields(),
|
|
*form.address_fields(), *form.authorization_fields(),
|
|
]
|
|
self.assertEqual(len({field.name for field in grouped}), len(grouped))
|
|
ids = [cls()["current_password"].auto_id for cls in (
|
|
ProfileForm, ProfileEmailForm, ProfilePasswordForm, CurrentPasswordForm,
|
|
)]
|
|
self.assertEqual(len(set(ids)), 4)
|
|
|
|
def test_multivalues_deduplicate_and_strip_control_characters(self):
|
|
field = ProfileValuesField()
|
|
self.assertEqual(field.clean(" Alice\r\nBob\rAlice\n\tCarol\x00\u200b "), ["Alice", "Bob", "Carol"])
|
|
self.assertEqual(field.clean(None), [])
|
|
self.assertEqual(field.clean(" \n "), [])
|
|
|
|
def test_value_count_length_and_total_input_limits(self):
|
|
field = ProfileValuesField(value_length=3)
|
|
self.assertEqual(field.clean("abc"), ["abc"])
|
|
for value in ("abcd", "\n".join(str(i) for i in range(17)), "x" * (MAX_INPUT_LENGTH + 1)):
|
|
with self.subTest(length=len(value)), self.assertRaises(ValidationError):
|
|
field.clean(value)
|
|
self.assertEqual(len(ProfileValuesField().clean("\n".join(str(i) for i in range(16)))), 16)
|
|
|
|
def test_surname_required_even_when_only_unsafe_or_blank_text(self):
|
|
for surname in ("", " \n ", "\x00\u200b"):
|
|
with self.subTest(surname=repr(surname)):
|
|
form = self.form({"sn": surname})
|
|
self.assertFalse(form.is_valid())
|
|
self.assertIn("sn", form.errors)
|
|
|
|
def test_postal_addresses_round_trip_multiple_multiline_values(self):
|
|
values = profile_values()
|
|
initial = profiles.initial(values)
|
|
self.assertEqual(initial["postalAddress"], "1 Example Street\nExample City\n\n2 Other Road\nOther City")
|
|
form = self.form(initial)
|
|
self.assertTrue(form.is_valid(), form.errors)
|
|
self.assertEqual(form.cleaned_data["postalAddress"], values["postalAddress"])
|
|
self.assertEqual(ProfileValuesField(postal=True).clean(" A \r\n B \r\n\r\n A\nB "), ["A$B"])
|
|
|
|
def test_postal_literal_dollar_and_backslash_round_trip(self):
|
|
stored = [r"Unit \24 Five$Backslash \5c24", r"Another\5cAddress"]
|
|
text = profiles.initial(profile_values(postalAddress=stored))["postalAddress"]
|
|
self.assertEqual(ProfileValuesField(postal=True).clean(text), stored)
|
|
|
|
def test_http_https_urls_with_optional_labels(self):
|
|
form = self.form({"labeledURI": "https://example.org/a?q=b Website\nhttp://example.net/"})
|
|
self.assertTrue(form.is_valid(), form.errors)
|
|
self.assertEqual(form.cleaned_data["labeledURI"], ["https://example.org/a?q=b Website", "http://example.net/"])
|
|
|
|
def test_urls_reject_other_schemes_credentials_and_backslashes(self):
|
|
for url in (
|
|
"javascript:alert(1)", "data:text/html,test", "file:///etc/passwd",
|
|
"ftp://example.org/file", "//example.org/", "https://user@example.org/",
|
|
"https://user:secret@example.org/", "https://example.org/\\path", "not-a-url",
|
|
):
|
|
with self.subTest(url=url):
|
|
form = self.form({"labeledURI": url})
|
|
self.assertFalse(form.is_valid())
|
|
self.assertIn("labeledURI", form.errors)
|
|
|
|
def test_supported_ssh_keys_multivalued_and_deduplicated_by_key_not_comment(self):
|
|
keys = [
|
|
ed25519.Ed25519PrivateKey.generate().public_key(),
|
|
rsa.generate_private_key(public_exponent=65537, key_size=2048).public_key(),
|
|
ec.generate_private_key(ec.SECP256R1()).public_key(),
|
|
ec.generate_private_key(ec.SECP384R1()).public_key(),
|
|
ec.generate_private_key(ec.SECP521R1()).public_key(),
|
|
]
|
|
lines = [key.public_bytes(serialization.Encoding.OpenSSH, serialization.PublicFormat.OpenSSH).decode() for key in keys]
|
|
form = self.form({"sshPublicKey": "\n".join([lines[0] + " laptop", lines[0] + " other comment", *lines[1:]])})
|
|
self.assertTrue(form.is_valid(), form.errors)
|
|
self.assertEqual(form.cleaned_data["sshPublicKey"], [lines[0] + " laptop", *lines[1:]])
|
|
|
|
def test_invalid_ssh_options_or_private_material_rejected(self):
|
|
public = ed25519.Ed25519PrivateKey.generate().public_key().public_bytes(
|
|
serialization.Encoding.OpenSSH, serialization.PublicFormat.OpenSSH,
|
|
).decode()
|
|
for value in ("ssh-ed25519 invalid", 'command="id" ' + public, "ssh-dss invalid", "-----BEGIN OPENSSH PRIVATE KEY-----"):
|
|
with self.subTest(value=value):
|
|
form = self.form({"sshPublicKey": value})
|
|
self.assertFalse(form.is_valid())
|
|
self.assertIn("sshPublicKey", form.errors)
|
|
|
|
def test_jpeg_is_resized_reencoded_and_metadata_removed(self):
|
|
raw = jpeg_bytes()
|
|
form = self.form(files={"jpegPhoto": photo_upload(raw)})
|
|
self.assertTrue(form.is_valid(), form.errors)
|
|
clean = form.cleaned_data["jpegPhoto"]
|
|
self.assertIsInstance(clean, bytes)
|
|
self.assertNotEqual(clean, raw)
|
|
with Image.open(BytesIO(clean)) as image:
|
|
self.assertEqual(image.format, "JPEG")
|
|
self.assertEqual(image.mode, "RGB")
|
|
self.assertEqual(image.size, (512, 256))
|
|
self.assertFalse(image.getexif())
|
|
self.assertNotIn("icc_profile", image.info)
|
|
self.assertNotIn("comment", image.info)
|
|
|
|
def test_fake_truncated_empty_and_non_jpeg_uploads_rejected(self):
|
|
png = BytesIO()
|
|
Image.new("RGB", (2, 2)).save(png, "PNG")
|
|
for raw in (b"", b"not an image", b"\xff\xd8fake JPEG", jpeg_bytes()[:30], png.getvalue()):
|
|
with self.subTest(length=len(raw)):
|
|
form = self.form(files={"jpegPhoto": photo_upload(raw)})
|
|
self.assertFalse(form.is_valid())
|
|
self.assertIn("jpegPhoto", form.errors)
|
|
|
|
def test_oversize_upload_size_and_actual_read_limits(self):
|
|
for declared_size in (MAX_PHOTO_BYTES + 1, 1):
|
|
with self.subTest(declared_size=declared_size):
|
|
upload = photo_upload(b"x" * (MAX_PHOTO_BYTES + 1))
|
|
upload.size = declared_size
|
|
form = self.form(files={"jpegPhoto": upload})
|
|
self.assertFalse(form.is_valid())
|
|
self.assertIn("jpegPhoto", form.errors)
|
|
|
|
def test_photo_dimension_limit(self):
|
|
form = self.form(files={"jpegPhoto": photo_upload(jpeg_bytes((4097, 1)))})
|
|
self.assertFalse(form.is_valid())
|
|
self.assertIn("jpegPhoto", form.errors)
|
|
|
|
def test_photo_omission_removal_and_conflicting_replacement(self):
|
|
form = self.form()
|
|
self.assertTrue(form.is_valid(), form.errors)
|
|
self.assertIsNone(form.cleaned_data["jpegPhoto"])
|
|
form = self.form({"remove_photo": "on"})
|
|
self.assertTrue(form.is_valid(), form.errors)
|
|
self.assertTrue(form.cleaned_data["remove_photo"])
|
|
form = self.form({"remove_photo": "on"}, {"jpegPhoto": photo_upload()})
|
|
self.assertFalse(form.is_valid())
|
|
self.assertIn("remove_photo", form.errors)
|
|
|
|
def test_current_and_new_password_whitespace_preserved_not_rendered(self):
|
|
new_password = " " + secrets.token_urlsafe(24) + " "
|
|
form = ProfilePasswordForm(data={
|
|
"current_password": self.password,
|
|
"password": new_password, "confirm_password": new_password,
|
|
})
|
|
self.assertTrue(form.is_valid(), form.errors)
|
|
self.assertEqual(form.cleaned_data["current_password"], self.password)
|
|
self.assertEqual(form.cleaned_data["password"], new_password)
|
|
self.assertEqual(form.cleaned_data["confirm_password"], new_password)
|
|
self.assertNotIn(self.password, form.as_p())
|
|
self.assertNotIn(new_password, form.as_p())
|
|
|
|
def test_password_required_length_and_confirmation(self):
|
|
for data, field in (
|
|
({"current_password": ""}, "current_password"),
|
|
({"current_password": "x" * 1025}, "current_password"),
|
|
):
|
|
with self.subTest(field=field, length=len(data[field])):
|
|
form = CurrentPasswordForm(data=data)
|
|
self.assertFalse(form.is_valid())
|
|
self.assertIn(field, form.errors)
|
|
for password, confirm in (("short", "short"), (self.password, self.password.strip())):
|
|
form = ProfilePasswordForm(data={"current_password": self.password, "password": password, "confirm_password": confirm})
|
|
self.assertFalse(form.is_valid())
|
|
|
|
def test_email_casefold_and_current_password_required(self):
|
|
form = ProfileEmailForm(data={"email": "Alice.New@EXAMPLE.ORG", "current_password": self.password})
|
|
self.assertTrue(form.is_valid(), form.errors)
|
|
self.assertEqual(form.cleaned_data["email"], "alice.new@example.org")
|
|
form = ProfileEmailForm(data={"email": "alice.new@example.org"})
|
|
self.assertFalse(form.is_valid())
|
|
self.assertIn("current_password", form.errors)
|
|
|
|
|
|
class ProfileViewTests(TestCase):
|
|
def setUp(self):
|
|
self.account = Account.objects.create(
|
|
uid="alice", email="alice@example.org", oidc_issuer=settings.DEX_ISSUER,
|
|
oidc_sub="sub-alice", ldap_uuid="uuid-alice", ldap_dn=ALICE_DN,
|
|
)
|
|
self.password = " " + secrets.token_urlsafe(24) + " "
|
|
self.values = profile_values()
|
|
self.sign_in(self.client, self.account)
|
|
self.profile_url = reverse("members:profile")
|
|
self.photo_url = reverse("members:profile_photo")
|
|
self.confirm_url = reverse("members:confirm_profile_email")
|
|
# Unlike a loose read mock, this checks the saved pin seen by the view.
|
|
self.read = self.start_patch("members.views.profiles.read_profile", side_effect=self.read_profile)
|
|
self.update = self.start_patch("members.views.profiles.update_profile")
|
|
self.change_password = self.start_patch("members.views.profiles.change_password")
|
|
self.verify_password = self.start_patch("members.views.profiles.verify_password")
|
|
self.change_email = self.start_patch("members.views.profiles.change_email")
|
|
self.start_patch("members.directory.connection", side_effect=AssertionError("Offline test attempted service LDAP"))
|
|
self.start_patch("members.directory._connection", side_effect=AssertionError("Offline test attempted user LDAP"))
|
|
|
|
def start_patch(self, name, **kwargs):
|
|
patcher = patch(name, **kwargs)
|
|
mock = patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
return mock
|
|
|
|
def sign_in(self, client, account):
|
|
session = client.session
|
|
session["account_id"] = str(account.pk)
|
|
session.save()
|
|
|
|
def read_profile(self, account):
|
|
self.assertEqual(account.pk, self.account.pk)
|
|
saved = Account.objects.get(pk=account.pk)
|
|
self.assertEqual((account.uid, account.ldap_uuid, account.ldap_dn), (saved.uid, "uuid-alice", ALICE_DN))
|
|
return copy.deepcopy(self.values)
|
|
|
|
def profile_data(self, **overrides):
|
|
return {
|
|
**profiles.initial(self.values), "action": "profile",
|
|
"revision": profiles.revision(self.account, self.values),
|
|
"current_password": self.password, **overrides,
|
|
}
|
|
|
|
def assert_no_writes(self):
|
|
for operation in (self.update, self.change_password, self.verify_password, self.change_email):
|
|
operation.assert_not_called()
|
|
|
|
def pending(self, *, account=None, expires_at=None, ldap_uuid="uuid-alice"):
|
|
token = secrets.token_urlsafe(32)
|
|
pending = ProfileEmailChange.objects.create(
|
|
token_hash=hashlib.sha256(token.encode()).hexdigest(),
|
|
account=account or self.account, email="alice.new@example.org",
|
|
ldap_uuid=ldap_uuid, expires_at=expires_at or timezone.now() + timedelta(hours=1),
|
|
)
|
|
return token, pending
|
|
|
|
def exchange(self, token):
|
|
return self.client.get(reverse("members:verify_profile_email", args=[token]))
|
|
|
|
def test_anonymous_endpoints_redirect_to_sign_in_page(self):
|
|
client = Client()
|
|
urls = (self.profile_url, self.photo_url, self.confirm_url, reverse("members:verify_profile_email", args=[secrets.token_urlsafe(32)]))
|
|
for url in urls:
|
|
with self.subTest(url=url):
|
|
self.assertRedirects(client.get(url), reverse("members:home"), fetch_redirect_response=False)
|
|
self.read.assert_not_called()
|
|
self.assert_no_writes()
|
|
|
|
def test_get_loads_full_profile_and_revision_for_signed_in_account_only(self):
|
|
response = self.client.get(self.profile_url, {"uid": "mallory", "dn": "cn=admin", "account_id": str(secrets.token_hex(16))})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(response.context["profile_loaded"])
|
|
self.assertEqual(response.context["email_values"], self.values["mail"])
|
|
self.assertEqual(response.context["form"].initial, profiles.initial(self.values))
|
|
proof = signing.loads(response.context["revision"], salt="ldap-profile", max_age=3600)
|
|
self.assertEqual(proof["account"], str(self.account.pk))
|
|
self.assertEqual(proof["uuid"], "uuid-alice")
|
|
self.assertContains(response, 'name="csrfmiddlewaretoken"')
|
|
self.assert_no_writes()
|
|
|
|
def test_csrf_required_for_each_mutating_action_and_email_confirmation(self):
|
|
client = Client(enforce_csrf_checks=True)
|
|
self.sign_in(client, self.account)
|
|
token, pending = self.pending()
|
|
self.assertEqual(client.get(reverse("members:verify_profile_email", args=[token])).status_code, 302)
|
|
for action in ("profile", "email", "password"):
|
|
with self.subTest(action=action):
|
|
self.assertEqual(client.post(self.profile_url, self.profile_data(action=action)).status_code, 403)
|
|
self.assertEqual(client.post(self.confirm_url, {"current_password": self.password}).status_code, 403)
|
|
self.assertTrue(ProfileEmailChange.objects.filter(pk=pending.pk).exists())
|
|
self.assert_no_writes()
|
|
# A real cookie/token pair passes middleware; unknown action reaches 400.
|
|
client.get(self.profile_url)
|
|
csrf = client.cookies[settings.CSRF_COOKIE_NAME].value
|
|
self.assertEqual(client.post(self.profile_url, {"action": "unknown", "csrfmiddlewaretoken": csrf}).status_code, 400)
|
|
|
|
def test_unknown_or_missing_action_is_bad_request(self):
|
|
for data in ({}, {"action": "delete"}, {"action": "userPassword"}):
|
|
with self.subTest(data=data):
|
|
self.assertEqual(self.client.post(self.profile_url, data).status_code, 400)
|
|
self.assert_no_writes()
|
|
|
|
def test_load_failure_disables_all_editing_get_and_post_and_redacts_logs(self):
|
|
self.read.side_effect = LDAPException(self.password)
|
|
with self.assertLogs("members.views", level="ERROR") as captured:
|
|
for method in (self.client.get, self.client.post):
|
|
response = method(self.profile_url, self.profile_data())
|
|
self.assertEqual(response.status_code, 503)
|
|
self.assertFalse(response.context["profile_loaded"])
|
|
self.assertNotContains(response, "<form", status_code=503)
|
|
for key in ("form", "email_form", "password_form", "revision"):
|
|
self.assertNotIn(key, response.context.flatten())
|
|
self.assertNotIn(self.password, "\n".join(captured.output))
|
|
self.assert_no_writes()
|
|
|
|
def test_invalid_profile_preserves_revision_and_does_not_write_or_echo_password(self):
|
|
response = self.client.post(self.profile_url, self.profile_data(sn="", revision="original-revision"))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("sn", response.context["form"].errors)
|
|
self.assertEqual(response.context["revision"], "original-revision")
|
|
self.assertNotContains(response, self.password)
|
|
self.assert_no_writes()
|
|
|
|
def test_invalid_email_and_password_do_not_authorize_or_write(self):
|
|
for data, key in (
|
|
({"action": "email", "email": "not-email", "current_password": self.password}, "email_form"),
|
|
({"action": "email", "email": "new@example.org"}, "email_form"),
|
|
({"action": "password", "current_password": self.password, "password": "short", "confirm_password": "short"}, "password_form"),
|
|
):
|
|
with self.subTest(action=data["action"]):
|
|
response = self.client.post(self.profile_url, data)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(response.context[key].errors)
|
|
self.assert_no_writes()
|
|
self.assertFalse(ProfileEmailChange.objects.exists())
|
|
|
|
def test_profile_save_uses_cleaned_allowlist_current_account_and_exact_revision(self):
|
|
data = self.profile_data(displayName="NewGamerTag", uid="mallory", dn="cn=admin", ldap_uuid="uuid-other", userPassword="injected", mail="attacker@example.org")
|
|
with self.assertLogs("members.views", level="INFO") as captured:
|
|
response = self.client.post(self.profile_url, data)
|
|
self.assertRedirects(response, self.profile_url + "?saved=1", fetch_redirect_response=False)
|
|
account, cleaned, revision = self.update.call_args.args
|
|
self.assertEqual((account.pk, account.ldap_uuid, account.ldap_dn), (self.account.pk, "uuid-alice", ALICE_DN))
|
|
self.assertEqual(revision, data["revision"])
|
|
self.assertEqual(cleaned["displayName"], ["NewGamerTag"])
|
|
self.assertEqual(cleaned["postalAddress"], self.values["postalAddress"])
|
|
self.assertEqual(cleaned["current_password"], self.password)
|
|
for name in ("uid", "dn", "ldap_uuid", "userPassword", "mail"):
|
|
self.assertNotIn(name, cleaned)
|
|
self.assertNotIn(self.password, "\n".join(captured.output))
|
|
self.account.refresh_from_db()
|
|
self.assertEqual(self.account.uid, "alice")
|
|
self.assertEqual(self.account.email, "alice@example.org")
|
|
|
|
def test_profile_write_failure_is_generic_and_password_never_logged(self):
|
|
self.update.side_effect = LDAPException("private server diagnostic " + self.password)
|
|
with self.assertLogs("members.views", level="ERROR") as captured:
|
|
response = self.client.post(self.profile_url, self.profile_data())
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(response.context["form"].non_field_errors())
|
|
output = "\n".join(captured.output)
|
|
self.assertIn("kind=LDAPException", output)
|
|
self.assertNotIn(self.password, output)
|
|
self.assertNotIn("private server diagnostic", output)
|
|
self.assertNotContains(response, self.password)
|
|
|
|
def test_password_change_uses_current_password_then_flushes_local_session(self):
|
|
new_password = " " + secrets.token_urlsafe(24) + " "
|
|
old_key = self.client.session.session_key
|
|
with self.assertLogs("members.views", level="INFO") as captured:
|
|
response = self.client.post(self.profile_url, {
|
|
"action": "password", "current_password": self.password,
|
|
"password": new_password, "confirm_password": new_password,
|
|
"uid": "mallory", "dn": "cn=admin",
|
|
})
|
|
self.assertEqual(response.status_code, 200)
|
|
account, current, new = self.change_password.call_args.args
|
|
self.assertEqual(account.pk, self.account.pk)
|
|
self.assertEqual((current, new), (self.password, new_password))
|
|
self.assertNotIn("account_id", self.client.session)
|
|
self.assertNotEqual(self.client.session.session_key, old_key)
|
|
from django.contrib.sessions.models import Session
|
|
self.assertFalse(Session.objects.filter(session_key=old_key).exists())
|
|
for secret in (self.password, new_password):
|
|
self.assertNotIn(secret, "\n".join(captured.output))
|
|
self.assertNotContains(response, secret)
|
|
self.update.assert_not_called()
|
|
self.change_email.assert_not_called()
|
|
|
|
def test_failed_password_change_retains_session(self):
|
|
self.change_password.side_effect = ValueError(self.password)
|
|
new_password = secrets.token_urlsafe(24)
|
|
with self.assertLogs("members.views", level="ERROR") as captured:
|
|
response = self.client.post(self.profile_url, {
|
|
"action": "password", "current_password": self.password,
|
|
"password": new_password, "confirm_password": new_password,
|
|
})
|
|
self.assertTrue(response.context["password_form"].non_field_errors())
|
|
self.assertEqual(self.client.session["account_id"], str(self.account.pk))
|
|
self.assertNotIn(self.password, "\n".join(captured.output))
|
|
|
|
def test_email_request_verifies_password_stores_only_hash_and_sends_link_not_write(self):
|
|
before = timezone.now()
|
|
response = self.client.post(self.profile_url, {"action": "email", "email": "Alice.New@EXAMPLE.ORG", "current_password": self.password})
|
|
self.assertRedirects(response, self.profile_url + "?email_sent=1", fetch_redirect_response=False)
|
|
account, password = self.verify_password.call_args.args
|
|
self.assertEqual(account.pk, self.account.pk)
|
|
self.assertEqual(password, self.password)
|
|
pending = ProfileEmailChange.objects.get(account=self.account)
|
|
self.assertEqual((pending.email, pending.ldap_uuid), ("alice.new@example.org", "uuid-alice"))
|
|
self.assertGreaterEqual(pending.expires_at, before + timedelta(hours=1))
|
|
self.assertLessEqual(pending.expires_at, timezone.now() + timedelta(hours=1))
|
|
self.assertEqual(len(mail.outbox), 1)
|
|
message = mail.outbox[0]
|
|
self.assertEqual(message.to, ["alice.new@example.org"])
|
|
link = next(line for line in message.body.splitlines() if line.startswith(settings.PUBLIC_URL))
|
|
token = resolve(urlsplit(link).path).kwargs["token"]
|
|
self.assertEqual(pending.token_hash, hashlib.sha256(token.encode()).hexdigest())
|
|
self.assertNotEqual(pending.token_hash, token)
|
|
self.assertNotIn(token, repr(list(ProfileEmailChange.objects.values())))
|
|
self.assertNotIn(self.password, message.body)
|
|
self.account.refresh_from_db()
|
|
self.assertEqual(self.account.email, "alice@example.org")
|
|
self.change_email.assert_not_called()
|
|
self.update.assert_not_called()
|
|
|
|
def test_new_email_request_replaces_previous_pending_token(self):
|
|
_, old = self.pending()
|
|
self.client.post(self.profile_url, {"action": "email", "email": "next@example.org", "current_password": self.password})
|
|
self.assertFalse(ProfileEmailChange.objects.filter(pk=old.pk).exists())
|
|
self.assertEqual(ProfileEmailChange.objects.get().email, "next@example.org")
|
|
|
|
def test_bad_current_password_does_not_create_email_token_or_send_mail(self):
|
|
self.verify_password.side_effect = ValueError(self.password)
|
|
with patch("members.views.send_mail") as send, self.assertLogs("members.views", level="ERROR"):
|
|
response = self.client.post(self.profile_url, {"action": "email", "email": "new@example.org", "current_password": self.password})
|
|
self.assertTrue(response.context["email_form"].non_field_errors())
|
|
self.assertFalse(ProfileEmailChange.objects.exists())
|
|
send.assert_not_called()
|
|
self.change_email.assert_not_called()
|
|
|
|
def test_smtp_failure_rolls_back_new_token_and_redacts_exception(self):
|
|
with patch("members.views.send_mail", side_effect=OSError(self.password)), self.assertLogs("members.views", level="ERROR") as captured:
|
|
response = self.client.post(self.profile_url, {"action": "email", "email": "new@example.org", "current_password": self.password})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(response.context["email_form"].non_field_errors())
|
|
self.assertFalse(ProfileEmailChange.objects.exists())
|
|
self.assertNotIn(self.password, "\n".join(captured.output))
|
|
self.change_email.assert_not_called()
|
|
|
|
def test_email_token_get_only_exchanges_for_server_side_hash(self):
|
|
token, pending = self.pending()
|
|
response = self.exchange(token)
|
|
self.assertRedirects(response, self.confirm_url, fetch_redirect_response=False)
|
|
self.assertNotIn(token, response.url)
|
|
self.assertEqual(self.client.session["profile_email_hash"], pending.pk)
|
|
self.assertTrue(ProfileEmailChange.objects.filter(pk=pending.pk).exists())
|
|
self.assert_no_writes()
|
|
response = self.client.get(self.confirm_url)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, pending.email)
|
|
self.assert_no_writes()
|
|
|
|
def test_confirmation_requires_current_password_and_consumes_token_once(self):
|
|
token, pending = self.pending()
|
|
self.exchange(token)
|
|
response = self.client.post(self.confirm_url, {})
|
|
self.assertIn("current_password", response.context["confirmation_form"].errors)
|
|
self.change_email.assert_not_called()
|
|
self.assertTrue(ProfileEmailChange.objects.filter(pk=pending.pk).exists())
|
|
old_key = self.client.session.session_key
|
|
response = self.client.post(self.confirm_url, {"current_password": self.password, "email": "attacker@example.org", "uid": "mallory"})
|
|
self.assertRedirects(response, self.profile_url + "?saved=1", fetch_redirect_response=False)
|
|
account, password, email = self.change_email.call_args.args
|
|
self.assertEqual((account.pk, account.ldap_uuid, password, email), (self.account.pk, "uuid-alice", self.password, pending.email))
|
|
self.account.refresh_from_db()
|
|
self.assertEqual(self.account.email, pending.email)
|
|
self.assertFalse(ProfileEmailChange.objects.filter(account=self.account).exists())
|
|
self.assertNotIn("profile_email_hash", self.client.session)
|
|
self.assertEqual(self.client.session["account_id"], str(self.account.pk))
|
|
self.assertNotEqual(self.client.session.session_key, old_key)
|
|
self.assertContains(self.exchange(token), "Link unavailable")
|
|
self.assertContains(self.client.post(self.confirm_url, {"current_password": self.password}), "Link unavailable")
|
|
self.change_email.assert_called_once()
|
|
|
|
def test_confirmation_rejects_another_signed_in_account_even_with_session_hash(self):
|
|
token, pending = self.pending()
|
|
bob = Account.objects.create(uid="bob", email="bob@example.org", oidc_issuer=settings.DEX_ISSUER, oidc_sub="sub-bob", ldap_uuid="uuid-bob", ldap_dn="uid=bob," + SEARCH_DN)
|
|
self.sign_in(self.client, bob)
|
|
self.assertContains(self.exchange(token), "Link unavailable")
|
|
session = self.client.session
|
|
session["profile_email_hash"] = pending.pk
|
|
session.save()
|
|
self.assertContains(self.client.post(self.confirm_url, {"current_password": self.password}), "Link unavailable")
|
|
self.assertTrue(ProfileEmailChange.objects.filter(pk=pending.pk).exists())
|
|
self.read.assert_not_called()
|
|
self.assert_no_writes()
|
|
|
|
def test_expired_unknown_and_expired_after_exchange_tokens_cannot_write(self):
|
|
expired_token, _ = self.pending(expires_at=timezone.now() - timedelta(seconds=1))
|
|
self.assertContains(self.exchange(expired_token), "Link unavailable")
|
|
self.assertContains(self.exchange(secrets.token_urlsafe(32)), "Link unavailable")
|
|
token, pending = self.pending()
|
|
self.exchange(token)
|
|
ProfileEmailChange.objects.filter(pk=pending.pk).update(expires_at=timezone.now() - timedelta(seconds=1))
|
|
self.assertContains(self.client.post(self.confirm_url, {"current_password": self.password}), "Link unavailable")
|
|
self.assert_no_writes()
|
|
|
|
def test_confirmation_rejects_changed_uuid_and_keeps_old_email(self):
|
|
token, pending = self.pending(ldap_uuid="uuid-before-recreation")
|
|
self.exchange(token)
|
|
with self.assertLogs("members.views", level="ERROR"):
|
|
response = self.client.post(self.confirm_url, {"current_password": self.password})
|
|
self.assertTrue(response.context["confirmation_form"].non_field_errors())
|
|
self.change_email.assert_not_called()
|
|
self.account.refresh_from_db()
|
|
self.assertEqual(self.account.email, "alice@example.org")
|
|
self.assertTrue(ProfileEmailChange.objects.filter(pk=pending.pk).exists())
|
|
|
|
def test_confirmation_ldap_failure_preserves_pending_and_redacts_logs(self):
|
|
token, pending = self.pending()
|
|
self.exchange(token)
|
|
self.change_email.side_effect = ValueError(self.password)
|
|
with self.assertLogs("members.views", level="ERROR") as captured:
|
|
response = self.client.post(self.confirm_url, {"current_password": self.password})
|
|
self.assertTrue(response.context["confirmation_form"].non_field_errors())
|
|
self.assertTrue(ProfileEmailChange.objects.filter(pk=pending.pk).exists())
|
|
self.assertNotIn(self.password, "\n".join(captured.output))
|
|
self.assertNotContains(response, self.password)
|
|
self.account.refresh_from_db()
|
|
self.assertEqual(self.account.email, "alice@example.org")
|
|
|
|
def test_photo_is_sanitized_and_bound_to_current_user_not_query_or_path(self):
|
|
self.values["jpegPhoto"] = [jpeg_bytes()]
|
|
response = self.client.get(self.photo_url, {"uid": "mallory", "dn": "cn=admin", "path": "../../etc/passwd"})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response["Content-Type"], "image/jpeg")
|
|
self.assertEqual(response["Content-Disposition"], 'inline; filename="profile.jpg"')
|
|
with Image.open(BytesIO(response.content)) as image:
|
|
self.assertEqual(image.size, (512, 256))
|
|
self.assertFalse(image.getexif())
|
|
self.assertNotEqual(response.content, self.values["jpegPhoto"][0])
|
|
for suffix in ("alice/", "../secret", "%2e%2e%2fsecret"):
|
|
with self.subTest(suffix=suffix), self.assertRaises(Resolver404):
|
|
resolve(self.photo_url + suffix)
|
|
self.assert_no_writes()
|
|
|
|
def test_missing_invalid_or_nonbytes_stored_photo_returns_404(self):
|
|
for photos in ([], [b"fake JPEG"], ["not bytes"]):
|
|
with self.subTest(photos=photos):
|
|
self.values["jpegPhoto"] = photos
|
|
self.assertEqual(self.client.get(self.photo_url).status_code, 404)
|
|
self.assert_no_writes()
|
|
|
|
def test_get_only_endpoints_reject_post(self):
|
|
token, _ = self.pending()
|
|
for url in (self.photo_url, reverse("members:verify_profile_email", args=[token])):
|
|
self.assertEqual(self.client.post(url).status_code, 405)
|
|
self.assert_no_writes()
|
|
|
|
|
|
@override_settings(LDAP_SEARCH_BASE_DN=SEARCH_DN, LDAP_GROUP_PLACEHOLDER_DN="cn=sentinel,dc=example,dc=org")
|
|
class ProfileDirectoryTests(TestCase):
|
|
def setUp(self):
|
|
self.account = Account.objects.create(
|
|
uid="alice", email="alice@example.org", oidc_issuer=settings.DEX_ISSUER,
|
|
oidc_sub="sub-alice", ldap_uuid="uuid-alice", ldap_dn=ALICE_DN,
|
|
)
|
|
self.password = secrets.token_urlsafe(24)
|
|
self.values = profile_values()
|
|
self.conn = Connection(Server("offline.invalid"), client_strategy=MOCK_SYNC)
|
|
self.conn.strategy.add_entry(ALICE_DN, self.values)
|
|
self.conn.bind()
|
|
self.addCleanup(self.conn.unbind)
|
|
# No network access can escape this module, even if a test regresses.
|
|
patcher = patch("members.directory.Connection", side_effect=AssertionError("Real LDAP prohibited"))
|
|
patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
|
|
def service(self):
|
|
manager = MagicMock()
|
|
manager.__enter__.return_value = self.conn
|
|
return patch("members.profiles.directory.connection", return_value=manager)
|
|
|
|
def authorized(self, *, values=None):
|
|
conn = MagicMock()
|
|
conn.modify.return_value = True
|
|
conn.extend.standard.modify_password.return_value = True
|
|
conn.entries = []
|
|
authorized = MagicMock()
|
|
authorized.return_value.__enter__.return_value = (conn, self.values if values is None else values)
|
|
return conn, authorized
|
|
|
|
def cleaned(self, **changes):
|
|
return {
|
|
**{name: list(self.values[name]) for name in profiles.TEXT_ATTRIBUTES},
|
|
"current_password": self.password, "jpegPhoto": None, "remove_photo": False,
|
|
**changes,
|
|
}
|
|
|
|
def assert_critical_assertion(self, call):
|
|
controls = call.kwargs["controls"]
|
|
self.assertEqual(len(controls), 1)
|
|
self.assertEqual(str(controls[0]["controlType"]), "1.3.6.1.1.12")
|
|
self.assertTrue(bool(controls[0]["criticality"]))
|
|
encoded = bytes(controls[0]["controlValue"])
|
|
self.assertIn(b"entryUUID", encoded)
|
|
self.assertIn(b"uuid-alice", encoded)
|
|
self.assertIn(b"modifyTimestamp", encoded)
|
|
self.assertIn(b"20261029120000Z", encoded)
|
|
|
|
def test_read_uses_uuid_filter_fixed_attributes_and_persists_renamed_dn(self):
|
|
self.account.ldap_dn = "uid=old," + SEARCH_DN
|
|
self.account.save(update_fields=["ldap_dn"])
|
|
with self.service(), patch.object(self.conn, "search", wraps=self.conn.search) as search:
|
|
values = profiles.read_profile(self.account)
|
|
self.assertEqual(values["uid"], ["alice"])
|
|
self.assertEqual(values["entryUUID"], ["uuid-alice"])
|
|
self.assertEqual(set(values), set(profiles.READ_ATTRIBUTES))
|
|
search.assert_called_once_with(SEARCH_DN, "(entryUUID=uuid-alice)", attributes=list(profiles.READ_ATTRIBUTES))
|
|
self.assertNotIn("userPassword", search.call_args.kwargs["attributes"])
|
|
self.account.refresh_from_db()
|
|
self.assertEqual(self.account.ldap_dn, ALICE_DN)
|
|
|
|
def test_unpinned_verified_identity_resolves_uid_and_saves_uuid(self):
|
|
self.account.ldap_uuid = None
|
|
self.account.ldap_dn = ""
|
|
self.account.save(update_fields=["ldap_uuid", "ldap_dn"])
|
|
with self.service(), patch.object(self.conn, "search", wraps=self.conn.search) as search:
|
|
profiles.read_profile(self.account)
|
|
self.assertEqual(search.call_args.args[1], "(&(objectClass=inetOrgPerson)(uid=alice))")
|
|
self.account.refresh_from_db()
|
|
self.assertEqual((self.account.ldap_uuid, self.account.ldap_dn), ("uuid-alice", ALICE_DN))
|
|
|
|
def test_missing_pinned_uuid_never_falls_back_to_matching_uid(self):
|
|
self.account.ldap_uuid = "uuid-deleted"
|
|
self.account.save(update_fields=["ldap_uuid"])
|
|
with self.service(), patch.object(self.conn, "search", wraps=self.conn.search) as search:
|
|
with self.assertRaisesRegex(ValueError, "missing or ambiguous"):
|
|
profiles.read_profile(self.account)
|
|
search.assert_called_once_with(SEARCH_DN, "(entryUUID=uuid-deleted)", attributes=list(profiles.READ_ATTRIBUTES))
|
|
self.account.refresh_from_db()
|
|
self.assertEqual(self.account.ldap_uuid, "uuid-deleted")
|
|
|
|
def test_unverified_subject_or_wrong_issuer_refused_before_directory_read(self):
|
|
for changes in ({"oidc_sub": None}, {"oidc_issuer": "https://other.example.org/"}):
|
|
with self.subTest(changes=changes):
|
|
Account.objects.filter(pk=self.account.pk).update(**changes)
|
|
with patch("members.profiles.directory.connection") as service:
|
|
with self.assertRaisesRegex(ValueError, "verified Dex"):
|
|
profiles.read_profile(self.account)
|
|
service.assert_not_called()
|
|
Account.objects.filter(pk=self.account.pk).update(oidc_sub="sub-alice", oidc_issuer=settings.DEX_ISSUER)
|
|
|
|
def test_uid_mismatch_duplicate_uuid_and_sentinel_refused(self):
|
|
self.conn.strategy.entries[ALICE_DN]["uid"] = [b"mallory"]
|
|
with self.service(), self.assertRaisesRegex(ValueError, "does not match"):
|
|
profiles.read_profile(self.account)
|
|
self.conn.strategy.entries[ALICE_DN]["uid"] = [b"alice"]
|
|
self.account.ldap_uuid = None
|
|
self.account.save(update_fields=["ldap_uuid"])
|
|
bob = Account.objects.create(uid="bob", email="bob@example.org", ldap_uuid="uuid-alice")
|
|
with self.service(), self.assertRaisesRegex(ValueError, "another account"):
|
|
profiles.read_profile(self.account)
|
|
bob.delete()
|
|
with self.service(), override_settings(LDAP_GROUP_PLACEHOLDER_DN=ALICE_DN.upper()), self.assertRaisesRegex(ValueError, "System entries"):
|
|
profiles.read_profile(self.account)
|
|
|
|
def test_authorized_uses_user_bind_base_scope_and_fixed_read_allowlist(self):
|
|
manager = MagicMock()
|
|
manager.__enter__.return_value = self.conn
|
|
with patch("members.profiles.directory._connection", return_value=manager) as connection, patch.object(self.conn, "search", wraps=self.conn.search) as search:
|
|
profiles.verify_password(self.account, self.password)
|
|
connection.assert_called_once_with(ALICE_DN, self.password)
|
|
search.assert_called_once_with(ALICE_DN, "(entryUUID=uuid-alice)", search_scope=BASE, attributes=list(profiles.READ_ATTRIBUTES))
|
|
|
|
def test_authorized_requires_current_password_uuid_dn_and_matching_identity(self):
|
|
for attribute in ("ldap_uuid", "ldap_dn"):
|
|
saved = getattr(self.account, attribute)
|
|
setattr(self.account, attribute, "")
|
|
with self.subTest(attribute=attribute), patch("members.profiles.directory._connection") as connection:
|
|
with self.assertRaises(ValueError):
|
|
profiles.verify_password(self.account, self.password)
|
|
connection.assert_not_called()
|
|
setattr(self.account, attribute, saved)
|
|
with patch("members.profiles.directory._connection") as connection:
|
|
with self.assertRaises(ValueError):
|
|
profiles.verify_password(self.account, "")
|
|
connection.assert_not_called()
|
|
self.conn.strategy.entries[ALICE_DN]["uid"] = [b"mallory"]
|
|
manager = MagicMock()
|
|
manager.__enter__.return_value = self.conn
|
|
with patch("members.profiles.directory._connection", return_value=manager), self.assertRaisesRegex(ValueError, "identity changed"):
|
|
profiles.verify_password(self.account, self.password)
|
|
|
|
def test_profile_update_allowlist_and_atomic_critical_assertion(self):
|
|
conn, authorized = self.authorized()
|
|
cleaned = self.cleaned(displayName=["Changed"], uid=["mallory"], mail=["bad@example.org"], userPassword=["injected"], objectClass=["admin"])
|
|
revision = profiles.revision(self.account, self.values)
|
|
with patch("members.profiles._authorized", authorized):
|
|
profiles.update_profile(self.account, cleaned, revision)
|
|
authorized.assert_called_once_with(self.account, self.password)
|
|
self.assertEqual(conn.modify.call_args.args, (ALICE_DN, {"displayName": [(MODIFY_REPLACE, ["Changed"])]}))
|
|
self.assert_critical_assertion(conn.modify.call_args)
|
|
|
|
def test_profile_edit_without_timestamp_fails_closed(self):
|
|
values = profile_values(modifyTimestamp=[])
|
|
conn, authorized = self.authorized(values=values)
|
|
with patch("members.profiles._authorized", authorized), self.assertRaises(ValueError):
|
|
profiles.update_profile(self.account, self.cleaned(displayName=["Changed"]), profiles.revision(self.account, values))
|
|
conn.modify.assert_not_called()
|
|
|
|
def test_ssh_edit_adds_only_known_auxiliary_class_when_missing(self):
|
|
from ldap3 import MODIFY_ADD
|
|
|
|
conn, authorized = self.authorized()
|
|
with patch("members.profiles._authorized", authorized):
|
|
profiles.update_profile(self.account, self.cleaned(sshPublicKey=["validated-public-key"]), profiles.revision(self.account, self.values))
|
|
changes = conn.modify.call_args.args[1]
|
|
self.assertEqual(changes["objectClass"], [(MODIFY_ADD, ["ldapPublicKey"])])
|
|
self.assertEqual(changes["sshPublicKey"], [(MODIFY_REPLACE, ["validated-public-key"])])
|
|
self.assert_critical_assertion(conn.modify.call_args)
|
|
|
|
def test_noop_does_not_write_and_blank_optional_value_clears(self):
|
|
conn, authorized = self.authorized()
|
|
with patch("members.profiles._authorized", authorized):
|
|
profiles.update_profile(self.account, self.cleaned(), profiles.revision(self.account, self.values))
|
|
conn.modify.assert_not_called()
|
|
profiles.update_profile(self.account, self.cleaned(description=[]), profiles.revision(self.account, self.values))
|
|
self.assertEqual(conn.modify.call_args.args[1], {"description": [(MODIFY_REPLACE, [])]})
|
|
|
|
def test_photo_replacement_and_removal_are_binary_ldap_changes(self):
|
|
for changes, expected in (({"jpegPhoto": b"sanitized photo"}, [b"sanitized photo"]), ({"remove_photo": True}, [])):
|
|
with self.subTest(changes=changes):
|
|
values = profile_values(jpegPhoto=[b"previous photo"])
|
|
conn, authorized = self.authorized(values=values)
|
|
with patch("members.profiles._authorized", authorized):
|
|
profiles.update_profile(self.account, self.cleaned(**changes), profiles.revision(self.account, values))
|
|
self.assertEqual(conn.modify.call_args.args[1], {"jpegPhoto": [(MODIFY_REPLACE, expected)]})
|
|
self.assert_critical_assertion(conn.modify.call_args)
|
|
|
|
def test_revision_bound_to_account_uuid_and_all_read_values(self):
|
|
valid = profiles.revision(self.account, self.values)
|
|
variants = [valid + "tampered", "", signing.dumps({"account": "other", "uuid": "uuid-alice", "digest": profiles._digest(self.values)}, salt="ldap-profile")]
|
|
with patch("members.profiles.signing.time.time", return_value=timezone.now().timestamp() - 3601):
|
|
variants.append(profiles.revision(self.account, self.values))
|
|
for revision in variants:
|
|
with self.subTest(revision=revision):
|
|
conn, authorized = self.authorized()
|
|
with patch("members.profiles._authorized", authorized), self.assertRaises((ValueError, signing.BadSignature)):
|
|
profiles.update_profile(self.account, self.cleaned(displayName=["Changed"]), revision)
|
|
conn.modify.assert_not_called()
|
|
for changed in (profile_values(displayName=["Concurrent edit"]), profile_values(jpegPhoto=[b"new photo"]), profile_values(modifyTimestamp=["20261029120100Z"])):
|
|
conn, authorized = self.authorized(values=changed)
|
|
with patch("members.profiles._authorized", authorized), self.assertRaisesRegex(ValueError, "profile changed"):
|
|
profiles.update_profile(self.account, self.cleaned(), valid)
|
|
conn.modify.assert_not_called()
|
|
self.account.ldap_uuid = "uuid-recreated"
|
|
conn, authorized = self.authorized()
|
|
with patch("members.profiles._authorized", authorized), self.assertRaises(ValueError):
|
|
profiles.update_profile(self.account, self.cleaned(), valid)
|
|
conn.modify.assert_not_called()
|
|
|
|
def test_password_uses_rfc3062_not_userpassword_attribute_or_local_hashing(self):
|
|
conn, authorized = self.authorized()
|
|
new_password = secrets.token_urlsafe(24)
|
|
with patch("members.profiles._authorized", authorized):
|
|
profiles.change_password(self.account, self.password, new_password)
|
|
authorized.assert_called_once_with(self.account, self.password)
|
|
call = conn.extend.standard.modify_password.call_args
|
|
self.assertEqual(call.args, ())
|
|
self.assertEqual(call.kwargs["old_password"], self.password)
|
|
self.assertEqual(call.kwargs["new_password"], new_password)
|
|
self.assertNotIn("controls", call.kwargs)
|
|
conn.modify.assert_not_called()
|
|
|
|
def test_email_change_replaces_mail_with_critical_assertion(self):
|
|
conn, authorized = self.authorized()
|
|
with patch("members.profiles._authorized", authorized):
|
|
profiles.change_email(self.account, self.password, "new@example.org")
|
|
conn.search.assert_called_once_with(SEARCH_DN, "(mail=new@example.org)", attributes=["entryUUID"])
|
|
self.assertEqual(conn.modify.call_args.args, (ALICE_DN, {"mail": [(MODIFY_REPLACE, ["new@example.org"])]}))
|
|
self.assert_critical_assertion(conn.modify.call_args)
|
|
|
|
def test_email_uniqueness_rejects_other_uuid_but_allows_current_entry(self):
|
|
for uuid, allowed in (("uuid-bob", False), ("uuid-alice", True)):
|
|
conn, authorized = self.authorized()
|
|
entry = MagicMock()
|
|
entry.entryUUID.value = uuid
|
|
conn.entries = [entry]
|
|
with self.subTest(uuid=uuid), patch("members.profiles._authorized", authorized):
|
|
if allowed:
|
|
profiles.change_email(self.account, self.password, "new@example.org")
|
|
conn.modify.assert_called_once()
|
|
else:
|
|
with self.assertRaises(ValueError):
|
|
profiles.change_email(self.account, self.password, "new@example.org")
|
|
conn.modify.assert_not_called()
|
|
|
|
def test_rejected_ldap_writes_raise_instead_of_reporting_success(self):
|
|
conn, authorized = self.authorized()
|
|
conn.modify.return_value = False
|
|
conn.extend.standard.modify_password.return_value = False
|
|
with patch("members.profiles._authorized", authorized):
|
|
with self.assertRaises(LDAPException):
|
|
profiles.update_profile(self.account, self.cleaned(displayName=["Changed"]), profiles.revision(self.account, self.values))
|
|
with self.assertRaises(LDAPException):
|
|
profiles.change_password(self.account, self.password, secrets.token_urlsafe(24))
|
|
with self.assertRaises(LDAPException):
|
|
profiles.change_email(self.account, self.password, "new@example.org")
|