#!/usr/bin/env python3 """Small dependency-free HTTP server for private image submissions.""" from __future__ import annotations import argparse import json import os import re import secrets import sys from email.parser import BytesParser from email.policy import default from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import unquote, urlsplit MAX_FILE_BYTES = 10 * 1024 * 1024 MAX_REQUEST_BYTES = MAX_FILE_BYTES + 1024 * 1024 STATIC_DIR = Path(__file__).resolve().parent / "static" STATIC_FILES = { "/styles.css": ("styles.css", "text/css; charset=utf-8"), "/app.js": ("app.js", "text/javascript; charset=utf-8"), } TARGET_PATTERN = re.compile(r"^/([A-Za-z0-9][A-Za-z0-9_-]{0,63})/?$") RESERVED_TARGETS = {"healthz"} def target_from_path(path: str) -> str | None: """Return a safe single-segment intake target, if the path contains one.""" match = TARGET_PATTERN.fullmatch(unquote(path)) if match is None or match.group(1) in RESERVED_TARGETS: return None return match.group(1) def detect_image_type(data: bytes) -> tuple[str, str] | None: """Return a safe extension and MIME type based on file signatures.""" if data.startswith(b"\xff\xd8\xff"): return ".jpg", "image/jpeg" if data.startswith(b"\x89PNG\r\n\x1a\n"): return ".png", "image/png" if len(data) >= 12 and data[4:8] == b"ftyp": brands = {data[8:12]} brands.update(data[index : index + 4] for index in range(16, min(len(data), 64), 4)) if brands & {b"heic", b"heix", b"hevc", b"hevx", b"heim", b"heis", b"mif1", b"msf1"}: return ".heic", "image/heic" return None def parse_single_upload(content_type: str, body: bytes) -> tuple[str, bytes]: """Parse one multipart field named image and reject ambiguous requests.""" if not content_type.lower().startswith("multipart/form-data;"): raise ValueError("Expected a multipart form upload.") message = BytesParser(policy=default).parsebytes( b"Content-Type: " + content_type.encode("latin-1") + b"\r\nMIME-Version: 1.0\r\n\r\n" + body ) if not message.is_multipart(): raise ValueError("The upload form is malformed.") file_parts = [] for part in message.iter_parts(): if part.get_content_disposition() != "form-data": continue filename = part.get_filename() if filename is not None: file_parts.append(part) if len(file_parts) != 1 or file_parts[0].get_param("name", header="content-disposition") != "image": raise ValueError("Submit exactly one image.") part = file_parts[0] filename = Path(part.get_filename() or "image").name payload = part.get_payload(decode=True) if payload is None or not payload: raise ValueError("The selected image is empty.") if len(payload) > MAX_FILE_BYTES: raise OverflowError("The image must be 10 MB or smaller.") return filename, payload class UploadHandler(BaseHTTPRequestHandler): server_version = "BlindUploader/1.0" @property def upload_dir(self) -> Path: return self.server.upload_dir # type: ignore[attr-defined] def do_GET(self) -> None: path = urlsplit(self.path).path if path == "/healthz": self.send_json(HTTPStatus.OK, {"status": "ok"}) return static_file = STATIC_FILES.get(path) if static_file is not None: filename, content_type = static_file self.send_static(filename, content_type) return if path == "/": self.send_error_response(HTTPStatus.IM_A_TEAPOT, "A valid upload target is required.") return if target_from_path(path) is None: self.send_error_response(HTTPStatus.NOT_FOUND, "A valid upload target is required.") return self.send_static("index.html", "text/html; charset=utf-8") def send_static(self, filename: str, content_type: str) -> None: data = (STATIC_DIR / filename).read_bytes() self.send_response(HTTPStatus.OK) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(data))) self.send_header("Cache-Control", "no-store") self.send_security_headers() self.end_headers() self.wfile.write(data) def do_POST(self) -> None: path = urlsplit(self.path).path if path == "/": self.send_error_response(HTTPStatus.IM_A_TEAPOT, "A valid upload target is required.") return target = target_from_path(path) if target is None: self.send_error_response(HTTPStatus.NOT_FOUND, "A valid upload target is required.") return try: content_length = int(self.headers.get("Content-Length", "")) except ValueError: content_length = -1 if content_length < 0: self.send_error_response(HTTPStatus.LENGTH_REQUIRED, "A Content-Length header is required.") return if content_length > MAX_REQUEST_BYTES: self.send_error_response(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, "The image must be 10 MB or smaller.") return body = self.rfile.read(content_length) try: _original_name, payload = parse_single_upload(self.headers.get("Content-Type", ""), body) detected = detect_image_type(payload) if detected is None: raise TypeError("Only JPG, PNG, and HEIC images are accepted.") extension, _mime_type = detected storage_id = secrets.token_hex(20) receipt = storage_id[:12] target_dir = self.upload_dir / target target_dir.mkdir(mode=0o700, exist_ok=True) target_dir.chmod(0o700) destination = target_dir / f"{storage_id}{extension}" descriptor = os.open(destination, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(descriptor, "wb") as output: output.write(payload) except OverflowError as error: self.send_error_response(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, str(error)) return except (TypeError, ValueError) as error: self.send_error_response(HTTPStatus.BAD_REQUEST, str(error)) return except OSError: self.log_error("Unable to persist an uploaded image") self.send_error_response(HTTPStatus.INTERNAL_SERVER_ERROR, "The image could not be stored. Please try again.") return self.send_json( HTTPStatus.CREATED, {"ok": True, "message": "Your image was submitted privately.", "receipt": receipt}, ) def do_HEAD(self) -> None: self.send_error_response(HTTPStatus.METHOD_NOT_ALLOWED, "Method not allowed.") def send_json(self, status: HTTPStatus, payload: dict[str, object]) -> None: body = json.dumps(payload).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.send_security_headers() self.end_headers() self.wfile.write(body) def send_error_response(self, status: HTTPStatus, message: str) -> None: self.send_json(status, {"ok": False, "message": message}) def send_security_headers(self) -> None: self.send_header("X-Content-Type-Options", "nosniff") self.send_header("X-Frame-Options", "DENY") self.send_header("X-Robots-Tag", "noindex, nofollow, noarchive") self.send_header("Referrer-Policy", "no-referrer") self.send_header( "Content-Security-Policy", "default-src 'self'; img-src 'self' blob:; script-src 'self'; style-src 'self'; " "base-uri 'none'; frame-ancestors 'none'; form-action 'self'", ) def log_message(self, format: str, *args: object) -> None: sys.stderr.write("%s - %s\n" % (self.address_string(), format % args)) def make_server(host: str, port: int, upload_dir: Path) -> ThreadingHTTPServer: upload_dir.mkdir(mode=0o700, parents=True, exist_ok=True) try: upload_dir.chmod(0o700) except OSError: pass server = ThreadingHTTPServer((host, port), UploadHandler) server.upload_dir = upload_dir # type: ignore[attr-defined] return server def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--host", default=os.environ.get("HOST", "127.0.0.1")) parser.add_argument("--port", type=int, default=int(os.environ.get("PORT", "8000"))) parser.add_argument( "--upload-dir", type=Path, default=Path(os.environ.get("UPLOAD_DIR", Path(__file__).resolve().parent / "data" / "uploads")), ) args = parser.parse_args() server = make_server(args.host, args.port, args.upload_dir.resolve()) print(f"Blind Uploader listening on http://{args.host}:{args.port}") print(f"Private uploads are stored in {server.upload_dir}") # type: ignore[attr-defined] try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() if __name__ == "__main__": main()