Profile: only edit fields Dex reads back
The Dex LDAP connector publishes uid, cn and mail as claims, so the profile page now offers just Name (cn), the verified email change and the password. The photo endpoint, SSH keys, addresses, phones and other extended attributes are no longer read or written; existing LDAP values are left untouched. Apache's profile upload limit drops back to the 256 KiB used by other member routes. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -1,6 +1,6 @@
|
||||
"""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
|
||||
Run only this module with config.test_settings. Credentials are
|
||||
ephemeral test data, not service credentials. Directory writes use mocks; LDAP
|
||||
reads additionally exercise ldap3's in-memory strategy.
|
||||
"""
|
||||
@@ -9,28 +9,22 @@ 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.urls import NoReverseMatch, 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,
|
||||
@@ -47,73 +41,49 @@ 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"],
|
||||
"cn": ["Alice 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 form(self, data=None):
|
||||
return ProfileForm(data={"cn": "Alice Example", "current_password": self.password, **(data or {})})
|
||||
|
||||
def test_friendly_labels_and_single_gamer_tag(self):
|
||||
form = self.form({"displayName": "AliceInGame"})
|
||||
def test_single_name_value_published_by_dex(self):
|
||||
form = self.form()
|
||||
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)
|
||||
self.assertEqual(form.fields["cn"].label, "Name")
|
||||
self.assertEqual(form.cleaned_data["cn"], ["Alice Example"])
|
||||
for name in ("First\nSecond", "x" * 257):
|
||||
with self.subTest(length=len(name)):
|
||||
invalid = self.form({"cn": name})
|
||||
self.assertFalse(invalid.is_valid())
|
||||
self.assertIn("cn", 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_only_dex_claim_attributes_are_editable(self):
|
||||
# Dex's LDAP connector publishes uid, cn and mail; mail has its own
|
||||
# verified flow and uid is immutable, so only cn is on this form.
|
||||
self.assertEqual(profiles.TEXT_ATTRIBUTES, ("cn",))
|
||||
self.assertEqual(set(ProfileForm.base_fields), {"cn", "current_password"})
|
||||
for name in ("displayName", "givenName", "sn", "jpegPhoto", "sshPublicKey", "postalAddress", "telephoneNumber", "description"):
|
||||
self.assertNotIn(name, profiles.READ_ATTRIBUTES)
|
||||
|
||||
def test_initial_uses_first_name_value_like_dex(self):
|
||||
self.assertEqual(profiles.initial(profile_values(cn=["Alice", "Alice Example"])), {"cn": "Alice"})
|
||||
self.assertEqual(profiles.initial(profile_values(cn=[])), {"cn": ""})
|
||||
|
||||
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(),
|
||||
]
|
||||
grouped = [*form.identity_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,
|
||||
@@ -134,114 +104,12 @@ class ProfileFormTests(SimpleTestCase):
|
||||
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})
|
||||
def test_name_required_even_when_only_unsafe_or_blank_text(self):
|
||||
for name in ("", " \n ", "\x00\u200b"):
|
||||
with self.subTest(name=repr(name)):
|
||||
form = self.form({"cn": name})
|
||||
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)
|
||||
self.assertIn("cn", form.errors)
|
||||
|
||||
def test_current_and_new_password_whitespace_preserved_not_rendered(self):
|
||||
new_password = " " + secrets.token_urlsafe(24) + " "
|
||||
@@ -288,7 +156,6 @@ class ProfileViewTests(TestCase):
|
||||
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)
|
||||
@@ -341,7 +208,7 @@ class ProfileViewTests(TestCase):
|
||||
|
||||
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)]))
|
||||
urls = (self.profile_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)
|
||||
@@ -358,6 +225,9 @@ class ProfileViewTests(TestCase):
|
||||
self.assertEqual(proof["account"], str(self.account.pk))
|
||||
self.assertEqual(proof["uuid"], "uuid-alice")
|
||||
self.assertContains(response, 'name="csrfmiddlewaretoken"')
|
||||
self.assertContains(response, 'name="cn" value="Alice Example"')
|
||||
for removed in ('name="sshPublicKey"', 'name="jpegPhoto"', 'name="displayName"', 'name="postalAddress"', "multipart/form-data"):
|
||||
self.assertNotContains(response, removed)
|
||||
self.assert_no_writes()
|
||||
|
||||
def test_csrf_required_for_each_mutating_action_and_email_confirmation(self):
|
||||
@@ -396,9 +266,9 @@ class ProfileViewTests(TestCase):
|
||||
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"))
|
||||
response = self.client.post(self.profile_url, self.profile_data(cn="", revision="original-revision"))
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("sn", response.context["form"].errors)
|
||||
self.assertIn("cn", response.context["form"].errors)
|
||||
self.assertEqual(response.context["revision"], "original-revision")
|
||||
self.assertNotContains(response, self.password)
|
||||
self.assert_no_writes()
|
||||
@@ -417,17 +287,16 @@ class ProfileViewTests(TestCase):
|
||||
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")
|
||||
data = self.profile_data(cn="New Name", displayName="NewGamerTag", sshPublicKey="ssh-ed25519 AAAA", 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["cn"], ["New Name"])
|
||||
self.assertEqual(cleaned["current_password"], self.password)
|
||||
for name in ("uid", "dn", "ldap_uuid", "userPassword", "mail"):
|
||||
for name in ("uid", "dn", "ldap_uuid", "userPassword", "mail", "displayName", "sshPublicKey"):
|
||||
self.assertNotIn(name, cleaned)
|
||||
self.assertNotIn(self.password, "\n".join(captured.output))
|
||||
self.account.refresh_from_db()
|
||||
@@ -612,31 +481,17 @@ class ProfileViewTests(TestCase):
|
||||
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_photo_endpoint_removed(self):
|
||||
with self.assertRaises(NoReverseMatch):
|
||||
reverse("members:profile_photo")
|
||||
with self.assertRaises(Resolver404):
|
||||
resolve(self.profile_url + "photo/")
|
||||
self.assertEqual(self.client.post(self.profile_url, {**self.profile_data(), "jpegPhoto": "x"}).status_code, 302)
|
||||
self.assertNotIn("jpegPhoto", self.update.call_args.args[1])
|
||||
|
||||
def test_get_only_endpoints_reject_post(self):
|
||||
token, _ = self.pending()
|
||||
for url in (self.photo_url, reverse("members:verify_profile_email", args=[token])):
|
||||
for url in (reverse("members:verify_profile_email", args=[token]),):
|
||||
self.assertEqual(self.client.post(url).status_code, 405)
|
||||
self.assert_no_writes()
|
||||
|
||||
@@ -651,7 +506,7 @@ class ProfileDirectoryTests(TestCase):
|
||||
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.strategy.add_entry(ALICE_DN, {**self.values, "objectClass": ["top", "person", "organizationalPerson", "inetOrgPerson"]})
|
||||
self.conn.bind()
|
||||
self.addCleanup(self.conn.unbind)
|
||||
# No network access can escape this module, even if a test regresses.
|
||||
@@ -676,7 +531,7 @@ class ProfileDirectoryTests(TestCase):
|
||||
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,
|
||||
"current_password": self.password,
|
||||
**changes,
|
||||
}
|
||||
|
||||
@@ -777,49 +632,31 @@ class ProfileDirectoryTests(TestCase):
|
||||
|
||||
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"])
|
||||
cleaned = self.cleaned(cn=["Changed"], displayName=["Ignored"], jpegPhoto=b"ignored", 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.assertEqual(conn.modify.call_args.args, (ALICE_DN, {"cn": [(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))
|
||||
profiles.update_profile(self.account, self.cleaned(cn=["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):
|
||||
def test_noop_does_not_write_and_extra_names_collapse_to_one(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)
|
||||
values = profile_values(cn=["Alice", "Alice Example"])
|
||||
conn, authorized = self.authorized(values=values)
|
||||
with patch("members.profiles._authorized", authorized):
|
||||
profiles.update_profile(self.account, self.cleaned(cn=["Alice"]), profiles.revision(self.account, values))
|
||||
self.assertEqual(conn.modify.call_args.args[1], {"cn": [(MODIFY_REPLACE, ["Alice"])]})
|
||||
|
||||
def test_revision_bound_to_account_uuid_and_all_read_values(self):
|
||||
valid = profiles.revision(self.account, self.values)
|
||||
@@ -830,9 +667,9 @@ class ProfileDirectoryTests(TestCase):
|
||||
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)
|
||||
profiles.update_profile(self.account, self.cleaned(cn=["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"])):
|
||||
for changed in (profile_values(cn=["Concurrent edit"]), profile_values(mail=["moved@example.org"]), 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)
|
||||
@@ -885,7 +722,7 @@ class ProfileDirectoryTests(TestCase):
|
||||
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))
|
||||
profiles.update_profile(self.account, self.cleaned(cn=["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):
|
||||
|
||||
Reference in New Issue
Block a user