"""Small privileged filesystem seam for durable runtime configuration publication."""

from __future__ import annotations

import fcntl
import hashlib
import json
import os
import stat
import subprocess
import sys
import tempfile
from pathlib import Path


def fail(msg: str) -> None:
    raise RuntimeError(msg)


def safe_id(v: str) -> bool:
    return bool(__import__("re").fullmatch(r"[a-z][a-z0-9-]{2,62}", v))


def safe_rev(v: str) -> bool:
    return bool(__import__("re").fullmatch(r"[0-9a-f]{40}", v))


def open_dir(parent: int | None, name: str, create: bool = False) -> int:
    # Darwin rejects O_NOFOLLOW|openat for directories (ELOOP); lstat the
    # component before opening and verify the resulting descriptor below. Linux
    # uses the stronger flag where available.
    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
    if sys.platform != "darwin":
        flags |= os.O_NOFOLLOW
    try:
        entry = os.stat(name, dir_fd=parent, follow_symlinks=False)
        if stat.S_ISLNK(entry.st_mode):
            fail("runtime config directory is not trusted")
        return os.open(name, flags, dir_fd=parent)
    except FileNotFoundError:
        if not create:
            raise
        os.mkdir(name, 0o700, dir_fd=parent)
        return os.open(name, flags, dir_fd=parent)


def checked_dir(fd: int, expected_mode: int = 0o700) -> None:
    s = os.fstat(fd)
    if (
        not stat.S_ISDIR(s.st_mode)
        or s.st_nlink < 1
        or stat.S_IMODE(s.st_mode) != expected_mode
        or s.st_uid != os.getuid()
    ):
        fail("runtime config directory is not trusted")


def walk(root: str, comps: list[str], create: bool = True) -> int:
    """Open an absolute path component-by-component without following symlinks.

    In particular, never use os.makedirs/root pathname resolution here: an attacker
    replacing an ancestor between those calls must not redirect publication.
    """
    if not os.path.isabs(root):
        fail("data root must be absolute")
    # macOS exposes temporary directories through the conventional /var and
    # /tmp symlinks. Resolve only these OS-owned aliases; workspace-owned
    # ancestors remain component checked and are never realpath-followed.
    if root == "/var" or root == "/tmp" or root.startswith(("/var/", "/tmp/")):
        root = "/private" + root
    parts = [part for part in Path(root).parts if part not in ("", "/")]
    if any(part in (".", "..") or "/" in part for part in parts + comps):
        fail("unsafe path component")
    fd = os.open("/", os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try:
        all_components = [*parts, *comps]
        for index, component in enumerate(all_components):
            nxt = open_dir(fd, component, create)
            # Ancestors such as /var/folders are installation-owned and commonly
            # 0755; the trusted runtime root and every workspace child are private.
            info = os.fstat(nxt)
            if (not stat.S_ISDIR(info.st_mode) or info.st_nlink < 1
                    or (index >= len(parts) - 1 and (info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700))):
                os.close(nxt)
                fail("runtime config directory is not trusted")
            os.close(fd)
            fd = nxt
        return fd
    except BaseException:
        os.close(fd)
        raise


def read_regular(fd: int, mode: int, expected: bytes | None = None) -> os.stat_result:
    s = os.fstat(fd)
    if (
        not stat.S_ISREG(s.st_mode)
        or s.st_nlink != 1
        or stat.S_IMODE(s.st_mode) != mode
        or s.st_uid != os.getuid()
    ):
        fail("runtime config file is not trusted")
    if expected is not None:
        os.lseek(fd, 0, os.SEEK_SET)
        chunks = []
        while True:
            x = os.read(fd, 1024 * 1024)
            if not x:
                break
            chunks.append(x)
        if b"".join(chunks) != expected:
            fail("same-revision runtime configuration changed")
    return s


def write_all(fd: int, data: bytes) -> None:
    pos = 0
    while pos < len(data):
        n = os.write(fd, data[pos:])
        if n <= 0:
            fail("short runtime config write")
        pos += n


def read_all(fd: int, limit: int = 16 * 1024 * 1024) -> bytes:
    os.lseek(fd, 0, os.SEEK_SET)
    chunks: list[bytes] = []
    total = 0
    while True:
        chunk = os.read(fd, min(1024 * 1024, limit - total))
        if not chunk:
            return b"".join(chunks)
        chunks.append(chunk)
        total += len(chunk)
        if total > limit:
            fail("runtime config file is too large")


def strict_manifest(value: object) -> dict:
    if not isinstance(value, dict):
        fail("runtime config manifest is invalid")
    required = {
        "version", "workspace_id", "workspace_revision", "descriptor_git_blob",
        "descriptor_sha256", "config_sha256", "config_dwh_binding", "config_dev",
        "config_ino", "config_size", "config_mode", "config_uid", "config_nlink",
    }
    if set(value) != required or value.get("version") != 1:
        fail("runtime config manifest is invalid")
    if not safe_id(value.get("workspace_id")) or not safe_rev(value.get("workspace_revision")):
        fail("runtime config manifest is invalid")
    if not isinstance(value.get("descriptor_git_blob"), str) or not safe_rev(value["descriptor_git_blob"]):
        fail("runtime config manifest is invalid")
    for key in ("descriptor_sha256", "config_sha256"):
        if not isinstance(value[key], str) or not __import__("re").fullmatch(r"[0-9a-f]{64}", value[key]):
            fail("runtime config manifest is invalid")
    binding_value = value.get("config_dwh_binding")
    if not isinstance(binding_value, dict) or set(binding_value) != {"workspace_id", "config_fingerprint", "input_fingerprint"} or any(not isinstance(x, str) for x in binding_value.values()):
        fail("runtime config manifest is invalid")
    for key in ("config_dev", "config_ino", "config_size", "config_uid", "config_nlink"):
        if not isinstance(value[key], str) or not value[key].isdigit():
            fail("runtime config manifest is invalid")
    if value["config_mode"] != "400":
        fail("runtime config manifest is invalid")
    return value


def publish(inp: dict) -> dict:
    root = inp.get("data_root")
    wid = inp.get("workspace_id")
    rev = inp.get("workspace_revision")
    if (
        not isinstance(root, str)
        or not os.path.isabs(root)
        or not safe_id(wid)
        or not safe_rev(rev)
    ):
        fail("invalid publication identity")
    try:
        content = bytes.fromhex(inp["config_hex"])
    except (TypeError, ValueError):
        fail("invalid config bytes")
    base = inp.get("manifest_base")
    if not isinstance(base, dict):
        fail("invalid manifest")
    if base.get("workspace_id") != wid or base.get("workspace_revision") != rev:
        fail("manifest identity mismatch")
    sessions = walk(root, ["sessions"], True)
    ws = open_dir(sessions, wid, True)
    checked_dir(ws)
    prep = open_dir(ws, "preprocessing", True)
    checked_dir(prep)
    cfgdir = open_dir(prep, "runtime-config", True)
    checked_dir(cfgdir)
    mandir = open_dir(prep, "runtime-config-manifests", True)
    checked_dir(mandir)
    lockfd = os.open(
        "runtime-config.lock", os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600, dir_fd=prep
    )
    try:
        ls = os.fstat(lockfd)
        if (
            not stat.S_ISREG(ls.st_mode)
            or ls.st_nlink != 1
            or stat.S_IMODE(ls.st_mode) != 0o600
            or ls.st_uid != os.getuid()
        ):
            fail("runtime config lock is not trusted")
        fcntl.flock(lockfd, fcntl.LOCK_EX)
        name = f"{rev}.yaml"
        mname = f"{rev}.json"

        def current(dfd, n, mode):
            try:
                fd = os.open(n, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=dfd)
            except FileNotFoundError:
                return None
            try:
                return (fd, read_regular(fd, mode))
            except:
                os.close(fd)
                raise

        got = current(cfgdir, name, 0o400)
        if got:
            fd, s = got
            os.lseek(fd, 0, os.SEEK_SET)
            old = read_all(fd)
            os.close(fd)
            if old != content:
                fail("same-revision runtime configuration changed")
        else:
            stage = f".{name}.staging-{os.getpid()}-{os.urandom(8).hex()}"
            fd = os.open(
                stage, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=cfgdir
            )
            try:
                write_all(fd, content)
                os.fchmod(fd, 0o400)
                os.fsync(fd)
                try:
                    os.link(
                        stage, name, src_dir_fd=cfgdir, dst_dir_fd=cfgdir, follow_symlinks=False
                    )
                except FileExistsError:
                    pass
                # Keep metadata ordering explicit even on filesystems where a
                # hardlink publication does not retain fchmod as expected.
                os.fchmod(fd, 0o400)
                os.fsync(fd)
            finally:
                os.close(fd)
                try:
                    os.unlink(stage, dir_fd=cfgdir)
                except FileNotFoundError:
                    pass
            got = current(cfgdir, name, 0o400)
            if not got:
                fail("runtime config publication failed")
            fd, s = got
            try:
                os.lseek(fd, 0, os.SEEK_SET)
                if read_all(fd) != content:
                    fail("same-revision runtime configuration changed")
            finally:
                os.close(fd)
            os.fsync(cfgdir)
        # Identity is deliberately recorded after final no-replace publication.
        got = current(cfgdir, name, 0o400)
        assert got
        fd, s = got
        os.close(fd)
        manifest = {"version": 1, **dict(base)}
        manifest.update(
            {
                "config_sha256": hashlib.sha256(content).hexdigest(),
                "config_dev": str(s.st_dev),
                "config_ino": str(s.st_ino),
                "config_size": str(s.st_size),
                "config_mode": format(stat.S_IMODE(s.st_mode), "o"),
                "config_uid": str(s.st_uid),
                "config_nlink": str(s.st_nlink),
            }
        )
        strict_manifest(manifest)
        mb = (json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n").encode()
        oldm = current(mandir, mname, 0o600)
        if oldm:
            mfd, _ = oldm
            os.lseek(mfd, 0, os.SEEK_SET)
            existing = read_all(mfd)
            os.close(mfd)
            try: strict_manifest(json.loads(existing.decode()))
            except (ValueError, TypeError, UnicodeError, RuntimeError): fail("runtime config manifest is invalid")
            if existing != mb:
                fail("same-revision runtime configuration changed")
        else:
            stage = f".{mname}.staging-{os.getpid()}-{os.urandom(8).hex()}"
            fd = os.open(
                stage, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=mandir
            )
            try:
                write_all(fd, mb)
                os.fchmod(fd, 0o600)
                os.fsync(fd)
                os.link(stage, mname, src_dir_fd=mandir, dst_dir_fd=mandir, follow_symlinks=False)
            except FileExistsError:
                pass
            finally:
                os.close(fd)
                try:
                    os.unlink(stage, dir_fd=mandir)
                except FileNotFoundError:
                    pass
            os.fsync(mandir)
        return {
            "path": f"{root}/sessions/{wid}/preprocessing/runtime-config/{name}",
            "manifestPath": f"{root}/sessions/{wid}/preprocessing/runtime-config-manifests/{mname}",
            "manifest": mb.decode(),
            "manifest_sha256": hashlib.sha256(mb).hexdigest(),
            "dev": s.st_dev,
            "ino": s.st_ino,
        }
    finally:
        os.close(lockfd)
        os.close(cfgdir)
        os.close(mandir)
        os.close(prep)
        os.close(ws)
        os.close(sessions)


def verified_snapshot(inp: dict) -> dict:
    root = inp.get("snapshots_root")
    rev = inp.get("workspace_revision")
    wid = inp.get("workspace_id")
    if (
        not isinstance(root, str)
        or not os.path.isabs(root)
        or not safe_rev(rev)
        or not safe_id(wid)
    ):
        fail("invalid snapshot identity")
    # Component-relative no-follow traversal all the way to the retained descriptor.
    sroot = walk(root, [], False)
    rdir = open_dir(sroot, rev, False)
    checked_dir(rdir)
    fd = os.open(f"{wid}.yaml", os.O_RDONLY | os.O_NOFOLLOW, dir_fd=rdir)
    try:
        read_regular(fd, 0o400)
        chunks = []
        while True:
            x = os.read(fd, 1024 * 1024)
            if not x:
                break
            chunks.append(x)
        source = b"".join(chunks)
    finally:
        os.close(fd)
    mf = os.open("snapshot.json", os.O_RDONLY | os.O_NOFOLLOW, dir_fd=rdir)
    try:
        read_regular(mf, 0o400)
        payload = b""
        while True:
            x = os.read(mf, 1024 * 1024)
            if not x:
                break
            payload += x
    finally:
        os.close(mf)
    try:
        manifest = json.loads(payload.decode())
    except (UnicodeDecodeError, json.JSONDecodeError):
        fail("workspace snapshot integrity check failed")
    if not isinstance(manifest, dict) or set(manifest) != {"head", "revisions", "files"}:
        fail("workspace snapshot integrity check failed")
    records = manifest.get("revisions")
    files = manifest.get("files")
    record = next((r for r in records if isinstance(r, dict) and r.get("id") == wid), None) if isinstance(records, list) else None
    expected_path = f"{root}/{rev}/{wid}.yaml"
    if (
        manifest.get("head") != rev or not isinstance(records, list) or not record
        or set(record) != {"id", "commit", "blob", "snapshotPath"}
        or record.get("commit") != rev or record.get("snapshotPath") != expected_path
        or not isinstance(record.get("blob"), str) or not safe_rev(record.get("blob"))
        or not isinstance(files, dict)
        or files.get(f"{wid}.yaml") != hashlib.sha256(source).hexdigest()
    ):
        fail("workspace snapshot integrity check failed")
    repo = inp.get("repository_root")
    if not isinstance(repo, str) or not os.path.isabs(repo):
        fail("invalid repository root")
    try:
        blob = subprocess.check_output(
            ["git", "-C", repo, "rev-parse", f"{rev}:workspaces/{wid}.yaml"],
            stderr=subprocess.DEVNULL, text=True, timeout=5,
        ).strip()
        git_source = subprocess.check_output(
            ["git", "-C", repo, "show", f"{rev}:workspaces/{wid}.yaml"],
            stderr=subprocess.DEVNULL, timeout=5,
        )
    except (OSError, subprocess.SubprocessError):
        fail("workspace Git revision is unavailable")
    try:
        import re
        from collections import Counter
        normalize = lambda value: re.findall(r"[A-Za-z0-9_.:/@+-]+", value)
        git_tokens = Counter(normalize(git_source.decode("utf-8")))
        snapshot_tokens = Counter(normalize(source.decode("utf-8")))
        # The registry canonicalizer may add schema defaults/reorder mappings.
        # Every token from the exact Git descriptor must nevertheless survive;
        # replacements (including non-rendered workspace.name) are rejected.
        equivalent = all(snapshot_tokens[k] >= count for k, count in git_tokens.items())
    except UnicodeDecodeError:
        equivalent = False
    if blob != record.get("blob") or not equivalent:
        fail("workspace Git descriptor identity mismatch")
    return {
        "workspace_id": wid,
        "workspace_revision": rev,
        "source": source.decode(),
        "sha256": hashlib.sha256(source).hexdigest(),
        "descriptor_git_blob": record.get("blob"),
        "snapshot_path": f"{root}/{rev}/{wid}.yaml",
    }


def binding(inp: dict) -> dict:
    try:
        raw = bytes.fromhex(inp["config_hex"])
    except (TypeError, ValueError):
        fail("invalid config bytes")
    # Use the harness' own Pydantic loader and config_dwh_binding; this is intentionally
    # not a TypeScript reimplementation of its normalization/fingerprinting rules.
    from tht.config import load_config
    from tht.jobs.dwh_pipeline import config_dwh_binding

    with tempfile.NamedTemporaryFile(
        prefix="runtime-binding-", suffix=".yaml", delete=False
    ) as stream:
        stream.write(raw)
        path = Path(stream.name)
    try:
        return config_dwh_binding(load_config(path))
    finally:
        try:
            path.unlink()
        except OSError:
            pass


def main() -> None:
    try:
        inp = json.load(sys.stdin)
        action = inp.get("action")
        if action == "publish":
            result = publish(inp)
        elif action == "verified-snapshot":
            result = verified_snapshot(inp)
        elif action == "binding":
            result = binding(inp)
        else:
            fail("unsupported runtime config action")
        print(json.dumps(result))
    except Exception as e:  # noqa: BLE001
        print(json.dumps({"error": str(e)}))
        raise SystemExit(1)


if __name__ == "__main__":
    main()