Profile

membership/members/profile_forms.py

@@ -0,0 +1,236 @@
+"""Validation-only forms for the shared LDAP profile; no directory writes."""
+
+import unicodedata
+import warnings
+from io import BytesIO
+from urllib.parse import urlsplit
+
+from cryptography.exceptions import UnsupportedAlgorithm
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import ec, ed25519, rsa
+from django import forms
+from django.core.exceptions import ValidationError
+from django.core.validators import URLValidator
+
+from .forms import PasswordForm
+
+
+MAX_INPUT_LENGTH = 32768
+MAX_VALUES = 16
+MAX_PHOTO_BYTES = 1024 * 1024
+
+
+def _safe_text(value: str) -> str:
+    # Normalize browser line endings before removing unsafe control characters.
+    value = value.replace("\r\n", "\n").replace("\r", "\n").replace("\t", " ")
+    return "".join(
+        char for char in value
+        if char == "\n" or unicodedata.category(char) not in {"Cc", "Cf", "Cs"}
+    )
+
+
+class ProfileValuesField(forms.CharField):
+    """Text input becomes an ordered, deduplicated list of LDAP strings."""
+
+    def __init__(self, *, value_length: int = 2048, postal: bool = False, max_values: int = MAX_VALUES, **kwargs):
+        self.value_length = value_length
+        self.postal = postal
+        self.max_values = max_values
+        kwargs.setdefault("required", False)
+        kwargs.setdefault("help_text", "One value per line. Up to 16 values; leave blank to clear.")
+        super().__init__(
+            max_length=MAX_INPUT_LENGTH,
+            widget=forms.Textarea(attrs={"rows": 3, "maxlength": MAX_INPUT_LENGTH}),
+            **kwargs,
+        )
+
+    def clean(self, value) -> list[str]:
+        if value is not None and len(str(value)) > MAX_INPUT_LENGTH:
+            raise ValidationError("Enter no more than 32,768 characters in this field.")
+        text = super().clean(_safe_text(str(value)) if value is not None else "")
+        if self.postal:
+            entries = []
+            lines = []
+            for line in text.split("\n") + [""]:
+                if line.strip():
+                    # RFC 4517 Postal Address escapes literal '$' and '\'.
+                    lines.append(line.strip().replace("\\", "\\5c").replace("$", "\\24"))
+                elif lines:
+                    entries.append("$".join(lines))
+                    lines = []
+        else:
+            entries = [line.strip() for line in text.split("\n") if line.strip()]
+        if len(entries) > self.max_values:
+            raise ValidationError(f"Enter no more than {self.max_values} value(s).")
+        if any(len(entry) > self.value_length for entry in entries):
+            raise ValidationError(f"Each value must be at most {self.value_length:,} characters.")
+        if self.required and not entries:
+            raise ValidationError("Enter at least one value.")
+        return list(dict.fromkeys(entries))
+
+
+def _current_password_field() -> forms.CharField:
+    return forms.CharField(
+        label="Current password", max_length=1024, strip=False,
+        widget=forms.PasswordInput(
+            render_value=False, attrs={"autocomplete": "current-password"}
+        ),
+        help_text="Enter your current eom.dev password to authorize this change.",
+    )
+
+
+class CurrentPasswordForm(forms.Form):
+    current_password = _current_password_field()
+
+    def __init__(self, *args, **kwargs):
+        kwargs.setdefault("auto_id", "id_confirmation_%s")
+        super().__init__(*args, **kwargs)
+
+
+class ProfileForm(CurrentPasswordForm):
+    description = ProfileValuesField(label="About you", value_length=4096)
+    displayName = ProfileValuesField(label="Gamer tag", max_values=1, help_text="One gamer tag, reused by gaming services. Leave blank to clear.")
+    givenName = ProfileValuesField(label="First name")
+    l = ProfileValuesField(label="City/town")
+    labeledURI = ProfileValuesField(label="Websites", help_text="One HTTP or HTTPS URL per line, optionally followed by a space and a label.")
+    mobile = ProfileValuesField(label="Mobile numbers")
+    o = ProfileValuesField(label="Organizations")
+    postalAddress = ProfileValuesField(label="Postal addresses", postal=True, help_text="Use a blank line between addresses. Line breaks within each address are preserved.")
+    postalCode = ProfileValuesField(label="Postal codes")
+    roomNumber = ProfileValuesField(label="Room numbers")
+    sn = ProfileValuesField(label="Last name", required=True, help_text="Enter at least one last name. One value per line.")
+    sshPublicKey = ProfileValuesField(label="SSH public keys", help_text="One public key per line: Ed25519, RSA (at least 2048 bits), or ECDSA NIST P-256/P-384/P-521. No private keys or authorized_keys options.")
+    st = ProfileValuesField(label="State/province")
+    street = ProfileValuesField(label="Street addresses")
+    telephoneNumber = ProfileValuesField(label="Phone numbers")
+    title = ProfileValuesField(label="Job titles")
+    jpegPhoto = forms.FileField(
+        label="Profile photo", required=False,
+        widget=forms.FileInput(attrs={"accept": "image/jpeg"}),
+        help_text="JPEG only, up to 1 MiB and 4096 pixels per side (16 million pixels total). Resized to fit 512 × 512; metadata is removed.",
+    )
+    remove_photo = forms.BooleanField(label="Remove current photo", required=False)
+
+    def __init__(self, *args, **kwargs):
+        # Initial text values are formatted by the caller, including postal $ decoding.
+        kwargs.setdefault("auto_id", "id_profile_%s")
+        super().__init__(*args, **kwargs)
+
+    def identity_fields(self):
+        return [self[name] for name in ("givenName", "sn", "displayName", "description", "o", "title")]
+
+    def contact_fields(self):
+        return [self[name] for name in ("labeledURI", "telephoneNumber", "mobile")]
+
+    def address_fields(self):
+        return [self[name] for name in ("postalAddress", "street", "roomNumber", "l", "st", "postalCode")]
+
+    def authorization_fields(self):
+        return [self["current_password"]]
+
+    def clean_labeledURI(self) -> list[str]:
+        entries = self.cleaned_data["labeledURI"]
+        validator = URLValidator(schemes=["https", "http"])
+        for entry in entries:
+            url = entry.partition(" ")[0]
+            try:
+                validator(url)
+                parsed = urlsplit(url)
+                if parsed.username is not None or parsed.password is not None or "\\" in url:
+                    raise ValueError("Credentials and backslashes are not allowed.")
+                if any(char.isspace() for char in url):
+                    raise ValueError("Whitespace is not allowed in URLs.")
+            except (ValidationError, ValueError):
+                raise ValidationError("Use an HTTP or HTTPS URL without credentials, optionally followed by a label.") from None
+        return entries
+
+    def clean_sshPublicKey(self) -> list[str]:
+        supported = {"ssh-ed25519", "ssh-rsa", "ecdsa-sha2-nistp256", "ecdsa-sha2-nistp384", "ecdsa-sha2-nistp521"}
+        entries = []
+        seen = set()
+        for value in self.cleaned_data["sshPublicKey"]:
+            parts = value.split()
+            if len(parts) < 2 or parts[0] not in supported:
+                raise ValidationError("Enter a supported SSH public key without options or private-key material.")
+            try:
+                key = serialization.load_ssh_public_key(value.encode("utf-8"))
+            except (ValueError, TypeError, UnsupportedAlgorithm):
+                raise ValidationError("Enter a valid supported SSH public key.") from None
+            valid = (
+                isinstance(key, ed25519.Ed25519PublicKey)
+                or isinstance(key, rsa.RSAPublicKey) and key.key_size >= 2048
+                or isinstance(key, ec.EllipticCurvePublicKey)
+                and isinstance(key.curve, (ec.SECP256R1, ec.SECP384R1, ec.SECP521R1))
+            )
+            if not valid:
+                raise ValidationError("Use Ed25519, RSA of at least 2048 bits, or ECDSA NIST P-256/P-384/P-521.")
+            identity = tuple(parts[:2])
+            if identity not in seen:
+                seen.add(identity)
+                entries.append(value)
+        return entries
+
+    def clean_jpegPhoto(self) -> bytes | None:
+        upload = self.cleaned_data.get("jpegPhoto")
+        if upload is None:
+            return None
+        if upload.size > MAX_PHOTO_BYTES:
+            raise ValidationError("Choose a JPEG no larger than 1 MiB.")
+        raw = upload.read(MAX_PHOTO_BYTES + 1)
+        if len(raw) > MAX_PHOTO_BYTES:
+            raise ValidationError("Choose a JPEG no larger than 1 MiB.")
+
+        # Import only on upload; Pillow is supplied by the application's dependencies.
+        from PIL import Image, ImageOps, UnidentifiedImageError
+
+        try:
+            with warnings.catch_warnings():
+                warnings.simplefilter("error", Image.DecompressionBombWarning)
+                with Image.open(BytesIO(raw), formats=["JPEG"]) as image:
+                    width, height = image.size
+                    if image.format != "JPEG" or not (0 < width <= 4096 and 0 < height <= 4096) or width * height > 16_000_000:
+                        raise ValidationError("Choose a JPEG up to 4096 pixels per side and 16 million pixels total.")
+                    image.verify()
+                with Image.open(BytesIO(raw), formats=["JPEG"]) as image:
+                    image.load()
+                    rgb = ImageOps.exif_transpose(image).convert("RGB")
+                    rgb.thumbnail((512, 512), Image.Resampling.LANCZOS)
+                    # A fresh pixel-only image prevents EXIF/ICC/comment metadata retention.
+                    clean = Image.new("RGB", rgb.size)
+                    clean.paste(rgb)
+                    output = BytesIO()
+                    clean.save(output, format="JPEG", quality=85)
+                    return output.getvalue()
+        except (UnidentifiedImageError, OSError, ValueError, SyntaxError, Image.DecompressionBombWarning, Image.DecompressionBombError):
+            raise ValidationError("Choose a valid, safely sized JPEG image.") from None
+
+    def clean(self):
+        data = super().clean()
+        if data.get("jpegPhoto") is not None and data.get("remove_photo"):
+            self.add_error("remove_photo", "Choose either a new photo or removal, not both.")
+        return data
+
+
+class ProfileEmailForm(CurrentPasswordForm):
+    email = forms.EmailField(label="New email address", max_length=254, widget=forms.EmailInput(attrs={"autocomplete": "email"}))
+    field_order = ["email", "current_password"]
+
+    def __init__(self, *args, **kwargs):
+        kwargs.setdefault("auto_id", "id_email_%s")
+        super().__init__(*args, **kwargs)
+
+    def clean_email(self) -> str:
+        return self.cleaned_data["email"].casefold()
+
+
+class ProfilePasswordForm(PasswordForm, CurrentPasswordForm):
+    field_order = ["current_password", "password", "confirm_password"]
+
+    def __init__(self, *args, **kwargs):
+        kwargs.setdefault("auto_id", "id_password_%s")
+        super().__init__(*args, **kwargs)
+        for name in ("password", "confirm_password"):
+            self.fields[name].strip = False
+            self.fields[name].widget = forms.PasswordInput(
+                render_value=False, attrs={"autocomplete": "new-password"}
+            )