#!/usr/bin/env python3
"""Ntfyx public plugin submissions. Python 3, standard library only.

Every submission and publication requires an administrator's in-app decision.
Unpublished content and uncertain operations stay in private, environment-bound
state. No account, credential, or downloaded-code execution is required.
"""
import argparse
import contextlib
import hashlib
import json
import os
from pathlib import Path
import re
import secrets
import sys
import tempfile
import urllib.error
import urllib.parse
import urllib.request


class SafeError(Exception):
    """Only fixed, non-secret errors may reach the terminal."""


class SafeParser(argparse.ArgumentParser):
    def error(self, message):
        # Do not echo potentially private input from invalid command arguments.
        raise SafeError("invalid_arguments_use_help")


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def encode(value):
    return json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8")


def read_json(path):
    try:
        with open(path, "rb") as handle:
            data = handle.read(131073)
        if len(data) > 131072:
            raise SafeError("input_too_large")
        return json.loads(data)
    except (OSError, ValueError, UnicodeError) as exc:
        raise SafeError("cannot_read_json") from exc


def server_origin(value):
    parsed = urllib.parse.urlsplit(value)
    local = parsed.hostname in ("127.0.0.1", "::1", "localhost")
    if (parsed.scheme != "https" and not (local and parsed.scheme == "http")) or (
        not parsed.hostname or parsed.username or parsed.password or parsed.query
        or parsed.fragment or parsed.path not in ("", "/", "/v1", "/v1/")
    ):
        raise SafeError("server_must_be_https_origin")
    return urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, "", "", ""))


def state_path(server, explicit=None):
    if explicit:
        return Path(explicit).expanduser()
    home = Path(os.environ.get("XDG_CONFIG_HOME", str(Path.home() / ".config")))
    tag = hashlib.sha256(server.encode()).hexdigest()[:16]
    return home / "ntfyx" / "plugin-submissions" / (tag + ".json")


def save_state(path, value):
    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    fd, temporary = tempfile.mkstemp(prefix=".pending-", dir=str(path.parent))
    try:
        if os.name == "posix":
            os.fchmod(fd, 0o600)
        with os.fdopen(fd, "wb") as handle:
            handle.write(encode(value) + b"\n")
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
        if os.name == "posix":
            directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
            try:
                os.fsync(directory)
            finally:
                os.close(directory)
    finally:
        with contextlib.suppress(FileNotFoundError):
            os.unlink(temporary)


def load_state(path, server):
    if not path.exists():
        return {"v": 2, "channel": "anonymous-review", "server": server}
    if path.is_symlink() or (os.name == "posix" and path.stat().st_mode & 0o077):
        raise SafeError("state_file_must_be_private_0600")
    value = read_json(path)
    if isinstance(value, dict) and value.get("v") == 1:
        raise SafeError("legacy_state_preserved_use_new_anonymous_state")
    if not isinstance(value, dict) or value.get("v") != 2 or value.get("channel") != "anonymous-review" or value.get("server") != server:
        raise SafeError("state_origin_mismatch")
    return value


def request(server, method, path, body=None):
    headers = {"Accept": "application/json", "User-Agent": "Ntfyx-Plugin-Submissions/2"}
    if body is not None:
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(server + "/v1" + path, data=encode(body) if body is not None else None,
                                 headers=headers, method=method)
    try:
        with urllib.request.build_opener(NoRedirect()).open(req, timeout=30) as response:
            raw = response.read(131073)
        if len(raw) > 131072:
            raise SafeError("response_too_large")
        result = json.loads(raw)
        if not isinstance(result, dict):
            raise SafeError("invalid_server_response")
        return result
    except urllib.error.HTTPError as exc:
        # Do not echo private submitted content from a raw error body.
        code = exc.code
        exc.close()
        raise SafeError("http_" + str(code)) from None
    except (urllib.error.URLError, TimeoutError, OSError):
        raise SafeError("transport_unknown_retry_original_operation") from None
    except (ValueError, UnicodeError):
        raise SafeError("invalid_server_response") from None


def identifier(value):
    if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,128}", value):
        raise SafeError("invalid_plugin_id")
    return value


def public_result(value):
    # Unpublished content and internal approval data are not terminal output.
    fields = ("plugin_id", "submission_id", "kind", "event_id", "revision", "status",
              "notification_status", "created_at", "expires_at", "decided_at", "server_time", "retry_after")
    return {key: value[key] for key in fields if key in value}


@contextlib.contextmanager
def state_lock(path):
    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    lock_path = path.with_suffix(path.suffix + ".lock")
    with open(lock_path, "a+b") as lock:
        if os.name == "posix":
            import fcntl
            os.chmod(lock_path, 0o600)
            try:
                fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                raise SafeError("another_client_operation_is_running") from None
        else:
            import msvcrt
            if lock_path.stat().st_size == 0:
                lock.write(b"0"); lock.flush()
            lock.seek(0)
            try:
                msvcrt.locking(lock.fileno(), msvcrt.LK_NBLCK, 1)
            except OSError:
                raise SafeError("another_client_operation_is_running") from None
        yield


def text(value, maximum):
    if not isinstance(value, str) or not value.strip() or re.search(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", value):
        raise SafeError("invalid_text_field")
    try:
        if len(value.encode("utf-16-le")) // 2 > maximum:
            raise SafeError("invalid_text_field")
    except UnicodeError:
        raise SafeError("invalid_text_field") from None
    return value


def validate_manifest(value):
    required = {"name", "summary", "version", "language", "frequency", "source_urls", "privacy_url", "support_url"}
    optional = {"name_en", "summary_en"}
    if not isinstance(value, dict) or not required <= value.keys() or value.keys() - required - optional:
        raise SafeError("invalid_manifest_fields")
    for key, limit in [("name", 80), ("name_en", 80), ("summary", 500), ("summary_en", 500), ("version", 40), ("language", 20)]:
        if key in value:
            text(value[key], limit)
    if value["frequency"] != "event" or not isinstance(value["source_urls"], list) or not 1 <= len(value["source_urls"]) <= 5:
        raise SafeError("invalid_source_frequency_or_urls")
    for url in [*value["source_urls"], value["privacy_url"], value["support_url"]]:
        text(url, 500)
        parsed = urllib.parse.urlsplit(url)
        if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password:
            raise SafeError("source_urls_must_be_https")
    return value


def validate_sample(value, publication=False):
    fields = {"title", "body", "event_id"} if publication else {"title", "body"}
    if not isinstance(value, dict) or set(value) != fields:
        raise SafeError("invalid_publication_fields" if publication else "invalid_sample_fields")
    text(value["title"], 120); text(value["body"], 8000)
    if len(encode({"title": value["title"], "body": value["body"]})) > 16384:
        raise SafeError("content_exceeds_16KiB_channel_budget")
    if publication and not re.fullmatch(r"[A-Za-z0-9_.:-]{1,80}", value["event_id"] if isinstance(value["event_id"], str) else ""):
        raise SafeError("invalid_event_id")
    if publication:
        # Match the existing 16 KiB message channel, including the service's
        # worst-case 80-UTF16-unit plugin name and daily Session suffix.
        payload = {"type": "message", "title": value["title"], "body": value["body"],
                   "task": {"id": "x" * 43, "name": "中" * 80 + " · 2000-00-00"}}
        if len(encode(payload)) > 16384:
            raise SafeError("publication_exceeds_16KiB_message_budget")
    return value


def complete_pending(state, path):
    pending = state.get("pending")
    if not isinstance(pending, dict):
        raise SafeError("no_pending_operation")
    result = request(state["server"], "POST", pending["path"], pending["body"])
    expected_kind = "plugin_submission" if pending["kind"] == "submit" else "publication"
    if result.get("kind") != expected_kind or result.get("status") not in ("pending_review", "approved", "rejected", "expired"):
        raise SafeError("invalid_submission_receipt")
    submission = identifier(result.get("submission_id")); plugin = identifier(result.get("plugin_id"))
    expected_plugin = pending.get("plugin_id") or pending["body"].get("plugin_id")
    if expected_plugin and plugin != expected_plugin:
        raise SafeError("invalid_plugin_receipt_binding")
    if pending["kind"] == "publish" and result.get("event_id") != pending["body"]["event_id"]:
        raise SafeError("invalid_event_receipt_binding")
    state["submission_id"] = submission
    state["plugin_id"] = plugin
    state["last_intent"] = pending
    state["last_receipt"] = public_result(result)
    del state["pending"]
    save_state(path, state)
    return state["last_receipt"]


def begin(state, path, kind, endpoint, body, plugin=None):
    if state.get("pending"):
        raise SafeError("pending_operation_exists_use_retry")
    pending = {"kind": kind, "path": endpoint, "body": body}
    if plugin:
        pending["plugin_id"] = plugin
    state["pending"] = pending
    save_state(path, state)  # fsync BEFORE the first network request.
    return complete_pending(state, path)


def main(argv=None):
    parser = SafeParser(description="Ntfyx public plugin submissions. Every item needs an in-app review; no account or credential.")
    parser.add_argument("--server", default="https://api.ntfyx.me", help="HTTPS API origin; production and staging have separate state")
    parser.add_argument("--state", help="Private draft/intent state; default: ~/.config/ntfyx/plugin-submissions/<origin-hash>.json")
    commands = parser.add_subparsers(dest="command", required=True)
    submit = commands.add_parser("submit", help="Submit metadata and one sample for phone review")
    submit.add_argument("--slug", required=True)
    submit.add_argument("--manifest", required=True)
    submit.add_argument("--sample", required=True)
    submit.add_argument("--plugin", help="Explicit approved dynamic plugin ID for a metadata update; omit for a new plugin")
    status = commands.add_parser("status", help="Read a plugin/content submission result; never approve or resend it")
    status.add_argument("--submission", nargs="?", const="last", help="Submission ID; default: latest acknowledged submission or publication")
    publish = commands.add_parser("publish", help="Submit one immutable source event for its own phone review")
    publish.add_argument("--plugin", help="Server-issued plugin ID; default: last submitted plugin")
    publish.add_argument("--file", required=True)
    commands.add_parser("retry", help="Retry the exact saved uncertain operation, with its original ID and content")
    args = parser.parse_args(argv)
    server = server_origin(args.server)
    path = state_path(server, args.state)
    with state_lock(path):
        state = load_state(path, server)
        if args.command == "submit":
            if not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,39}", args.slug):
                raise SafeError("slug_must_be_1_to_40_lowercase_alphanumeric_or_hyphen")
            body = {
                "operation_id": secrets.token_hex(16), "slug": args.slug,
                "manifest": validate_manifest(read_json(args.manifest)),
                "sample": validate_sample(read_json(args.sample)),
            }
            if args.plugin:
                body["plugin_id"] = identifier(args.plugin)
            result = begin(state, path, "submit", "/market/submissions", body)
        elif args.command == "status":
            if args.submission in (None, "last") and state.get("pending"):
                raise SafeError("outcome_unknown_use_retry_original_operation")
            submission = identifier(state.get("submission_id") if args.submission in (None, "last") else args.submission)
            received = request(server, "GET", "/market/submissions/" + submission)
            if received.get("submission_id") != submission:
                raise SafeError("invalid_status_receipt_binding")
            result = public_result(received)
        elif args.command == "publish":
            plugin = identifier(args.plugin or state.get("plugin_id"))
            result = begin(state, path, "publish", "/market/plugins/" + plugin + "/publications",
                           validate_sample(read_json(args.file), publication=True), plugin)
        else:
            result = complete_pending(state, path)
        print(json.dumps(result, ensure_ascii=False))
    return result


if __name__ == "__main__":
    try:
        main()
    except SafeError as error:
        print(json.dumps({"error": str(error)}, ensure_ascii=False), file=sys.stderr)
        sys.exit(1)
    except KeyboardInterrupt:
        print('{"error":"interrupted_retry_original_operation"}', file=sys.stderr)
        sys.exit(130)