feat: add private image upload serviceHEADmain

1 files changed, 238 insertions, 0 deletions

diff --git a/server.py b/server.py
new file mode 100644
index 0000000..91e657b
--- /dev/null
+++ b/server.py
@@ -0,0 +1,238 @@
+#!/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()