"""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 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 register_heif_opener() Image.MAX_IMAGE_PIXELS = 60_000_000 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((2560, 2560), 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) clean.save(destination, "JPEG", quality=88, optimize=True) def create_app(data_dir=None, token=None): app = Flask(__name__) app.config.update(MAX_CONTENT_LENGTH=100 * 1024 * 1024, MAX_FORM_MEMORY_SIZE=64 * 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): 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()) 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())["photos"]: state["status"] = "published" except (OSError, ValueError, KeyError): pass # A successful push is not yet a successful deployment. return dict(id=job, count=manifest["count"], **state, status_url=f"/api/photos/{job}", url=f"/art#photo-{job}") @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") def upload(): photos = request.files.getlist("photos") if not 1 <= len(photos) <= 20: abort(400, "Send between 1 and 20 files in the photos form field") caption = request.form.get("caption", "").strip() if len(caption) > 4000: abort(400, "Caption must be at most 4000 characters") # Ordered byte hashes + caption make retries idempotent, even across restarts. digest = hashlib.sha256(caption.encode()) temporary = Path(tempfile.mkdtemp(dir=data / "temporary")) try: (temporary / "originals").mkdir() (temporary / "images").mkdir() for index, photo in enumerate(photos): raw = 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") job = digest.hexdigest() date = datetime.now(timezone.utc).isoformat() manifest = {"id": job, "date": date, "caption": caption, "count": len(photos)} atomic_json(temporary / "manifest.json", manifest) content = ["---", "layout: post", f'title: "photo-{job}"', f'date: "{date}"', "categories: art", "---", "", f'
', ""] for index in range(len(photos)): content.append(f'

') if caption: content.append('' + public_text(caption).replace("\n", "
") + "") (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), 202 if created else 200 finally: shutil.rmtree(temporary, ignore_errors=True) return app