839 lines
32 KiB
Python
839 lines
32 KiB
Python
"""Small, dependency-free local web server for the roster formatter."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from dataclasses import dataclass
|
|
from datetime import date, datetime, timezone
|
|
from email import policy
|
|
from email.parser import BytesParser
|
|
from http import HTTPStatus
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
import stat as stat_module
|
|
from tempfile import NamedTemporaryFile, TemporaryDirectory
|
|
from threading import Event, Lock, Thread
|
|
from typing import Mapping
|
|
from urllib.parse import quote, unquote, urlsplit
|
|
|
|
from roster_formatter import (
|
|
BASE_DIR,
|
|
CURRENT_LAYOUT_FIELDS,
|
|
DEFAULT_LAYOUT,
|
|
LEGACY_LAYOUT_FIELDS,
|
|
LayoutSettings,
|
|
RosterFormatError,
|
|
RosterPeriod,
|
|
format_combined_roster,
|
|
format_roster,
|
|
)
|
|
|
|
|
|
LOGGER = logging.getLogger("roster.web")
|
|
DEFAULT_HOST = "127.0.0.1"
|
|
DEFAULT_PORT = 8787
|
|
DEFAULT_RETENTION_MONTHS = 3
|
|
DEFAULT_MAX_UPLOAD_MB = 20
|
|
RETENTION_CHECK_SECONDS = 6 * 60 * 60
|
|
DEFAULT_ALLOWED_HOSTS = frozenset({"127.0.0.1", "localhost", "::1", "0.0.0.0"})
|
|
LAYOUT_DEFAULTS_FILENAME = ".roster-layout-defaults.json"
|
|
LAYOUT_DEFAULTS_VERSION = 2
|
|
MAX_DEFAULTS_REQUEST_BYTES = 16 * 1024
|
|
|
|
|
|
class RequestError(ValueError):
|
|
def __init__(self, status: HTTPStatus, message: str) -> None:
|
|
super().__init__(message)
|
|
self.status = status
|
|
self.message = message
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UploadedFile:
|
|
filename: str
|
|
content_type: str
|
|
data: bytes
|
|
|
|
|
|
class OutputStore:
|
|
"""Atomic persistent PNG storage with filename-based calendar retention."""
|
|
|
|
def __init__(self, root: Path, retention_months: int = DEFAULT_RETENTION_MONTHS) -> None:
|
|
if retention_months < 0:
|
|
raise ValueError("retention_months cannot be negative")
|
|
self.root = root.resolve()
|
|
self.retention_months = retention_months
|
|
self._lock = Lock()
|
|
self.root.mkdir(parents=True, exist_ok=True)
|
|
|
|
@property
|
|
def layout_defaults_path(self) -> Path:
|
|
return self.root / LAYOUT_DEFAULTS_FILENAME
|
|
|
|
@staticmethod
|
|
def _read_regular_file(path: Path, max_bytes: int | None = None) -> bytes | None:
|
|
try:
|
|
path_stat = path.stat(follow_symlinks=False)
|
|
except OSError:
|
|
return None
|
|
if not stat_module.S_ISREG(path_stat.st_mode):
|
|
return None
|
|
if max_bytes is not None and path_stat.st_size > max_bytes:
|
|
return None
|
|
|
|
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
|
|
try:
|
|
descriptor = os.open(path, flags)
|
|
except OSError:
|
|
return None
|
|
try:
|
|
opened_stat = os.fstat(descriptor)
|
|
if not stat_module.S_ISREG(opened_stat.st_mode):
|
|
return None
|
|
if max_bytes is not None and opened_stat.st_size > max_bytes:
|
|
return None
|
|
source = os.fdopen(descriptor, "rb")
|
|
descriptor = -1
|
|
with source:
|
|
return source.read()
|
|
finally:
|
|
if descriptor >= 0:
|
|
os.close(descriptor)
|
|
|
|
def load_layout_defaults(self) -> LayoutSettings:
|
|
with self._lock:
|
|
raw = self._read_regular_file(
|
|
self.layout_defaults_path,
|
|
max_bytes=MAX_DEFAULTS_REQUEST_BYTES,
|
|
)
|
|
if raw is None:
|
|
return DEFAULT_LAYOUT
|
|
|
|
try:
|
|
payload = json.loads(raw.decode("utf-8"))
|
|
if not isinstance(payload, dict) or set(payload) != {"version", "layout"}:
|
|
raise ValueError("unsupported settings format")
|
|
layout = payload.get("layout")
|
|
if not isinstance(layout, dict):
|
|
raise ValueError("settings fields do not match")
|
|
version = payload.get("version")
|
|
submitted_fields = frozenset(layout)
|
|
if version == 1 and submitted_fields == LEGACY_LAYOUT_FIELDS:
|
|
pass
|
|
elif version == LAYOUT_DEFAULTS_VERSION and submitted_fields == CURRENT_LAYOUT_FIELDS:
|
|
pass
|
|
else:
|
|
raise ValueError("unsupported settings version or fields")
|
|
values = {name: str(value) for name, value in layout.items()}
|
|
return LayoutSettings.from_mapping(values, defaults=DEFAULT_LAYOUT)
|
|
except (UnicodeDecodeError, json.JSONDecodeError, ValueError, TypeError) as exc:
|
|
LOGGER.warning("Ignoring invalid saved layout defaults: %s", exc)
|
|
return DEFAULT_LAYOUT
|
|
|
|
def save_layout_defaults(self, settings: LayoutSettings) -> None:
|
|
data = (
|
|
json.dumps(
|
|
{
|
|
"version": LAYOUT_DEFAULTS_VERSION,
|
|
"layout": settings.as_json_values(),
|
|
},
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
+ "\n"
|
|
).encode("utf-8")
|
|
|
|
with self._lock:
|
|
temporary_path: Path | None = None
|
|
try:
|
|
with NamedTemporaryFile(
|
|
mode="wb",
|
|
prefix=".roster-defaults-",
|
|
suffix=".json",
|
|
dir=self.root,
|
|
delete=False,
|
|
) as temporary:
|
|
temporary.write(data)
|
|
temporary.flush()
|
|
os.fsync(temporary.fileno())
|
|
temporary_path = Path(temporary.name)
|
|
os.replace(temporary_path, self.layout_defaults_path)
|
|
temporary_path = None
|
|
finally:
|
|
if temporary_path is not None:
|
|
temporary_path.unlink(missing_ok=True)
|
|
|
|
@staticmethod
|
|
def _cutoff(today: date | datetime, months: int) -> RosterPeriod:
|
|
current = today.date() if isinstance(today, datetime) else today
|
|
month_index = current.year * 12 + current.month - 1 - months
|
|
return RosterPeriod(year=month_index // 12, month=month_index % 12 + 1)
|
|
|
|
def _prune_locked(self, today: date | datetime) -> list[str]:
|
|
if self.retention_months == 0:
|
|
return []
|
|
|
|
cutoff = self._cutoff(today, self.retention_months)
|
|
cutoff_date = date(cutoff.year, cutoff.month, 1)
|
|
removed: list[str] = []
|
|
for path in self.root.iterdir():
|
|
try:
|
|
path_stat = path.stat(follow_symlinks=False)
|
|
except OSError:
|
|
continue
|
|
if not stat_module.S_ISREG(path_stat.st_mode):
|
|
continue
|
|
period = RosterPeriod.from_output_filename(path.name)
|
|
modified_date = datetime.fromtimestamp(path_stat.st_mtime).date()
|
|
if period is None or (period >= cutoff and modified_date >= cutoff_date):
|
|
continue
|
|
path.unlink()
|
|
removed.append(path.name)
|
|
return sorted(removed)
|
|
|
|
def prune(self, today: date | datetime | None = None) -> list[str]:
|
|
with self._lock:
|
|
return self._prune_locked(today or date.today())
|
|
|
|
def save(
|
|
self,
|
|
filename: str,
|
|
data: bytes,
|
|
today: date | datetime | None = None,
|
|
) -> tuple[Path, list[str], bool]:
|
|
if RosterPeriod.from_output_filename(filename) is None:
|
|
raise ValueError("Refusing to store an unexpected roster filename")
|
|
|
|
with self._lock:
|
|
temporary_path: Path | None = None
|
|
try:
|
|
with NamedTemporaryFile(
|
|
mode="wb",
|
|
prefix=".roster-",
|
|
suffix=".tmp",
|
|
dir=self.root,
|
|
delete=False,
|
|
) as temporary:
|
|
temporary.write(data)
|
|
temporary.flush()
|
|
os.fsync(temporary.fileno())
|
|
temporary_path = Path(temporary.name)
|
|
|
|
final_path = self.root / filename
|
|
os.replace(temporary_path, final_path)
|
|
temporary_path = None
|
|
removed = self._prune_locked(today or date.today())
|
|
return final_path, removed, final_path.exists()
|
|
finally:
|
|
if temporary_path is not None:
|
|
temporary_path.unlink(missing_ok=True)
|
|
|
|
def list_items(self) -> list[dict[str, object]]:
|
|
with self._lock:
|
|
items: list[tuple[RosterPeriod, dict[str, object]]] = []
|
|
for path in self.root.iterdir():
|
|
try:
|
|
path_stat = path.stat(follow_symlinks=False)
|
|
except OSError:
|
|
continue
|
|
if not stat_module.S_ISREG(path_stat.st_mode):
|
|
continue
|
|
period = RosterPeriod.from_output_filename(path.name)
|
|
if period is None:
|
|
continue
|
|
items.append(
|
|
(
|
|
period,
|
|
{
|
|
"filename": path.name,
|
|
"label": period.label,
|
|
"url": f"/downloads/{quote(path.name)}",
|
|
"size": path_stat.st_size,
|
|
"modified": datetime.fromtimestamp(
|
|
path_stat.st_mtime,
|
|
tz=timezone.utc,
|
|
).isoformat(),
|
|
},
|
|
)
|
|
)
|
|
items.sort(key=lambda item: item[0], reverse=True)
|
|
return [item for _, item in items]
|
|
|
|
def verify_writable(self) -> None:
|
|
"""Exercise both file creation and deletion for the health check."""
|
|
|
|
with self._lock:
|
|
with NamedTemporaryFile(
|
|
mode="wb",
|
|
prefix=".roster-health-",
|
|
suffix=".tmp",
|
|
dir=self.root,
|
|
) as probe:
|
|
probe.write(b"ok")
|
|
probe.flush()
|
|
|
|
def read(self, filename: str) -> bytes | None:
|
|
if RosterPeriod.from_output_filename(filename) is None:
|
|
return None
|
|
with self._lock:
|
|
return self._read_regular_file(self.root / filename)
|
|
|
|
|
|
class RosterWebApp:
|
|
def __init__(
|
|
self,
|
|
output_dir: Path,
|
|
retention_months: int = DEFAULT_RETENTION_MONTHS,
|
|
max_upload_bytes: int = DEFAULT_MAX_UPLOAD_MB * 1024 * 1024,
|
|
allowed_hosts: set[str] | None = None,
|
|
) -> None:
|
|
self.web_dir = BASE_DIR / "web"
|
|
self.logo_path = BASE_DIR / "assets" / "logo.png"
|
|
self.store = OutputStore(output_dir, retention_months)
|
|
self.max_upload_bytes = max_upload_bytes
|
|
self.max_request_bytes = 2 * max_upload_bytes + 1024 * 1024
|
|
self.allowed_hosts = DEFAULT_ALLOWED_HOSTS | frozenset(
|
|
host.lower().rstrip(".") for host in (allowed_hosts or set())
|
|
)
|
|
# PyMuPDF does not support concurrent rendering in multiple threads.
|
|
self._format_lock = Lock()
|
|
|
|
def index_html(self) -> bytes:
|
|
template = (self.web_dir / "index.html").read_text(encoding="utf-8")
|
|
saved_layout = self.store.load_layout_defaults()
|
|
replacements = {
|
|
"__DEFAULT_MONTH__": RosterPeriod.suggested().html_value,
|
|
"__RETENTION_MONTHS__": str(self.store.retention_months),
|
|
"__MAX_UPLOAD_MB__": str(self.max_upload_bytes // (1024 * 1024)),
|
|
}
|
|
for name, value in saved_layout.as_form_values().items():
|
|
replacements[f"__{name.upper()}__"] = value
|
|
for name, value in DEFAULT_LAYOUT.as_form_values().items():
|
|
replacements[f"__FACTORY_{name.upper()}__"] = value
|
|
rendered = template
|
|
for placeholder, value in replacements.items():
|
|
rendered = rendered.replace(placeholder, value)
|
|
return rendered.encode("utf-8")
|
|
|
|
def format_upload(
|
|
self,
|
|
uploads: list[UploadedFile],
|
|
values: Mapping[str, str],
|
|
) -> tuple[bytes, str, bool, list[str]]:
|
|
if len(uploads) not in {1, 2}:
|
|
raise RosterFormatError("Choose one or two roster PDFs to upload.")
|
|
for index, upload in enumerate(uploads, start=1):
|
|
prefix = "" if len(uploads) == 1 else f"PDF {index}: "
|
|
if not upload.filename:
|
|
raise RosterFormatError(f"{prefix}Choose a roster PDF to upload.")
|
|
if Path(upload.filename).suffix.lower() != ".pdf":
|
|
raise RosterFormatError(
|
|
f"{prefix}Choose a file whose name ends in .pdf."
|
|
)
|
|
if not upload.data:
|
|
raise RosterFormatError(f"{prefix}The uploaded PDF is empty.")
|
|
if len(upload.data) > self.max_upload_bytes:
|
|
maximum = self.max_upload_bytes // (1024 * 1024)
|
|
raise RequestError(
|
|
HTTPStatus.REQUEST_ENTITY_TOO_LARGE,
|
|
f"{prefix}Each PDF must be {maximum} MB or smaller.",
|
|
)
|
|
|
|
period = RosterPeriod.parse(values.get("month"))
|
|
settings = LayoutSettings.from_mapping(
|
|
values,
|
|
defaults=self.store.load_layout_defaults(),
|
|
)
|
|
|
|
with self._format_lock:
|
|
with TemporaryDirectory(prefix="roster-web-") as temp_dir:
|
|
work_dir = Path(temp_dir)
|
|
input_paths = [
|
|
work_dir / f"upload-{index}.pdf"
|
|
for index in range(1, len(uploads) + 1)
|
|
]
|
|
output_path = work_dir / period.output_filename
|
|
for input_path, upload in zip(input_paths, uploads, strict=True):
|
|
input_path.write_bytes(upload.data)
|
|
if len(input_paths) == 1:
|
|
format_roster(input_paths[0], output_path, period, settings)
|
|
else:
|
|
format_combined_roster(
|
|
input_paths[0],
|
|
input_paths[1],
|
|
output_path,
|
|
period,
|
|
settings,
|
|
)
|
|
png = output_path.read_bytes()
|
|
|
|
_, removed, saved = self.store.save(period.output_filename, png)
|
|
return png, period.output_filename, saved, removed
|
|
|
|
def save_layout_defaults(self, values: Mapping[str, str]) -> LayoutSettings:
|
|
expected = set(DEFAULT_LAYOUT.as_form_values())
|
|
if set(values) != expected:
|
|
raise RosterFormatError("Submit every layout setting before saving defaults.")
|
|
if any(not values[name].strip() for name in expected):
|
|
raise RosterFormatError("Layout defaults cannot contain blank values.")
|
|
settings = LayoutSettings.from_mapping(values)
|
|
self.store.save_layout_defaults(settings)
|
|
return settings
|
|
|
|
|
|
def parse_multipart(content_type: str, body: bytes) -> tuple[dict[str, str], list[UploadedFile]]:
|
|
if "\r" in content_type or "\n" in content_type:
|
|
raise RequestError(HTTPStatus.BAD_REQUEST, "Invalid upload content type.")
|
|
|
|
try:
|
|
header = (
|
|
f"Content-Type: {content_type}\r\nMIME-Version: 1.0\r\n\r\n"
|
|
).encode("ascii", errors="strict")
|
|
except UnicodeEncodeError as exc:
|
|
raise RequestError(HTTPStatus.BAD_REQUEST, "Invalid upload content type.") from exc
|
|
message = BytesParser(policy=policy.default).parsebytes(header + body)
|
|
if not message.is_multipart():
|
|
raise RequestError(
|
|
HTTPStatus.BAD_REQUEST,
|
|
"The upload must use multipart form data.",
|
|
)
|
|
|
|
values: dict[str, str] = {}
|
|
uploads: list[UploadedFile] = []
|
|
for part in message.iter_parts():
|
|
if part.get_content_disposition() != "form-data":
|
|
continue
|
|
name = part.get_param("name", header="content-disposition")
|
|
if not name:
|
|
continue
|
|
|
|
payload = part.get_payload(decode=True)
|
|
if payload is None:
|
|
payload = b""
|
|
filename = part.get_filename()
|
|
if filename is not None:
|
|
if name == "roster":
|
|
uploads.append(
|
|
UploadedFile(
|
|
filename=filename,
|
|
content_type=part.get_content_type(),
|
|
data=payload,
|
|
)
|
|
)
|
|
continue
|
|
|
|
if len(payload) > 4096:
|
|
raise RequestError(HTTPStatus.BAD_REQUEST, "A form value is too long.")
|
|
charset = part.get_content_charset() or "utf-8"
|
|
try:
|
|
values[name] = payload.decode(charset)
|
|
except (LookupError, UnicodeDecodeError) as exc:
|
|
raise RequestError(HTTPStatus.BAD_REQUEST, "Invalid form text encoding.") from exc
|
|
|
|
if len(uploads) not in {1, 2}:
|
|
raise RequestError(
|
|
HTTPStatus.BAD_REQUEST,
|
|
"Upload one or two roster PDFs.",
|
|
)
|
|
return values, uploads
|
|
|
|
|
|
class RosterHTTPServer(ThreadingHTTPServer):
|
|
allow_reuse_address = True
|
|
daemon_threads = True
|
|
|
|
def __init__(self, address: tuple[str, int], app: RosterWebApp) -> None:
|
|
self.web_app = app
|
|
super().__init__(address, RosterRequestHandler)
|
|
|
|
|
|
class RosterRequestHandler(BaseHTTPRequestHandler):
|
|
protocol_version = "HTTP/1.1"
|
|
server_version = "RosterServer/1.0"
|
|
sys_version = ""
|
|
|
|
@property
|
|
def app(self) -> RosterWebApp:
|
|
return self.server.web_app # type: ignore[attr-defined]
|
|
|
|
def setup(self) -> None:
|
|
super().setup()
|
|
self.connection.settimeout(30)
|
|
|
|
def end_headers(self) -> None:
|
|
self.send_header("X-Content-Type-Options", "nosniff")
|
|
self.send_header("X-Frame-Options", "DENY")
|
|
self.send_header("Referrer-Policy", "no-referrer")
|
|
self.send_header(
|
|
"Content-Security-Policy",
|
|
"default-src 'self'; script-src 'self'; style-src 'self'; "
|
|
"img-src 'self' blob:; connect-src 'self'; base-uri 'none'; "
|
|
"form-action 'self'; frame-ancestors 'none'",
|
|
)
|
|
super().end_headers()
|
|
|
|
def _send_bytes(
|
|
self,
|
|
status: HTTPStatus,
|
|
data: bytes,
|
|
content_type: str,
|
|
headers: Mapping[str, str] | None = None,
|
|
) -> None:
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(data)))
|
|
if self.close_connection:
|
|
self.send_header("Connection", "close")
|
|
if headers:
|
|
for name, value in headers.items():
|
|
self.send_header(name, value)
|
|
self.end_headers()
|
|
if self.command != "HEAD":
|
|
self.wfile.write(data)
|
|
|
|
def _send_json(self, status: HTTPStatus, payload: Mapping[str, object]) -> None:
|
|
data = json.dumps(payload, separators=(",", ":")).encode("utf-8")
|
|
self._send_bytes(
|
|
status,
|
|
data,
|
|
"application/json; charset=utf-8",
|
|
{"Cache-Control": "no-store"},
|
|
)
|
|
|
|
def _same_origin(self) -> bool:
|
|
origin = self.headers.get("Origin")
|
|
if not origin:
|
|
return True
|
|
try:
|
|
parsed = urlsplit(origin)
|
|
except ValueError:
|
|
return False
|
|
return (
|
|
parsed.scheme in {"http", "https"}
|
|
and parsed.netloc.lower() == self.headers.get("Host", "").lower()
|
|
)
|
|
|
|
def _host_allowed(self) -> bool:
|
|
raw_host = self.headers.get("Host", "")
|
|
if not raw_host or any(character in raw_host for character in ("@", "/", "\\", " ", "\t")):
|
|
return False
|
|
try:
|
|
parsed = urlsplit(f"//{raw_host}")
|
|
host = parsed.hostname
|
|
parsed.port # Force validation of a supplied port.
|
|
except ValueError:
|
|
return False
|
|
return bool(host and host.lower().rstrip(".") in self.app.allowed_hosts)
|
|
|
|
def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API
|
|
path = urlsplit(self.path).path
|
|
try:
|
|
if not self._host_allowed():
|
|
self._send_json(HTTPStatus.FORBIDDEN, {"error": "Host is not allowed."})
|
|
return
|
|
if path == "/":
|
|
self._send_bytes(
|
|
HTTPStatus.OK,
|
|
self.app.index_html(),
|
|
"text/html; charset=utf-8",
|
|
{"Cache-Control": "no-store"},
|
|
)
|
|
return
|
|
if path == "/static/styles.css":
|
|
self._send_bytes(
|
|
HTTPStatus.OK,
|
|
(self.app.web_dir / "styles.css").read_bytes(),
|
|
"text/css; charset=utf-8",
|
|
{"Cache-Control": "no-store"},
|
|
)
|
|
return
|
|
if path == "/static/app.js":
|
|
self._send_bytes(
|
|
HTTPStatus.OK,
|
|
(self.app.web_dir / "app.js").read_bytes(),
|
|
"text/javascript; charset=utf-8",
|
|
{"Cache-Control": "no-store"},
|
|
)
|
|
return
|
|
if path == "/static/theme.js":
|
|
self._send_bytes(
|
|
HTTPStatus.OK,
|
|
(self.app.web_dir / "theme.js").read_bytes(),
|
|
"text/javascript; charset=utf-8",
|
|
{"Cache-Control": "no-store"},
|
|
)
|
|
return
|
|
if path == "/assets/logo.png":
|
|
self._send_bytes(
|
|
HTTPStatus.OK,
|
|
self.app.logo_path.read_bytes(),
|
|
"image/png",
|
|
{"Cache-Control": "no-store"},
|
|
)
|
|
return
|
|
if path == "/health":
|
|
try:
|
|
self.app.store.verify_writable()
|
|
except OSError:
|
|
self._send_json(
|
|
HTTPStatus.SERVICE_UNAVAILABLE,
|
|
{"status": "error", "error": "Output directory is not writable."},
|
|
)
|
|
return
|
|
self._send_json(HTTPStatus.OK, {"status": "ok"})
|
|
return
|
|
if path == "/api/history":
|
|
self._send_json(
|
|
HTTPStatus.OK,
|
|
{
|
|
"items": self.app.store.list_items(),
|
|
"retention_months": self.app.store.retention_months,
|
|
},
|
|
)
|
|
return
|
|
if path.startswith("/downloads/"):
|
|
filename = unquote(path.removeprefix("/downloads/"))
|
|
data = self.app.store.read(filename)
|
|
if data is None:
|
|
self._send_json(HTTPStatus.NOT_FOUND, {"error": "Roster not found."})
|
|
return
|
|
self._send_bytes(
|
|
HTTPStatus.OK,
|
|
data,
|
|
"image/png",
|
|
{
|
|
"Content-Disposition": f'attachment; filename="{filename}"',
|
|
"Cache-Control": "no-store",
|
|
},
|
|
)
|
|
return
|
|
|
|
self._send_json(HTTPStatus.NOT_FOUND, {"error": "Not found."})
|
|
except (BrokenPipeError, ConnectionResetError):
|
|
return
|
|
except Exception:
|
|
LOGGER.exception("GET %s failed", path)
|
|
self._send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Server error."})
|
|
|
|
def do_POST(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API
|
|
# Never reuse a connection after a POST. Some validation happens before
|
|
# the request body is read, so closing prevents leftover bytes from being
|
|
# interpreted as a second, Origin-less request.
|
|
self.close_connection = True
|
|
path = urlsplit(self.path).path
|
|
try:
|
|
if not self._host_allowed():
|
|
raise RequestError(HTTPStatus.FORBIDDEN, "Host is not allowed.")
|
|
if path not in {"/format", "/api/layout-defaults"}:
|
|
raise RequestError(HTTPStatus.NOT_FOUND, "Not found.")
|
|
if not self._same_origin():
|
|
raise RequestError(HTTPStatus.FORBIDDEN, "Cross-origin requests are not allowed.")
|
|
if self.headers.get("Transfer-Encoding") is not None:
|
|
raise RequestError(
|
|
HTTPStatus.BAD_REQUEST,
|
|
"Transfer-encoded requests are not supported.",
|
|
)
|
|
|
|
raw_length = self.headers.get("Content-Length")
|
|
if raw_length is None:
|
|
raise RequestError(HTTPStatus.LENGTH_REQUIRED, "Request size is required.")
|
|
try:
|
|
content_length = int(raw_length)
|
|
except ValueError as exc:
|
|
raise RequestError(HTTPStatus.BAD_REQUEST, "Invalid upload size.") from exc
|
|
if content_length <= 0:
|
|
raise RequestError(HTTPStatus.BAD_REQUEST, "The request is empty.")
|
|
|
|
content_type = self.headers.get("Content-Type", "")
|
|
if path == "/api/layout-defaults":
|
|
if content_length > MAX_DEFAULTS_REQUEST_BYTES:
|
|
raise RequestError(
|
|
HTTPStatus.REQUEST_ENTITY_TOO_LARGE,
|
|
"The saved defaults request is too large.",
|
|
)
|
|
if content_type.split(";", 1)[0].strip().lower() != "application/json":
|
|
raise RequestError(
|
|
HTTPStatus.BAD_REQUEST,
|
|
"Saved defaults must use JSON.",
|
|
)
|
|
body = self.rfile.read(content_length)
|
|
if len(body) != content_length:
|
|
raise RequestError(HTTPStatus.BAD_REQUEST, "The request was incomplete.")
|
|
try:
|
|
payload = json.loads(body.decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise RequestError(
|
|
HTTPStatus.BAD_REQUEST,
|
|
"Saved defaults must contain valid JSON.",
|
|
) from exc
|
|
if not isinstance(payload, dict):
|
|
raise RequestError(
|
|
HTTPStatus.BAD_REQUEST,
|
|
"Saved defaults must be a JSON object.",
|
|
)
|
|
values = {str(name): str(value) for name, value in payload.items()}
|
|
settings = self.app.save_layout_defaults(values)
|
|
self._send_json(
|
|
HTTPStatus.OK,
|
|
{
|
|
"status": "saved",
|
|
"defaults": settings.as_form_values(),
|
|
},
|
|
)
|
|
return
|
|
|
|
if content_length > self.app.max_request_bytes:
|
|
raise RequestError(
|
|
HTTPStatus.REQUEST_ENTITY_TOO_LARGE,
|
|
"The upload is too large.",
|
|
)
|
|
if not content_type.lower().startswith("multipart/form-data"):
|
|
raise RequestError(
|
|
HTTPStatus.BAD_REQUEST,
|
|
"The upload must use multipart form data.",
|
|
)
|
|
body = self.rfile.read(content_length)
|
|
if len(body) != content_length:
|
|
raise RequestError(HTTPStatus.BAD_REQUEST, "The upload was incomplete.")
|
|
|
|
values, uploads = parse_multipart(content_type, body)
|
|
png, filename, saved, removed = self.app.format_upload(uploads, values)
|
|
if removed:
|
|
LOGGER.info("Retention removed: %s", ", ".join(removed))
|
|
self._send_bytes(
|
|
HTTPStatus.OK,
|
|
png,
|
|
"image/png",
|
|
{
|
|
"Content-Disposition": f'attachment; filename="{filename}"',
|
|
"Cache-Control": "no-store",
|
|
"X-Roster-Saved": "true" if saved else "false",
|
|
},
|
|
)
|
|
except RequestError as exc:
|
|
self._send_json(exc.status, {"error": exc.message})
|
|
except RosterFormatError as exc:
|
|
self._send_json(HTTPStatus.UNPROCESSABLE_ENTITY, {"error": str(exc)})
|
|
except (BrokenPipeError, ConnectionResetError):
|
|
return
|
|
except Exception:
|
|
LOGGER.exception("POST %s failed", path)
|
|
message = (
|
|
"The layout defaults could not be saved. Try again."
|
|
if path == "/api/layout-defaults"
|
|
else "The roster could not be formatted. Try again."
|
|
)
|
|
self._send_json(
|
|
HTTPStatus.INTERNAL_SERVER_ERROR,
|
|
{"error": message},
|
|
)
|
|
|
|
def log_message(self, format: str, *args: object) -> None:
|
|
LOGGER.info("%s - %s", self.address_string(), format % args)
|
|
|
|
|
|
def _environment_int(name: str, fallback: int) -> int:
|
|
raw = os.environ.get(name)
|
|
if raw is None:
|
|
return fallback
|
|
try:
|
|
return int(raw)
|
|
except ValueError as exc:
|
|
raise SystemExit(f"{name} must be a whole number") from exc
|
|
|
|
|
|
def _retention_maintenance(store: OutputStore, stop: Event) -> None:
|
|
while not stop.wait(RETENTION_CHECK_SECONDS):
|
|
try:
|
|
removed = store.prune()
|
|
if removed:
|
|
LOGGER.info("Scheduled retention removed: %s", ", ".join(removed))
|
|
except OSError:
|
|
LOGGER.exception("Scheduled retention cleanup failed")
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="Run the local roster web app")
|
|
parser.add_argument("--host", default=os.environ.get("ROSTER_HOST", DEFAULT_HOST))
|
|
parser.add_argument(
|
|
"--port",
|
|
type=int,
|
|
default=_environment_int("ROSTER_PORT", DEFAULT_PORT),
|
|
)
|
|
parser.add_argument(
|
|
"--output-dir",
|
|
type=Path,
|
|
default=Path(os.environ.get("ROSTER_OUTPUT_DIR", BASE_DIR / "output")),
|
|
)
|
|
parser.add_argument(
|
|
"--retention-months",
|
|
type=int,
|
|
default=_environment_int("ROSTER_RETENTION_MONTHS", DEFAULT_RETENTION_MONTHS),
|
|
help="delete roster PNGs more than this many calendar months old; 0 disables cleanup",
|
|
)
|
|
parser.add_argument(
|
|
"--max-upload-mb",
|
|
type=int,
|
|
default=_environment_int("ROSTER_MAX_UPLOAD_MB", DEFAULT_MAX_UPLOAD_MB),
|
|
)
|
|
return parser
|
|
|
|
|
|
def main() -> int:
|
|
args = build_parser().parse_args()
|
|
if not 1 <= args.port <= 65535:
|
|
raise SystemExit("port must be between 1 and 65535")
|
|
if args.retention_months < 0:
|
|
raise SystemExit("retention months cannot be negative")
|
|
if not 1 <= args.max_upload_mb <= 100:
|
|
raise SystemExit("max upload size must be between 1 and 100 MB")
|
|
|
|
allowed_hosts = {
|
|
host.strip().lower().rstrip(".")
|
|
for host in os.environ.get("ROSTER_ALLOWED_HOSTS", "").split(",")
|
|
if host.strip()
|
|
}
|
|
if args.host not in {"0.0.0.0", "::"}:
|
|
allowed_hosts.add(args.host.lower().rstrip("."))
|
|
|
|
app = RosterWebApp(
|
|
output_dir=args.output_dir,
|
|
retention_months=args.retention_months,
|
|
max_upload_bytes=args.max_upload_mb * 1024 * 1024,
|
|
allowed_hosts=allowed_hosts,
|
|
)
|
|
removed = app.store.prune()
|
|
if removed:
|
|
LOGGER.info("Startup retention removed: %s", ", ".join(removed))
|
|
|
|
server = RosterHTTPServer((args.host, args.port), app)
|
|
maintenance_stop = Event()
|
|
maintenance_thread: Thread | None = None
|
|
if args.retention_months > 0:
|
|
maintenance_thread = Thread(
|
|
target=_retention_maintenance,
|
|
args=(app.store, maintenance_stop),
|
|
name="roster-retention",
|
|
daemon=True,
|
|
)
|
|
maintenance_thread.start()
|
|
LOGGER.info("Roster formatter available at http://%s:%s", args.host, args.port)
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
LOGGER.info("Stopping roster formatter")
|
|
finally:
|
|
maintenance_stop.set()
|
|
if maintenance_thread is not None:
|
|
maintenance_thread.join(timeout=2)
|
|
server.server_close()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
|
raise SystemExit(main())
|