chemenu/tools/chemenu/upload.py

"""The MCP write path's one choke point: `mcp-upload/`, and nothing else
(Gitea #32).

`#19`'s MCP read server has an absence property - nothing under
`chemenu.commands` is importable from it, so a write function does not exist
in that process's reach. Once a `submit` tool exists that property stops
being true by itself: something in the server process now writes. What
replaces it is not an absence but a **positive list**, enforced in code
rather than promised in prose:

    The server process may write into exactly one directory - `mcp-upload/`
    under the served root - and every write in this module resolves its
    target and refuses anything that lands outside it.

`_write_atomic_within()` is that one choke point: every file this module
writes (a submitted file, its manifest) goes through it. The ledger append is
the one exception, and it is a narrower case of the same rule rather than a
gap in it - its destination is a hardcoded constant (`mcp-upload/ledger.jsonl`),
never a caller-supplied name, so there is no path to sanitise in the first
place.

Two stages, two different grants of trust:

    mcp-upload/<id>/   material nobody has looked at - `submit` writes here
       |  wikitool upload accept <id>     <- a human decides (Exit 42 gate)
    incoming/          the ordinary local intake (Gitea #58/#67)
       |  wikitool raw accept ...
    raw/

`upload accept`/`upload reject` (the reviewer commands, `commands/upload_cmd.py`)
are **not** importable from here or from `chemenu.mcp.server` - they live under
`chemenu.commands`, structurally unreachable from the server, same as every
other write command #19 already keeps out.

Stdlib only, like `chemenu.telemetry` - this module is imported by the MCP
server process on every request, not only at CLI dispatch.
"""
from __future__ import annotations

import base64
import binascii
import datetime
import hashlib
import json
import os
import secrets
import shutil
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Optional

from chemenu import config
from chemenu.errors import ValidationError

DEFAULT_IDENTITY_HEADER = "X-Forwarded-User"

# Submissions older than this stop counting toward a submitter's quota. A
# rolling window rather than a calendar day - "resets at midnight" is a
# surprise no operator asked for, and a rolling window needs nothing stored
# beyond the ledger that already exists for other reasons.
_QUOTA_WINDOW = datetime.timedelta(days=1)

_FORBIDDEN_FILENAME_CHARS = ("/", "\\", "\x00")


@dataclass(frozen=True)
class UploadConfig:
    """The opt-in, read from `.wikitool-upload.json` - see
    `config.UPLOAD_CONFIG_FILENAME`."""

    identity_header: str
    max_bytes: int
    allowed_extensions: tuple[str, ...]
    submissions_per_day: int
    bytes_per_day: int


def read_config(root: "Path | str") -> Optional[UploadConfig]:
    """The upload opt-in for `root`, or `None` when it is absent - which means
    the write path does not exist, not that it is unrestricted.

    A malformed file is a `ValidationError`, never a silent "no limits": this
    file decides whether an unauthenticated write path is offered at all, so a
    corrupted safeguard must not read as a disabled one - the same posture
    `git_publish.read_allowed_push_urls` takes for `.wikitool-remotes.json`,
    deliberately not the "ignore what does not parse" posture
    `telemetry.policy` takes for its own config, because that one only ever
    narrows an existing on/off default and this one creates a capability that
    otherwise does not exist.
    """
    path = Path(root) / config.UPLOAD_CONFIG_FILENAME
    if not path.is_file():
        return None
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ValidationError(
            f"{config.UPLOAD_CONFIG_FILENAME} is unreadable ({exc}). It decides whether the "
            "upload tool is offered at all, so a broken file is not treated as 'no limits' - "
            "fix it or delete it deliberately."
        ) from exc
    if not isinstance(data, dict):
        raise ValidationError(f"{config.UPLOAD_CONFIG_FILENAME} must contain a JSON object.")

    expected = (
        '{"schema": 1, "identity_header": "X-Forwarded-User", "max_bytes": 10485760, '
        '"allowed_extensions": [".md", ".pdf"], '
        '"quota": {"submissions_per_day": 20, "bytes_per_day": 52428800}}'
    )
    try:
        identity_header = str(data.get("identity_header") or DEFAULT_IDENTITY_HEADER)
        max_bytes = int(data["max_bytes"])
        raw_extensions = data["allowed_extensions"]
        extensions = tuple(sorted({str(ext).lower() for ext in raw_extensions}))
        quota = data["quota"]
        submissions_per_day = int(quota["submissions_per_day"])
        bytes_per_day = int(quota["bytes_per_day"])
    except (KeyError, TypeError, ValueError) as exc:
        raise ValidationError(
            f"{config.UPLOAD_CONFIG_FILENAME} is missing or misshapes a required field ({exc}). "
            f"Expected: {expected}"
        ) from exc

    if max_bytes <= 0:
        raise ValidationError(f"{config.UPLOAD_CONFIG_FILENAME}: max_bytes must be positive.")
    if submissions_per_day <= 0 or bytes_per_day <= 0:
        raise ValidationError(f"{config.UPLOAD_CONFIG_FILENAME}: quota values must be positive.")
    if not extensions:
        raise ValidationError(f"{config.UPLOAD_CONFIG_FILENAME}: allowed_extensions must not be empty.")
    not_dotted = [ext for ext in extensions if not ext.startswith(".")]
    if not_dotted:
        raise ValidationError(
            f"{config.UPLOAD_CONFIG_FILENAME}: allowed_extensions must each start with '.': {not_dotted}"
        )

    return UploadConfig(
        identity_header=identity_header,
        max_bytes=max_bytes,
        allowed_extensions=extensions,
        submissions_per_day=submissions_per_day,
        bytes_per_day=bytes_per_day,
    )


def sanitize_filename(name: str) -> str:
    """A bare, safe basename, or `ValidationError` - never a path.

    An einreicher chooses this string, so it is adversarial input: no path
    separator, no `..`, no null byte, no leading dot (a dotfile is never a
    legitimate submission name), no empty stem. What survives is still just a
    name - the submission id (generated, never caller-supplied) is what keeps
    two submissions from colliding on disk.
    """
    if name is None or not name.strip():
        raise ValidationError("filename is empty.")
    candidate = name.strip()
    if any(ch in candidate for ch in _FORBIDDEN_FILENAME_CHARS):
        raise ValidationError(
            f"filename must not contain a path separator or a null byte: {name!r}"
        )
    if candidate in (".", ".."):
        raise ValidationError(f"filename must not be '.' or '..': {name!r}")
    if candidate.startswith("."):
        raise ValidationError(f"filename must not start with a dot: {name!r}")
    if Path(candidate).name != candidate:
        raise ValidationError(f"filename must be a bare name, not a path: {name!r}")
    if not Path(candidate).stem:
        raise ValidationError(f"filename has no stem: {name!r}")
    return candidate


def new_submission_id(now: Optional[datetime.datetime] = None) -> str:
    """`<YYYY-MM-DD>T<HHMMSS>Z-<8 hex>` - sortable, never caller-chosen, so a
    colliding name can never overwrite a different submission."""
    moment = now or datetime.datetime.now(datetime.timezone.utc)
    return f"{moment.strftime('%Y-%m-%dT%H%M%SZ')}-{secrets.token_hex(4)}"


def _now_iso() -> str:
    return datetime.datetime.now(datetime.timezone.utc).isoformat()


def _parse_iso(value: Any) -> Optional[datetime.datetime]:
    if not isinstance(value, str):
        return None
    try:
        parsed = datetime.datetime.fromisoformat(value)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=datetime.timezone.utc)
    return parsed


def _upload_root(root: "Path | str") -> Path:
    return (Path(root) / "mcp-upload").resolve()


def _write_atomic_within(root: "Path | str", relative: Path, data: bytes) -> Path:
    """The one choke point every content/manifest write in this module goes
    through: resolve the target, refuse anything outside `mcp-upload/` - a
    resolved path also closes a symlink or a `..` in `relative` - then write
    it via a temp file plus `os.replace` so a reader never observes a partial
    file."""
    base = _upload_root(root)
    target = (base / relative).resolve()
    try:
        target.relative_to(base)
    except ValueError:
        raise ValidationError(f"refusing to write outside mcp-upload/: {relative}")
    target.parent.mkdir(parents=True, exist_ok=True)
    tmp = target.with_name(f".{target.name}.{secrets.token_hex(4)}.tmp")
    tmp.write_bytes(data)
    os.replace(tmp, target)
    return target


def _pending_dir(root: "Path | str", submission_id: str) -> Path:
    """The quarantine directory for `submission_id`, refusing a `submission_id`
    that would resolve outside `mcp-upload/` - this one comes from a CLI
    argument, a human, not the generator above, so it gets the same
    containment check as a write."""
    base = _upload_root(root)
    candidate = (base / submission_id).resolve()
    try:
        candidate.relative_to(base)
    except ValueError:
        raise ValidationError(f"'{submission_id}' is not a valid submission id.")
    return candidate


def _append_ledger(root: "Path | str", event: dict) -> None:
    """Append one event to `mcp-upload/ledger.jsonl` - the destination is this
    literal constant, never a caller-supplied path, which is what makes an
    append-mode write (rather than `_write_atomic_within`'s replace) safe:
    there is nothing here for untrusted input to redirect."""
    base = _upload_root(root)
    base.mkdir(parents=True, exist_ok=True)
    path = base / "ledger.jsonl"
    line = json.dumps(event, sort_keys=True) + "\n"
    with path.open("a", encoding="utf-8") as fh:
        fh.write(line)


def _read_ledger(root: "Path | str") -> list[dict]:
    path = _upload_root(root) / "ledger.jsonl"
    if not path.is_file():
        return []
    events: list[dict] = []
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            parsed = json.loads(line)
        except json.JSONDecodeError:
            continue
        if isinstance(parsed, dict):
            events.append(parsed)
    return events


def _pending_manifests(root: "Path | str") -> list[dict]:
    base = _upload_root(root)
    if not base.is_dir():
        return []
    out = []
    for manifest_path in sorted(base.glob("*/manifest.json")):
        try:
            data = json.loads(manifest_path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            continue
        if isinstance(data, dict):
            out.append(data)
    return out


def digest_to_pending_id(root: "Path | str", digest: str) -> Optional[str]:
    """The id of a submission already waiting with this exact sha256, if any.
    Only *pending* submissions are consulted - an accepted or rejected one no
    longer has a manifest under `mcp-upload/` - so a re-submission after a
    rejection is not blocked by this check."""
    for manifest in _pending_manifests(root):
        if manifest.get("sha256") == digest:
            return manifest.get("id")
    return None


def _check_quota(root: "Path | str", cfg: UploadConfig, submitter: str, size: int) -> None:
    window_start = datetime.datetime.now(datetime.timezone.utc) - _QUOTA_WINDOW
    count = 0
    total_bytes = 0
    for event in _read_ledger(root):
        if event.get("event") != "submitted" or event.get("submitter") != submitter:
            continue
        when = _parse_iso(event.get("time"))
        if when is None or when < window_start:
            continue
        count += 1
        total_bytes += int(event.get("size") or 0)
    if count + 1 > cfg.submissions_per_day:
        raise ValidationError(
            f"Quota exceeded: '{submitter}' has submitted {count} file(s) in the last 24h "
            f"(limit {cfg.submissions_per_day}). Nothing was written."
        )
    if total_bytes + size > cfg.bytes_per_day:
        raise ValidationError(
            f"Quota exceeded: '{submitter}' has submitted {total_bytes} byte(s) in the last 24h "
            f"(limit {cfg.bytes_per_day} bytes). Nothing was written."
        )


def submit(
    root: "Path | str",
    cfg: UploadConfig,
    *,
    filename: str,
    content_b64: str,
    submitter: Optional[str],
) -> dict:
    """Accept one submission into the quarantine, or raise `ValidationError`
    without writing anything.

    `submitter` must already be the value the identity header carried - this
    function does not know about HTTP, headers, or which one is configured
    (`cfg.identity_header` names it only for the refusal message). A `None`
    or empty `submitter` is refused outright: an unattributable submission is
    impossible by construction, not merely discouraged.
    """
    if not submitter or not submitter.strip():
        raise ValidationError(
            f"No identity header ({cfg.identity_header}) on this request - refusing to accept "
            "an unattributable submission. Nothing was written."
        )
    submitter = submitter.strip()

    clean_name = sanitize_filename(filename)
    ext = Path(clean_name).suffix.lower()
    if ext not in cfg.allowed_extensions:
        raise ValidationError(
            f"'{ext or '(none)'}' is not an allowed extension. Allowed: "
            f"{', '.join(cfg.allowed_extensions)}"
        )

    # The base64 length is an upper bound on the decoded size (len*3/4), high
    # by at most the 0-2 padding characters a valid encoding carries: refuse
    # before decoding whenever *even the most optimistic reading* still
    # exceeds the limit, so an oversized submission cannot allocate memory in
    # its own size just to be measured, without rejecting a legitimate
    # payload sitting exactly at the limit on padding alone. The post-decode
    # check below is the exact enforcement; this is only the early exit for
    # what is unambiguously too large.
    approx = (len(content_b64 or "") * 3) // 4
    if approx - 2 > cfg.max_bytes:
        raise ValidationError(
            f"Submission is too large (~{approx} bytes, limit {cfg.max_bytes}). "
            "Nothing was written."
        )
    try:
        data = base64.b64decode(content_b64 or "", validate=True)
    except binascii.Error as exc:
        raise ValidationError(f"content_base64 is not valid base64 ({exc}).") from exc
    if not data:
        raise ValidationError("Submission is empty. Nothing was written.")
    if len(data) > cfg.max_bytes:
        raise ValidationError(
            f"Submission is too large ({len(data)} bytes, limit {cfg.max_bytes}). "
            "Nothing was written."
        )

    digest = hashlib.sha256(data).hexdigest()
    pending = digest_to_pending_id(root, digest)
    if pending is not None:
        raise ValidationError(
            f"This exact content is already waiting for review as '{pending}'. "
            "Nothing was written."
        )

    _check_quota(root, cfg, submitter, len(data))

    submission_id = new_submission_id()
    _write_atomic_within(root, Path(submission_id) / clean_name, data)
    manifest = {
        "schema": 1,
        "id": submission_id,
        "filename": clean_name,
        "size": len(data),
        "sha256": digest,
        "submitter": submitter,
        "submitter_source": cfg.identity_header,
        "submitted_at": _now_iso(),
    }
    _write_atomic_within(
        root,
        Path(submission_id) / "manifest.json",
        (json.dumps(manifest, indent=2, sort_keys=True) + "\n").encode("utf-8"),
    )
    _append_ledger(root, {
        "event": "submitted",
        "id": submission_id,
        "submitter": submitter,
        "time": manifest["submitted_at"],
        "size": len(data),
        "sha256": digest,
    })
    return manifest


def list_submissions(root: "Path | str") -> list[dict]:
    """Every manifest currently waiting in the quarantine, oldest id first
    (the id's own timestamp prefix sorts that way)."""
    return sorted(_pending_manifests(root), key=lambda m: m.get("id", ""))


def read_manifest(root: "Path | str", submission_id: str) -> dict:
    path = _pending_dir(root, submission_id) / "manifest.json"
    if not path.is_file():
        raise ValidationError(f"No submission '{submission_id}' is waiting in mcp-upload/.")
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ValidationError(f"Submission '{submission_id}' has a corrupt manifest ({exc}).") from exc
    if not isinstance(data, dict):
        raise ValidationError(f"Submission '{submission_id}' has a corrupt manifest.")
    return data


def confirm_token(manifest: dict) -> str:
    """sha256 over id/filename/size/sha256/submitter, cut to 12 hex chars -
    same shape as `git_publish.changeset_token`: same submission, same token;
    anything about it moving (a re-submission under the same id is
    impossible, but a stale token from an old manifest is not) changes it."""
    payload = json.dumps(
        {key: manifest.get(key) for key in ("id", "filename", "size", "sha256", "submitter")},
        sort_keys=True,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:12]


def promote(root: "Path | str", submission_id: str) -> Path:
    """Move a submission's file into `incoming/`, delete its quarantine
    directory, and append an `accepted` ledger event. The gate (Exit 42, a
    `--confirm` token) is `commands/upload_cmd.py`'s job, not this function's
    - by the time this runs, clearance has already happened.

    Every check runs before anything moves: an unknown id, a missing file on
    disk, or an already-occupied `incoming/<filename>` all refuse with
    nothing touched. Not atomic across the three effects (move, directory
    cleanup, ledger append) - the same "one filesystem move, then a write"
    shape `raw_cmd.raw_accept_command` already has - but every step it does
    take is ordered so that an interruption leaves file content intact
    either in the quarantine or in `incoming/`, never neither.
    """
    manifest = read_manifest(root, submission_id)
    src_dir = _pending_dir(root, submission_id)
    src = src_dir / manifest["filename"]
    if not src.is_file():
        raise ValidationError(f"Submission '{submission_id}' is missing its file on disk.")

    dest = Path(root) / "incoming" / manifest["filename"]
    if dest.exists():
        raise ValidationError(
            f"incoming/{manifest['filename']} already exists - rename or clear it first. "
            f"Nothing was moved for '{submission_id}'."
        )

    dest.parent.mkdir(parents=True, exist_ok=True)
    src.rename(dest)
    shutil.rmtree(src_dir)
    _append_ledger(root, {
        "event": "accepted",
        "id": submission_id,
        "submitter": manifest.get("submitter"),
        "time": _now_iso(),
        "size": manifest.get("size"),
        "sha256": manifest.get("sha256"),
    })
    return dest


def reject(root: "Path | str", submission_id: str, reason: str) -> None:
    """Delete a submission's material, keeping only its ledger trail - the
    ledger entry is written *before* the delete, so an interruption between
    the two still leaves the record of why it was rejected."""
    if not reason or not reason.strip():
        raise ValidationError("--reason is required and must not be empty.")
    manifest = read_manifest(root, submission_id)
    src_dir = _pending_dir(root, submission_id)
    _append_ledger(root, {
        "event": "rejected",
        "id": submission_id,
        "submitter": manifest.get("submitter"),
        "time": _now_iso(),
        "size": manifest.get("size"),
        "sha256": manifest.get("sha256"),
        "reason": reason.strip(),
    })
    shutil.rmtree(src_dir)