"""Authenticated photo intake. Only completed directories are visible to the builder.""" import hashlib import hmac import html import io import json import os from pathlib import Path import re import shutil import tempfile from datetime import datetime, timezone import warnings import zipfile from flask import Flask, abort, jsonify, request from PIL import Image, ImageCms, ImageOps, UnidentifiedImageError from pillow_heif import register_heif_opener from werkzeug.exceptions import HTTPException from posts import parse_upload register_heif_opener() Image.MAX_IMAGE_PIXELS = 60_000_000 GALLERY_MAX_EDGE = 1600 GALLERY_MAX_BYTES = 300 * 1024 warnings.simplefilter("error", Image.DecompressionBombWarning) def atomic_json(path, value): fd, temporary = tempfile.mkstemp(dir=path.parent) try: with os.fdopen(fd, "w") as output: json.dump(value, output) output.flush() os.fsync(output.fileno()) os.replace(temporary, path) finally: Path(temporary).unlink(missing_ok=True) def public_text(text): # Escape both HTML and Liquid: Jekyll evaluates Liquid before Markdown. return html.escape(text, quote=True).replace("{", "{").replace("}", "}") def web_image(raw, destination): with Image.open(io.BytesIO(raw)) as original: if original.format not in {"JPEG", "PNG", "WEBP", "HEIF"}: raise ValueError("Use JPEG, PNG, WebP, or HEIC photos") image = ImageOps.exif_transpose(original) image.thumbnail((GALLERY_MAX_EDGE, GALLERY_MAX_EDGE), Image.Resampling.LANCZOS) if image.info.get("icc_profile"): image = ImageCms.profileToProfile( image, ImageCms.ImageCmsProfile(io.BytesIO(image.info["icc_profile"])), ImageCms.createProfile("sRGB"), outputMode="RGB", ) if image.mode in {"RGBA", "LA"} or "transparency" in image.info: rgba = image.convert("RGBA") background = Image.new("RGB", image.size, "white") background.paste(rgba, mask=rgba.getchannel("A")) image = background else: image = image.convert("RGB") # Copy pixels only; no EXIF, GPS, XMP, comments, or original profile. clean = Image.new("RGB", image.size) clean.paste(image) # Keep a useful display size and moderate JPEG quality, even for grainy photos. # If quality alone cannot meet the budget, reduce dimensions instead of # introducing severe compression artifacts. Originals remain untouched. while True: for quality in (82, 78, 74): output = io.BytesIO() clean.save(output, "JPEG", quality=quality, optimize=True, progressive=True) if output.tell() <= GALLERY_MAX_BYTES: Path(destination).write_bytes(output.getvalue()) return clean.thumbnail((max(1, int(clean.width * 0.85)), max(1, int(clean.height * 0.85))), Image.Resampling.LANCZOS) def create_app(data_dir=None, token=None): app = Flask(__name__) app.config.update(MAX_CONTENT_LENGTH=100 * 1024 * 1024, # Must exceed Werkzeug's 64 KiB multipart read buffer plus headers. MAX_FORM_MEMORY_SIZE=512 * 1024, MAX_FORM_PARTS=40) data = Path(data_dir or os.environ.get("PHOTO_DATA", "/data")) if token is None: token = Path(os.environ.get("PHOTO_TOKEN_FILE", "/run/secrets/photo_token")).read_text().strip() if len(token) < 32 or not token.isascii(): raise ValueError("Photo token must be at least 32 ASCII characters") expected_auth = ("Bearer " + token).encode() for directory in ("submissions", "status", "temporary"): (data / directory).mkdir(parents=True, exist_ok=True) @app.before_request def authenticate(): if request.path != "/healthz": auth = request.headers.get("Authorization", "").encode() if not hmac.compare_digest(auth, expected_auth): abort(401, "Invalid upload token") @app.errorhandler(HTTPException) def error_response(error): return jsonify(error=error.description), error.code @app.get("/healthz") def health(): return {"status": "ok"} def response_for(job, kind="photo"): manifest_path = data / "submissions" / job / "manifest.json" if not re.fullmatch(r"[a-f0-9]{64}", job) or not manifest_path.is_file(): abort(404) manifest = json.loads(manifest_path.read_text()) if manifest.get("kind", "photo") != kind: abort(404) status_path = data / "status" / (job + ".json") state = json.loads(status_path.read_text()) if status_path.exists() else {"status": "queued"} if state["status"] == "pushed": deployed = Path(os.environ.get("PHOTO_DEPLOYMENT_FILE", "/published/current/photo-publication.json")) try: if job in json.loads(deployed.read_text()).get(kind + "s", []): state["status"] = "published" except (OSError, ValueError, KeyError): pass # A successful push is not yet a successful deployment. page = "art" if kind == "photo" else "snippets/" result = dict(id=job, count=manifest["count"], status_url=f"/api/{kind}s/{job}", url=None if kind == "post" else f"/{page}#{kind}-{job}") result.update(state) return result @app.get("/api/posts/") def post_status(job): return response_for(job, "post") @app.post("/api/posts//retry") def post_retry(job): current = response_for(job, "post") if current["status"] == "failed": atomic_json(data / "status" / (job + ".json"), {"status": "queued"}) return response_for(job, "post"), 202 @app.post("/api/posts") def upload_post(): files = request.files.getlist("file") if len(files) != 1: abort(400, "Send one Markdown file in the file form field") raw = files[0].read(1024 * 1024 + 1) try: incoming = parse_upload(files[0].filename, raw) except ValueError as error: abort(400, str(error)) job = hashlib.sha256(b"post\0" + incoming["key"].encode() + b"\0" + raw).hexdigest() temporary = Path(tempfile.mkdtemp(dir=data / "temporary")) try: atomic_json(temporary / "post.json", incoming) (temporary / "original.md").write_bytes(raw) atomic_json(temporary / "manifest.json", dict(id=job, kind="post", count=1, date=datetime.now(timezone.utc).isoformat())) try: temporary.rename(data / "submissions" / job) created = True except OSError: if not (data / "submissions" / job / "manifest.json").is_file(): raise created = False return response_for(job, "post"), 202 if created else 200 finally: shutil.rmtree(temporary, ignore_errors=True) @app.get("/api/snippets/") def snippet_status(job): return response_for(job, "snippet") @app.post("/api/snippets//retry") def snippet_retry(job): current = response_for(job, "snippet") if current["status"] == "failed": atomic_json(data / "status" / (job + ".json"), {"status": "queued"}) return response_for(job, "snippet"), 202 @app.get("/api/photos/") def status(job): return response_for(job) @app.post("/api/photos//retry") def retry(job): current = response_for(job) if current["status"] == "failed": atomic_json(data / "status" / (job + ".json"), {"status": "queued"}) return response_for(job), 202 @app.post("/api/photos") @app.post("/api/snippets") def upload(): kind = "snippet" if request.path == "/api/snippets" else "photo" photos = request.files.getlist("photos") if kind == "snippet" and "photo" in request.files: single = request.files.getlist("photo") if len(single) != 1 or photos: abort(400, "Send one photo, photos, or an archive; do not mix image fields") photos = single archives = request.files.getlist("archive") if archives: if photos or len(archives) != 1: abort(400, "Send either photos or one ZIP archive, not both") try: # Python 3.10's SpooledTemporaryFile lacks ZipFile's seekable API. with zipfile.ZipFile(io.BytesIO(archives[0].read())) as archive: members = [item for item in archive.infolist() if not item.is_dir() and not item.filename.startswith("__MACOSX/") and Path(item.filename).name != ".DS_Store"] if not 1 <= len(members) <= 20: abort(400, "The ZIP must contain between 1 and 20 images") if sum(item.file_size for item in members) > 100 * 1024 * 1024: abort(413, "Uncompressed ZIP images must be at most 100 MiB") # Read members only; never extract archive paths onto the filesystem. photos = [archive.read(item) for item in members] except (zipfile.BadZipFile, RuntimeError, NotImplementedError, OSError): abort(400, "Send a valid, unencrypted ZIP archive of images") if not (0 if kind == "snippet" else 1) <= len(photos) <= 20: abort(400, "Send between 1 and 20 files in the photos form field") text = request.form.get("text", "").strip() if kind == "snippet" else "" if len(text) > 20000: abort(400, "Snippet text must be at most 20000 characters") if kind == "snippet" and not text and not photos: abort(400, "Send text or images for the snippet") caption = request.form.get("caption", "").strip() if len(caption) > 4000: abort(400, "Caption must be at most 4000 characters") highlight = request.form.get("highlight", "").strip() if len(highlight) > 200: abort(400, "Highlight name must be at most 200 characters") # Ordered byte hashes + caption make retries idempotent, even across restarts. digest = hashlib.sha256(caption.encode()) if kind == "snippet": digest.update(b"\0snippet\0" + text.encode()) temporary = Path(tempfile.mkdtemp(dir=data / "temporary")) try: (temporary / "originals").mkdir() (temporary / "images").mkdir() for index, photo in enumerate(photos): raw = photo if isinstance(photo, bytes) else photo.read() digest.update(hashlib.sha256(raw).digest()) (temporary / "originals" / str(index)).write_bytes(raw) try: web_image(raw, temporary / "images" / f"{index:02}.jpg") except (UnidentifiedImageError, OSError, ValueError, SyntaxError, Image.DecompressionBombError, Image.DecompressionBombWarning, ImageCms.PyCMSError): abort(400, f"Photo {index + 1} is unsupported, damaged, or exceeds 60 megapixels") # Preserve existing IDs when omitted; named highlights participate in deduplication. if highlight: digest.update(b"\0highlight\0" + highlight.encode()) job = digest.hexdigest() date = datetime.now(timezone.utc).isoformat() manifest = {"id": job, "date": date, "caption": caption, "highlight": highlight, "count": len(photos)} if kind == "snippet": manifest.update(kind=kind, text=text) atomic_json(temporary / "manifest.json", manifest) content = ["---", "layout: post", f'title: "photo-{job}"', f'date: "{date}"', "categories: art"] if highlight: content.append(f"anchor: {json.dumps(highlight)}") content.extend(["---", ""]) # The gallery already renders an anchor before the first photo for highlights. if not highlight: content.extend([f'
', ""]) for index in range(len(photos)): content.append(f'

') if caption: content.append('' + public_text(caption).replace("\n", "
") + "") if kind == "snippet": content = [f'
', ""] for paragraph in (text, caption): if paragraph: safe = public_text(paragraph) # Link standalone URL lines; all other shared text stays literal. lines = [f'{line}' if re.fullmatch(r"https?://[^\s<>]+", line) else line for line in safe.splitlines()] content.append('

' + '
\n'.join(lines) + '

') for index in range(len(photos)): content.append(f'

') content.extend(["", "----", ""]) (temporary / "entry.md").write_text("\n".join(content) + "\n") # Atomic rename handles simultaneous identical submissions without replacing one. try: temporary.rename(data / "submissions" / job) created = True except OSError: if not (data / "submissions" / job / "manifest.json").is_file(): raise created = False return response_for(job, kind), 202 if created else 200 finally: shutil.rmtree(temporary, ignore_errors=True) return app