"""MEDIA_ROOT enforcement. Every path a request names goes through here. Inputs and outputs must be absolute and stay inside the real MEDIA_ROOT after symlinks are resolved. A path that does not exist yet is checked through its nearest existing ancestor (realpath resolves the existing prefix). """ from __future__ import annotations import os import stat from .errors import InvalidPath, MediaError def _inside(root: str, candidate: str) -> bool: return candidate == root or candidate.startswith(root.rstrip(os.sep) + os.sep) def resolve_inside(root: str, requested: str) -> str: """Return the real absolute location of `requested`, or raise InvalidPath.""" if not isinstance(requested, str) or not requested or "\0" in requested: raise InvalidPath() if not os.path.isabs(requested): raise InvalidPath("path must be absolute") real_root = os.path.realpath(root) lexical = os.path.normpath(requested) # Reject lexical traversal even when it would land back inside the root. if ".." in requested.split("/"): raise InvalidPath() real = os.path.realpath(lexical) if not _inside(real_root, real): raise InvalidPath() return real def input_file(root: str, requested: str, max_bytes: int | None = None) -> str: real = resolve_inside(root, requested) try: info = os.stat(real) except FileNotFoundError as error: raise InvalidPath("input file does not exist") from error if not stat.S_ISREG(info.st_mode): raise InvalidPath("input is not a regular file") if max_bytes is not None and info.st_size > max_bytes: raise MediaError("too_large", "input exceeds MEDIA_MAX_INPUT_BYTES") return real def input_dir(root: str, requested: str) -> str: real = resolve_inside(root, requested) if not os.path.isdir(real): raise InvalidPath("directory does not exist") return real def output_file(root: str, requested: str) -> str: """Resolve an output file path and create its parent directories inside the root.""" real = resolve_inside(root, requested) if os.path.isdir(real): raise InvalidPath("output path is a directory") parent = os.path.dirname(real) os.makedirs(parent, exist_ok=True) # The parent may have been swapped for a symlink in between; check again. resolve_inside(root, parent) return real def output_dir(root: str, requested: str) -> str: real = resolve_inside(root, requested) os.makedirs(real, exist_ok=True) resolve_inside(root, real) if not os.path.isdir(real): raise InvalidPath("output directory is not a directory") return real