#!/usr/bin/env python3
"""
SUBSTRATA localhost transport bridge alpha.

Allows the SUBSTRATA HTTPS console to request one outbound LXMF transmission
and retrieve native LXMF inbound messages from a machine running rsLXMF.
The bridge cannot read browser-vault keys, only encrypted signed capsules.

Security: loopback only, explicit exact Origin allowlist, short-lived random
bearer token printed at launch, strict body bounds, no shell invocation,
one-at-a-time CLI dispatch, duplicate suppression, bounded ciphertext-only
inbox spool and explicit browser acknowledgment. Not an anonymity service or
an LXMF delivery-receipt implementation. Inbound capsule signatures are
verified by WebCrypto in the browser before they enter its local database.
"""
from __future__ import annotations

import argparse
import hashlib
import hmac
import json
import os
import re
import secrets
import stat
import subprocess
import tempfile
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from datetime import datetime, timezone
from urllib.parse import urlsplit, parse_qs

DEFAULT_ORIGIN = "https://substrata.fun"
DESTINATION = re.compile(r"^[0-9a-fA-F]{32}$")
CAPSULE_ID = re.compile(r"^[0-9a-f-]{36}$")
MAX_BODY_BYTES = 150_000
MAX_LXM_BYTES = 150_000
MAX_SPOOL_ITEMS = 250
HEX_HASH = re.compile(r"^[0-9a-f]{64}$")


def validate_origin(value: str) -> str:
    """Allow one exact HTTPS origin, or an explicit loopback HTTP origin."""
    try:
        origin = urlsplit(value)
        port = origin.port
        valid = (
            bool(origin.hostname) and bool(origin.netloc) and
            not any(character.isspace() for character in value) and
            origin.username is None and origin.password is None and
            not origin.path and not origin.query and not origin.fragment and
            (port is None or 1 <= port <= 65535) and
            (origin.scheme == "https" or (origin.scheme == "http" and
             origin.hostname in {"localhost", "127.0.0.1", "::1"}))
        )
    except ValueError:
        valid = False
    if not valid:
        raise ValueError("Use an exact HTTPS origin or explicit loopback HTTP origin, without a path.")
    return value


def protect_windows_path(path: Path, directory: bool = False):
    """Replace inherited Windows grants with owner and SYSTEM access only.

    chmod on Windows only changes the read-only flag. Use the native DACL
    instead, and fail closed if the filesystem cannot enforce it.
    """
    import ctypes
    from ctypes import wintypes

    security = ctypes.WinDLL("advapi32", use_last_error=True)
    kernel = ctypes.WinDLL("kernel32", use_last_error=True)
    convert = security.ConvertStringSecurityDescriptorToSecurityDescriptorW
    convert.argtypes = [wintypes.LPCWSTR, wintypes.DWORD,
                        ctypes.POINTER(ctypes.c_void_p), ctypes.c_void_p]
    convert.restype = wintypes.BOOL
    apply = security.SetFileSecurityW
    apply.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, ctypes.c_void_p]
    apply.restype = wintypes.BOOL
    kernel.LocalFree.argtypes = [ctypes.c_void_p]
    kernel.LocalFree.restype = ctypes.c_void_p
    flags = "OICI" if directory else ""
    descriptor = ctypes.c_void_p()
    if not convert(f"D:P(A;{flags};FA;;;OW)(A;{flags};FA;;;SY)", 1,
                   ctypes.byref(descriptor), None):
        raise ctypes.WinError(ctypes.get_last_error())
    try:
        # DACL_SECURITY_INFORMATION | PROTECTED_DACL_SECURITY_INFORMATION
        if not apply(str(path), 0x00000004 | 0x80000000, descriptor):
            raise ctypes.WinError(ctypes.get_last_error())
    finally:
        kernel.LocalFree(descriptor)


def parse_send_request(body: bytes) -> tuple[str, dict]:
    if not body or len(body) > MAX_BODY_BYTES:
        raise ValueError("Request body must be under 150 KB.")
    try:
        data = json.loads(body)
    except (UnicodeDecodeError, json.JSONDecodeError):
        raise ValueError("Expected a JSON send request.") from None
    if not isinstance(data, dict):
        raise ValueError("Expected a JSON object.")
    destination = data.get("destination")
    if not isinstance(destination, str) or not DESTINATION.fullmatch(destination):
        raise ValueError("Reticulum destination must be exactly 32 hexadecimal characters.")
    capsule = data.get("capsule")
    required = {
        "protocol", "v", "id", "kind", "from", "to",
        "ephemeral", "salt", "iv", "createdAt", "ciphertext", "signature",
    }
    if not isinstance(capsule, dict) or capsule.get("protocol") != "substrata.courier" or \
       capsule.get("v") != 1 or capsule.get("kind") != "message" or \
       not isinstance(capsule.get("id"), str) or not CAPSULE_ID.fullmatch(capsule["id"]) or \
       not isinstance(capsule.get("ciphertext"), str) or \
       not isinstance(capsule.get("signature"), str) or set(capsule) != required:
        raise ValueError("Only encrypted SUBSTRATA message capsules can be sent.")
    if len(capsule["ciphertext"]) > 110_000 or len(capsule["signature"]) > 200:
        raise ValueError("Capsule exceeds allowed limits.")
    return destination.lower(), capsule


def _read_msgpack_bytes(buffer: bytes, offset: int) -> tuple[bytes, int]:
    """Read one MessagePack str/bin field; never interpret active content."""
    if offset >= len(buffer):
        raise ValueError("Incomplete LXMF MessagePack field.")
    marker = buffer[offset]
    offset += 1
    if 0xA0 <= marker <= 0xBF:
        length = marker & 31
    elif marker in (0xC4, 0xD9, 0xC5, 0xDA, 0xC6, 0xDB):
        width = {0xC4: 1, 0xD9: 1, 0xC5: 2, 0xDA: 2, 0xC6: 4, 0xDB: 4}[marker]
        if offset + width > len(buffer):
            raise ValueError("Incomplete LXMF MessagePack length.")
        length = int.from_bytes(buffer[offset:offset + width], "big")
        offset += width
    else:
        raise ValueError("Unexpected LXMF content field encoding.")
    if length > MAX_BODY_BYTES or offset + length > len(buffer):
        raise ValueError("LXMF content exceeds bounds or is incomplete.")
    return buffer[offset:offset + length], offset + length


def extract_lxm_capsule(raw: bytes) -> tuple[str, str, dict] | None:
    """Extract only an opaque SUBSTRATA capsule from a received native LXMF wire.

    rsLXMF persists 16-byte destination, 16-byte source, 64-byte signature,
    followed by MessagePack([timestamp, title, content, fields, optional stamp]).
    rsLXMF itself owns the native authentication/admission stage. This parser
    is not a full general-purpose LXMF decoder or native signature verifier.
    Browser code independently verifies the embedded SUBSTRATA signature.
    """
    if not isinstance(raw, bytes) or not 99 <= len(raw) <= MAX_LXM_BYTES:
        raise ValueError("LXMF message length is invalid.")
    marker = raw[96]
    offset = 97
    if 0x90 <= marker <= 0x9F:
        count = marker - 0x90
    elif marker in (0xDC, 0xDD):
        width = 2 if marker == 0xDC else 4
        if offset + width > len(raw):
            raise ValueError("Incomplete LXMF field count.")
        count = int.from_bytes(raw[offset:offset + width], "big")
        offset += width
    else:
        raise ValueError("Invalid LXMF MessagePack array.")
    if not 4 <= count <= 5:
        raise ValueError("Unsupported LXMF field count.")
    if offset >= len(raw) or raw[offset] not in (0xCA, 0xCB):
        raise ValueError("Invalid LXMF timestamp field.")
    offset += 5 if raw[offset] == 0xCA else 9
    if offset > len(raw):
        raise ValueError("Incomplete LXMF timestamp.")
    _, offset = _read_msgpack_bytes(raw, offset)  # Title, ignored.
    content, _ = _read_msgpack_bytes(raw, offset)
    try:
        capsule = json.loads(content.decode("utf-8"))
    except (UnicodeError, json.JSONDecodeError):
        return None  # Ordinary LXMF text, not a SUBSTRATA capsule.
    try:
        _, capsule = parse_send_request(json.dumps({
            "destination": "0" * 32, "capsule": capsule,
        }, separators=(",", ":")).encode("utf-8"))
    except (ValueError, TypeError):
        return None
    return raw[:16].hex(), raw[16:32].hex(), capsule


class InboxSpool:
    """At-least-once bounded ciphertext handoff across native hook processes."""

    def __init__(self, directory: Path):
        self.directory = Path(directory)
        if self.directory.is_symlink():
            raise ValueError("An inbox spool cannot be a symlink.")
        self.directory.mkdir(mode=0o700, parents=True, exist_ok=True)
        if os.name == "nt":
            protect_windows_path(self.directory, directory=True)
        current_mode = stat.S_IMODE(self.directory.stat().st_mode)
        if os.name == "posix" and current_mode & 0o077:
            raise ValueError("Inbox spool is accessible by other users. Set directory mode to 0700.")
        self.lock = threading.Lock()

    def ingest_file(self, message_file: Path) -> str:
        path = Path(message_file)
        if path.suffix != ".lxm" or path.is_symlink() or not path.is_file():
            raise ValueError("The inbound hook must receive a regular .lxm file.")
        if not 99 <= path.stat().st_size <= MAX_LXM_BYTES:
            raise ValueError("Inbound LXMF message is outside the size limit.")
        with path.open("rb") as stream:
            raw = stream.read(MAX_LXM_BYTES + 1)
        message = extract_lxm_capsule(raw)
        if message is None:
            return "ignored-non-substrata"
        destination, source, capsule = message
        identifier = hashlib.sha256(raw).hexdigest()
        record = {
            "id": identifier,
            "lxmfSource": source,
            "lxmfDestination": destination,
            "receivedAt": datetime.now(timezone.utc).isoformat(),
            "capsule": capsule,
        }
        path = self.directory / (identifier + ".json")
        with self.lock:
            if path.exists():
                return "duplicate-ignored"
            if len(list(self.directory.glob("*.json"))) >= MAX_SPOOL_ITEMS:
                raise ValueError("Inbox spool is full. Import or acknowledge messages first.")
            # Write and fsync a private temp file before publishing the queue
            # entry with an atomic hard-link. An interrupted hook cannot leave
            # a partially written .json entry that future dedupe would hide.
            descriptor, temporary = tempfile.mkstemp(prefix=".incoming-", suffix=".tmp", dir=self.directory)
            try:
                with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
                    # Transfer descriptor ownership before permission setup so
                    # every failure closes it before Windows tries to unlink.
                    if os.name == "nt":
                        protect_windows_path(Path(temporary))
                    else:
                        os.fchmod(stream.fileno(), 0o600)
                    json.dump(record, stream, separators=(",", ":"), ensure_ascii=False)
                    stream.flush()
                    os.fsync(stream.fileno())
                os.link(temporary, path)
            except FileExistsError:
                return "duplicate-ignored"
            finally:
                try:
                    os.unlink(temporary)
                except FileNotFoundError:
                    pass
        return "queued-encrypted"

    def pending(self, limit: int = 20, after: str = "") -> tuple[list[dict], int, str | None]:
        if not 1 <= limit <= 25:
            raise ValueError("Inbox limit must be between 1 and 25.")
        if after and not HEX_HASH.fullmatch(after):
            raise ValueError("Inbox cursor must be a valid entry ID.")
        with self.lock:
            paths = sorted(self.directory.glob("*.json"))
            remaining = [path for path in paths if path.stem > after]
            selected = remaining[:limit]
            result = []
            for path in selected:
                if path.is_symlink() or path.stat().st_size > MAX_BODY_BYTES:
                    continue
                try:
                    value = json.loads(path.read_text(encoding="utf-8"))
                except (UnicodeError, json.JSONDecodeError, OSError):
                    continue
                if isinstance(value, dict) and value.get("id") == path.stem and HEX_HASH.fullmatch(path.stem):
                    result.append(value)
            next_cursor = selected[-1].stem if len(remaining) > limit and selected else None
            return result, len(paths), next_cursor

    def acknowledge(self, identifiers: list[str]) -> int:
        if not isinstance(identifiers, list) or not 1 <= len(identifiers) <= 25 or \
           any(not isinstance(item, str) or not HEX_HASH.fullmatch(item) for item in identifiers):
            raise ValueError("Acknowledgment must contain 1–25 valid inbox entry IDs.")
        removed = 0
        with self.lock:
            for identifier in dict.fromkeys(identifiers):
                path = self.directory / (identifier + ".json")
                if path.is_symlink():
                    continue
                try:
                    path.unlink()
                    removed += 1
                except FileNotFoundError:
                    pass  # Idempotent acknowledgment.
        return removed


class OutboundSender:
    def __init__(self, lxmd: Path, lxmf_config: Path, rns_config: Path):
        self.lxmd = lxmd
        self.lxmf_config = lxmf_config
        self.rns_config = rns_config
        self.lock = threading.Lock()
        self.accepted: dict[str, float] = {}

    def send(self, destination: str, capsule: dict) -> str:
        identifier = destination + ":" + capsule["id"]
        with self.lock:
            now = time.monotonic()
            self.accepted = {k: v for k, v in self.accepted.items() if now - v < 600}
            if identifier in self.accepted:
                return "duplicate-suppressed"
            # Reserve during CLI dispatch too: do not double-dispatch on concurrent calls.
            self.accepted[identifier] = now
            filename = None
            try:
                with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", prefix="substrata-capsule-",
                                                 suffix=".json", delete=False) as temporary:
                    filename = temporary.name
                    if os.name == "nt":
                        protect_windows_path(Path(filename))
                    else:
                        os.chmod(filename, 0o600)
                    json.dump(capsule, temporary, separators=(",", ":"), ensure_ascii=False)
                command = [
                    str(self.lxmd), "--config", str(self.lxmf_config),
                    "--rnsconfig", str(self.rns_config), "--send", destination,
                    "--send-file", filename, "--send-method", "opportunistic",
                    "--send-timeout-secs", "30",
                ]
                # Native CLI can wait 15s for announce + up to 2x30s for proof.
                # Allow it to finish and do not tie up the browser request thread.
                result = subprocess.run(command, capture_output=True, text=True, timeout=90, check=False)
                if result.returncode:
                    raise RuntimeError("LXMF CLI returned a nonzero status. Check the daemon and its configured routes.")
                return "accepted-by-local-cli"
            except (OSError, subprocess.TimeoutExpired, RuntimeError):
                self.accepted.pop(identifier, None)
                raise RuntimeError("The local LXMF command failed or timed out. No delivery claim was made.") from None
            finally:
                if filename:
                    try:
                        os.unlink(filename)
                    except OSError:
                        pass


class OutboundJobs:
    """Run potentially slow native CLI sends away from the HTTP response thread.

    A client retry sees the existing job, never starts the same capsule again.
    Jobs are process-local; a restart can make an in-flight result unknown.
    """

    def __init__(self, sender: OutboundSender):
        self.sender = sender
        self.lock = threading.Lock()
        self.jobs: dict[str, dict] = {}

    def submit(self, destination: str, capsule: dict) -> dict:
        identifier = hashlib.sha256((destination + ":" + capsule["id"]).encode()).hexdigest()
        with self.lock:
            current = self.jobs.get(identifier)
            if current:
                return dict(current)
            active = sum(1 for entry in self.jobs.values() if entry["status"] == "running")
            if active >= 2:
                raise RuntimeError("Two local sends are already active. Wait for their status before sending again.")
            # Keep bounded status history; never discard active jobs.
            finished = [key for key, value in self.jobs.items() if value["status"] != "running"]
            for key in finished[:max(0, len(finished) - 120)]:
                self.jobs.pop(key, None)
            job = {
                "jobId": identifier, "status": "running",
                "createdAt": datetime.now(timezone.utc).isoformat(),
                "delivery": "unconfirmed",
            }
            self.jobs[identifier] = job

        def run_job():
            try:
                result = self.sender.send(destination, capsule)
                state = "cli-reported-delivery" if result == "accepted-by-local-cli" else "duplicate-suppressed"
            except RuntimeError:
                state = "failed-or-unknown"
            with self.lock:
                job["status"] = state
                job["completedAt"] = datetime.now(timezone.utc).isoformat()
                job["delivery"] = "cli-reported" if state == "cli-reported-delivery" else "unconfirmed"

        threading.Thread(target=run_job, name="substrata-native-send", daemon=True).start()
        return dict(job)

    def status(self, identifier: str) -> dict | None:
        if not HEX_HASH.fullmatch(identifier):
            return None
        with self.lock:
            job = self.jobs.get(identifier)
            return dict(job) if job else None


def handler_for(sender: OutboundSender, token: str, allowed_origin: str, spool: InboxSpool | None = None):
    jobs = OutboundJobs(sender)
    class Handler(BaseHTTPRequestHandler):
        server_version = "SUBSTRATA-Bridge/0.2"

        def log_message(self, message: str, *args: object) -> None:
            # Paths and response codes only; never log bearer, body or content.
            if args:
                print("Bridge request processed.", flush=True)

        def _cors(self):
            origin = self.headers.get("Origin")
            if origin == allowed_origin:
                self.send_header("Access-Control-Allow-Origin", allowed_origin)
                self.send_header("Vary", "Origin")
                self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
                self.send_header("Access-Control-Allow-Headers", "authorization, content-type")
                self.send_header("Access-Control-Allow-Private-Network", "true")

        def _send(self, status: int, value: dict):
            body = json.dumps(value, separators=(",", ":")).encode()
            self.send_response(status)
            self._cors()
            self.send_header("Content-Type", "application/json")
            self.send_header("Cache-Control", "no-store")
            self.send_header("Content-Length", str(len(body)))
            self.send_header("X-Content-Type-Options", "nosniff")
            self.end_headers()
            self.wfile.write(body)

        def _allowed_origin(self) -> bool:
            origin = self.headers.get("Origin")
            # Non-browser clients without Origin still need the secret token.
            return origin is None or origin == allowed_origin

        def _authorized(self) -> bool:
            bearer = self.headers.get("Authorization", "")
            return self._allowed_origin() and hmac.compare_digest(bearer, "Bearer " + token)

        def do_OPTIONS(self):
            if self.headers.get("Origin") != allowed_origin:
                return self._send(403, {"error": "Origin not allowed."})
            self.send_response(204)
            self._cors()
            self.send_header("Content-Length", "0")
            self.send_header("Access-Control-Max-Age", "600")
            self.end_headers()

        def do_GET(self):
            if not self._authorized():
                return self._send(403, {"error": "Unauthorized local bridge request."})
            uri = urlsplit(self.path)
            if uri.path == "/health":
                return self._send(200, {
                    "protocol": "substrata.bridge", "v": 2,
                    "status": "ready", "transport": "local-rsLXMF-cli",
                    "outbound": True, "inbound": spool is not None,
                    "deliveryReceipts": False,
                })
            if uri.path == "/inbox" and spool:
                query = parse_qs(uri.query)
                try:
                    limit = int(query.get("limit", ["20"])[0])
                    after = query.get("after", [""])[0]
                    pending, total, cursor = spool.pending(limit, after)
                except (ValueError, OSError):
                    return self._send(400, {"error": "Invalid inbox request."})
                return self._send(200, {
                    "protocol": "substrata.bridge.inbox", "v": 1,
                    "entries": pending, "totalPending": total,
                    "nextCursor": cursor,
                    "browserSignatureVerified": False,
                })
            if uri.path == "/send-status":
                identifier = parse_qs(uri.query).get("id", [""])[0]
                status = jobs.status(identifier)
                if status is None:
                    return self._send(404, {"error": "Job not found. If the local bridge restarted, delivery state is unknown."})
                return self._send(200, status)
            return self._send(404, {"error": "Not found."})

        def do_POST(self):
            if not self._authorized():
                return self._send(403, {"error": "Unauthorized local bridge request."})
            path = urlsplit(self.path).path
            if path not in ("/send", "/ack") or (path == "/ack" and not spool):
                return self._send(404, {"error": "Not found."})
            if not self.headers.get("Content-Type", "").lower().startswith("application/json"):
                return self._send(415, {"error": "Only JSON requests are supported."})
            try:
                size = int(self.headers.get("Content-Length", "0"))
            except ValueError:
                return self._send(400, {"error": "Invalid content length."})
            if size < 1 or size > MAX_BODY_BYTES:
                return self._send(413, {"error": "Request exceeds the 150 KB limit."})
            body = self.rfile.read(size)
            if path == "/ack":
                try:
                    record = json.loads(body)
                    if not isinstance(record, dict) or set(record) != {"ids"}:
                        raise ValueError("Unknown acknowledgment fields.")
                    cleared = spool.acknowledge(record["ids"])
                except (ValueError, UnicodeDecodeError, json.JSONDecodeError):
                    return self._send(400, {"error": "Invalid acknowledgment."})
                return self._send(200, {"status": "acknowledged", "removed": cleared})
            try:
                destination, capsule = parse_send_request(body)
            except ValueError as error:
                return self._send(400, {"error": str(error)})
            try:
                job = jobs.submit(destination, capsule)
            except RuntimeError as error:
                return self._send(429, {"error": str(error)})
            self._send(202, {**job, "capsuleId": capsule["id"]})

    return Handler


def main():
    parser = argparse.ArgumentParser(description="SUBSTRATA loopback-only LXMF bridge")
    parser.add_argument("--lxmd", type=Path, help="Path to the built lxmd-rs executable")
    parser.add_argument("--lxmf-config", type=Path, help="rsLXMF config directory")
    parser.add_argument("--rns-config", type=Path, help="rsReticulum config directory")
    parser.add_argument("--spool", type=Path, help="Shared 0700 directory for opaque inbound capsules")
    parser.add_argument("--ingest", action="store_true", help="Hook mode: read one LXMF .lxm from positional path")
    parser.add_argument("message_file", nargs="?", type=Path, help="Native inbound hook's saved .lxm path")
    parser.add_argument("--origin", default=DEFAULT_ORIGIN, type=validate_origin,
                        help="Exact trusted HTTPS origin, or loopback HTTP origin for local development")
    parser.add_argument("--port", type=int, default=8741)
    args = parser.parse_args()
    if args.ingest:
        if not args.spool or not args.message_file:
            parser.error("Inbound hook requires --spool DIR and the .lxm file path.")
        try:
            result = InboxSpool(args.spool.resolve()).ingest_file(args.message_file)
        except (ValueError, OSError) as error:
            parser.error(str(error))
        print("SUBSTRATA INBOUND / " + result, flush=True)
        return
    if not args.lxmd or not args.lxmf_config or not args.rns_config:
        parser.error("Bridge server requires --lxmd, --lxmf-config, and --rns-config.")
    if args.message_file:
        parser.error("A message file is accepted only in --ingest mode.")
    if not args.lxmd.is_file() or not os.access(args.lxmd, os.X_OK):
        parser.error("The lxmd-rs executable does not exist or is not executable.")
    if not (args.lxmf_config / "config").is_file() or not (args.rns_config / "config").is_file():
        parser.error("Both configuration directories must contain a config file.")
    if not 1024 <= args.port <= 65535:
        parser.error("Select an unprivileged port.")
    token = secrets.token_urlsafe(32)
    sender = OutboundSender(args.lxmd.resolve(), args.lxmf_config.resolve(), args.rns_config.resolve())
    spool = InboxSpool((args.spool or (args.lxmf_config / ".substrata-inbox")).resolve())
    server = ThreadingHTTPServer(("127.0.0.1", args.port), handler_for(sender, token, args.origin, spool))
    print("SUBSTRATA LOCAL BRIDGE / OUTBOUND + ENCRYPTED INBOX ALPHA")
    print("Native inbound spool:", spool.directory)
    print("Origin allowed:", args.origin)
    print("Listening only on: http://127.0.0.1:" + str(args.port))
    print("Pairing secret for SUBSTRATA console (do not share elsewhere):")
    print(token, flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nBridge stopped.")
    finally:
        server.server_close()


if __name__ == "__main__":
    main()