"""Shared upload validation and tenant-scoped storage paths.""" import secrets import uuid from pathlib import PurePosixPath from django.conf import settings from django.utils.deconstruct import deconstructible from rest_framework.exceptions import ValidationError # Length of the random filename stem, in bytes of entropy before base64. # 16 bytes = 128 bits, rendered as 22 url-safe characters. _FILENAME_ENTROPY_BYTES = 16 # Where files land when their owning organization cannot be determined at save # time (see tenant_upload_path). Deliberately outside every org// prefix so # such files are easy to find and sweep, and can never be mistaken for a # particular tenant's data. UNASSIGNED_PREFIX = "_unassigned" def _random_stem() -> str: """CSPRNG filename stem. Deliberately NOT uuid7, which this project uses for primary keys: uuid7 encodes a 48-bit millisecond timestamp in its leading bits, so two files uploaded moments apart share a long prefix and only the trailing random bits differ. That is desirable for a B-tree PK and wrong for a filename, which may be the only secret standing in front of a presigned URL. """ return secrets.token_urlsafe(_FILENAME_ENTROPY_BYTES) def _id_segment(value) -> str: """A primary key rendered as canonical dashless hex. The apps in this project do not agree on what a uuid7 pk *is*: apps.core returns a uuid.UUID, while companies/projects/events and others return an already-dashless hex string. So `str(pk)` yields the dashed form for a row read back from the database and the dashless form for one still in memory — the same tenant, two different prefixes, silently splitting its files across two directories and defeating any prefix-scoped R2 credential. Normalizing here rather than unifying the uuid7 helpers: those feed primary keys across 22 apps, and changing them is a data-migration-shaped problem, not an upload-path one. """ try: return uuid.UUID(str(value)).hex except (ValueError, AttributeError, TypeError): # A non-UUID pk (or None) still needs a usable, traversal-free segment. return PurePosixPath(str(value).replace("\\", "/")).name or UNASSIGNED_PREFIX def _safe_suffix(filename: str) -> str: """The original extension, lowercased and length-capped, or "". Kept so Content-Type sniffing and browser downloads behave, which matters more in practice than hiding the file type — the stem is already random. Taken via PurePosixPath so a crafted name cannot inject path separators. """ suffix = PurePosixPath(filename.replace("\\", "/")).suffix.lower() if not suffix or len(suffix) > 10 or not suffix[1:].isalnum(): return "" return suffix @deconstructible class tenant_upload_path: """upload_to callable producing org///. The organization id is the leftmost segment so the tenant boundary is the first thing in every key. That is what makes prefix-scoped R2 credentials possible later, and makes a cross-tenant bug obvious in access logs rather than invisible inside a shared flat directory. Path structure is defence in depth, not the access control: reads are authorized by apps.core.media_views.serve_media, which resolves each file's owner. Prefixes matter because a presigned R2 URL bypasses that view entirely, so an unguessable key is the only thing limiting a leaked URL. Only affects NEW uploads. Rows written before this was applied keep their original flat keys and stay protected by the authorizing view alone. A class rather than a closure-returning factory because Django serializes upload_to into migrations and cannot reference a nested function; a @deconstructible instance serializes as tenant_upload_path(""). """ def __init__(self, category: str): self.category = category def __eq__(self, other): # Without this, makemigrations sees a new instance every run and emits # an endless stream of no-op AlterField migrations. return isinstance(other, tenant_upload_path) and other.category == self.category def __hash__(self): return hash((type(self).__name__, self.category)) def __call__(self, instance, filename): # Imported lazily: tenancy reaches Django models, and models.py modules # instantiate this at class-definition time. from apps.core.tenancy import organization_of organization = organization_of(instance) # An organization is not always resolvable at save time — a parent FK may # still be unset when the file is written on a fresh create(). Falling # back keeps the upload working; raising here would turn an ordering # detail into a user-facing 500. org_segment = _id_segment(organization.id) if organization is not None else UNASSIGNED_PREFIX return f"org/{org_segment}/{self.category}/{_random_stem()}{_safe_suffix(filename)}" @deconstructible class own_organization_upload_path(tenant_upload_path): """upload_to for fields on Organization itself, keyed by its own id. Organization has no `organization` field to walk, so the generic callable would send every tenant's branding to _unassigned/. The row *is* the tenant, so its own pk is the prefix. An unsaved Organization already has a pk: the model defaults it to uuid7() at instantiation rather than leaving the database to assign it, so the prefix is stable even when the logo is set on the very first save. """ def __call__(self, instance, filename): return f"org/{_id_segment(instance.pk)}/{self.category}/{_random_stem()}{_safe_suffix(filename)}" @deconstructible class user_upload_path(tenant_upload_path): """upload_to for fields owned by a User rather than an organization. Deliberately outside every org/ prefix. A user may belong to several organizations, so filing their avatar under whichever membership happens to be first would strand it there — serve_media would then refuse the user their own avatar while they had a different organization active. `users//` still segments per subject, so the key is not a flat shared directory even though it is not tenant-scoped. """ def __call__(self, instance, filename): return f"users/{_id_segment(instance.pk)}/{self.category}/{_random_stem()}{_safe_suffix(filename)}" def validate_upload(uploaded_file) -> None: max_size = getattr(settings, "ATTACHMENT_MAX_FILE_SIZE", 50 * 1024 * 1024) if uploaded_file.size > max_size: raise ValidationError( f"file too large ({uploaded_file.size} bytes); max {max_size}" ) allowed = getattr(settings, "ATTACHMENT_ALLOWED_MIME_PREFIXES", ()) if allowed: mime = (getattr(uploaded_file, "content_type", "") or "").lower() if not any(mime.startswith(p) for p in allowed): raise ValidationError(f"mime type '{mime}' not allowed")