"""Exact access ACL repair for the reviewed Prod nginx user; no layout change."""
from __future__ import annotations

import ctypes
import json
import os
from pathlib import Path
import pwd
import re
import subprocess
import tempfile


class AccessACL:
    """Use the host libacl (already present in Prod), without installing tools."""
    def __init__(self):
        self.lib = ctypes.CDLL("libacl.so.1", use_errno=True)
        for name, args, result in (
            ("acl_get_file", [ctypes.c_char_p, ctypes.c_int], ctypes.c_void_p),
            ("acl_to_text", [ctypes.c_void_p, ctypes.c_void_p], ctypes.c_void_p),
            ("acl_from_text", [ctypes.c_char_p], ctypes.c_void_p),
            ("acl_set_file", [ctypes.c_char_p, ctypes.c_int, ctypes.c_void_p], ctypes.c_int),
            ("acl_free", [ctypes.c_void_p], ctypes.c_int),
        ):
            function = getattr(self.lib, name)
            function.argtypes, function.restype = args, result

    def read(self, path):
        acl = self.lib.acl_get_file(os.fsencode(path), 0x8000)  # ACL_TYPE_ACCESS
        if not acl:
            raise OSError(ctypes.get_errno(), "cannot read access ACL", str(path))
        text = None
        try:
            text = self.lib.acl_to_text(acl, None)
            if not text:
                raise OSError(ctypes.get_errno(), "cannot format access ACL")
            # libacl renders known qualifiers as names. Keep our reviewed UID
            # numeric so an already-applied ACL compares equal on exact retry.
            return ctypes.string_at(text).decode().replace("user:www-data:", "user:33:")
        finally:
            if text:
                self.lib.acl_free(text)
            self.lib.acl_free(acl)

    def write(self, path, text):
        acl = self.lib.acl_from_text(text.encode())
        if not acl:
            raise ValueError("invalid access ACL")
        try:
            if self.lib.acl_set_file(os.fsencode(path), 0x8000, acl) != 0:
                raise OSError(ctypes.get_errno(), "cannot set access ACL", str(path))
        finally:
            self.lib.acl_free(acl)


READ_PROBE = """
import json, os, pathlib, sys
p = json.load(sys.stdin)
if os.geteuid() != 33:
    raise SystemExit('unexpected nginx user')
def failed(error):
    raise error
for directory, dirs, files in os.walk(p['frontend'], onerror=failed):
    for path in [pathlib.Path(directory), *(pathlib.Path(directory) / n for n in dirs + files)]:
        if path.is_symlink() or os.access(path, os.W_OK):
            raise SystemExit('frontend is symlinked or writable')
        if not path.is_dir():
            with path.open('rb') as stream:
                while stream.read(65536):
                    pass
for path in p['parents']:
    if not os.access(path, os.X_OK) or os.access(path, os.R_OK):
        raise SystemExit('ancestor is not traverse-only')
for path in p['private']:
    if os.access(path, os.R_OK) or os.access(path, os.X_OK):
        raise SystemExit('private Prod path is accessible')
"""


def ensure_frontend_acl(release: Path, frontend: Path) -> Path | None:
    """Back up, repair only five access ACLs, then prove nginx access before cutover."""
    if (not release.is_absolute() or release.resolve() != release
        or release.parent.name != "releases" or release.parent.parent.name != "d3"
        or not re.fullmatch(r"staging-d3-release-[A-Za-z0-9_.-]+", release.name)
        or frontend.parent != release / "frontend-releases"
        or not re.fullmatch(r"d3-[0-9a-f]{64}", frontend.name)
        or frontend.resolve() != frontend or not (frontend / "index.html").is_file()):
        raise ValueError("Prod frontend ACL target is not an exact release artifact")
    # This is the observed/reviewed Prod host contract, not a new configurable user.
    users = subprocess.run(["ps", "-C", "nginx", "-o", "euid="],
                           check=True, capture_output=True, text=True, timeout=10)
    if pwd.getpwnam("www-data").pw_uid != 33 or set(users.stdout.split()) != {"0", "33"}:
        raise ValueError("Prod nginx must have root master and www-data (uid 33) workers")
    parents = [release.parent.parent, release.parent, release]
    protected = [release / ".runtime", release / "run_hubspot_options_adapter.sh"]
    acl = AccessACL()
    records, desired = [], {}
    for path in parents + protected:
        if path.is_symlink() or path.resolve() != path:
            raise ValueError("Prod ACL repair refuses symlinked paths")
        if not (path.is_file() if path == protected[1] else path.is_dir()):
            raise ValueError("Prod ACL repair target has an unexpected file type")
        info = path.stat()
        if info.st_mode & 0o7000:
            raise ValueError("Prod ACL repair refuses special permission bits")
        if os.geteuid() not in {0, info.st_uid}:
            raise ValueError("Prod ACL repair requires the existing path owner")
        before = acl.read(path)
        entries = set(before.splitlines())
        base = {entry for entry in entries if entry.startswith(("user::", "group::", "other::"))}
        # Never widen a pre-existing mask or discard another user's ACL. Unknown
        # ACLs need separate review; allow only the original modes and our repair.
        group = next((entry[7:] for entry in base if entry.startswith("group::")), "")
        extra = {"user:33:--x", "mask::--x"} if path in parents else {"user:33:---", f"mask::{group}"}
        if (entries - base - extra or "user::rwx" not in base
            or group not in {"---", "r-x"} or not base & {"other::---", "other::r-x"}
            or (path in parents and base != {"user::rwx", "group::---", "other::---"})):
            raise ValueError(f"Prod ACL repair refuses unreviewed permissions: {path}")
        desired[path] = "\n".join(sorted(base | extra)) + "\n"
        records.append({"path": str(path), "uid": info.st_uid, "gid": info.st_gid,
                        "mode": info.st_mode & 0o7777, "acl": before})
    changed = [item for item in records if set(item["acl"].splitlines()) != set(desired[Path(item["path"])].splitlines())]
    backup = None
    if changed:
        directory = release / ".runtime/frontend-acl-backups"
        if directory.is_symlink():
            raise ValueError("Prod ACL backup directory is a symlink")
        directory.mkdir(mode=0o700, exist_ok=True)
        if directory.stat().st_mode & 0o777 != 0o700:
            raise ValueError("Prod ACL backup directory must be private")
        fd, name = tempfile.mkstemp(prefix="before-", suffix=".json", dir=directory)
        backup = Path(name)
        with os.fdopen(fd, "w") as stream:
            json.dump({"releaseRoot": str(release), "paths": records}, stream, indent=2)
            stream.flush()
            os.fsync(stream.fileno())
        fd = os.open(directory, os.O_RDONLY)
        try:
            os.fsync(fd)
        finally:
            os.close(fd)
    try:
        # Deny private paths before granting traversal; never set default ACLs.
        for path in protected + list(reversed(parents)):
            if any(item["path"] == str(path) for item in changed):
                acl.write(path, desired[path])
        private = [release / name for name in (".runtime", "env", "env/runtime.env", "logs", "run", "artifacts", "run_hubspot_options_adapter.sh")]
        subprocess.run(
            ["sudo", "-n", "-u", "www-data", "--", "/usr/bin/python3", "-I", "-c", READ_PROBE],
            input=json.dumps({"frontend": str(frontend), "parents": list(map(str, parents)), "private": list(map(str, private))}),
            check=True, capture_output=True, text=True, timeout=30,
        )
    except Exception:
        # Restore only this attempt's changes, including an already contained
        # predecessor's ACLs. The durable backup remains if restoration fails.
        failures = []
        for item in changed:
            try:
                acl.write(Path(item["path"]), item["acl"])
            except OSError as error:
                failures.append(error)
        if failures:
            raise OSError(f"ACL restore incomplete; retained backup: {backup}") from failures[0]
        raise
    return backup