237 lines
11 KiB
Python
Executable File
237 lines
11 KiB
Python
Executable File
"""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"}
|
||
)
|