path: root/server.py

#!/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()