"""Strict service-level exact-path policy for isolated Git writers."""
from __future__ import annotations

import grp
import hashlib
import json
import os
import re
import stat
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path, PurePosixPath, PureWindowsPath
from typing import Any
from urllib.parse import unquote


MAX_POLICY_BYTES = 65_536
POLICY_FIELDS = frozenset(
    {
        "schema_version",
        "policy_id",
        "audience",
        "repository_identity",
        "allowed_branch",
        "allowed_operations",
    }
)
OPERATION_FIELDS = frozenset({"exact_paths", "max_content_bytes", "content_profile"})
SUPPORTED_OPERATIONS = frozenset({"commit_file.atomic"})
SUPPORTED_CONTENT_PROFILES = frozenset({"systemd-codex-review-worker-v1"})
_SECRET_RE = re.compile(
    r"(?i)(authorization\s*:|bearer\s+|(?:token|password|secret)\s*[:=])"
)
_SYSTEMD_NAME_RE = re.compile(r"[A-Za-z][A-Za-z0-9]*\Z")
_SYSTEMD_MAX_LINE_BYTES = 4096
_SYSTEMD_ALLOWED_DIRECTIVES = {
    "Unit": frozenset({"Description", "After", "Wants"}),
    "Service": frozenset(
        {
            "Type",
            "User",
            "Group",
            "WorkingDirectory",
            "Environment",
            "ExecStart",
            "Restart",
            "RestartSec",
            "NoNewPrivileges",
            "PrivateTmp",
            "PrivateDevices",
            "ProtectSystem",
            "ProtectHome",
            "ProtectKernelTunables",
            "ProtectKernelModules",
            "ProtectControlGroups",
            "LockPersonality",
            "RestrictSUIDSGID",
            "RestrictRealtime",
            "MemoryDenyWriteExecute",
            "RuntimeDirectory",
            "StateDirectory",
            "CacheDirectory",
            "ReadOnlyPaths",
            "ReadWritePaths",
            "UMask",
        }
    ),
    "Install": frozenset({"WantedBy"}),
}
_SYSTEMD_MULTI_DIRECTIVES = frozenset(
    {
        ("Service", "Environment"),
        ("Service", "ReadOnlyPaths"),
        ("Service", "ReadWritePaths"),
    }
)
_SYSTEMD_EXACT_SINGLE_VALUES = {
    ("Unit", "Description"): "AIOS Codex Review/Test Worker",
    ("Unit", "After"): "network-online.target",
    ("Unit", "Wants"): "network-online.target",
    ("Service", "Type"): "simple",
    ("Service", "User"): "aiman",
    ("Service", "Group"): "gitops",
    (
        "Service",
        "WorkingDirectory",
    ): "/usr/libexec/aios-codex-review-test-worker/current",
    (
        "Service",
        "ExecStart",
    ): "/usr/bin/python3 -m task_plane.workers.codex_review_test",
    ("Service", "Restart"): "on-failure",
    ("Service", "RestartSec"): "2",
    ("Service", "NoNewPrivileges"): "true",
    ("Service", "PrivateTmp"): "true",
    ("Service", "PrivateDevices"): "true",
    ("Service", "ProtectSystem"): "strict",
    ("Service", "ProtectHome"): "true",
    ("Service", "ProtectKernelTunables"): "true",
    ("Service", "ProtectKernelModules"): "true",
    ("Service", "ProtectControlGroups"): "true",
    ("Service", "LockPersonality"): "true",
    ("Service", "RestrictSUIDSGID"): "true",
    ("Service", "RestrictRealtime"): "true",
    ("Service", "MemoryDenyWriteExecute"): "true",
    ("Service", "RuntimeDirectory"): "aios-codex-review-test-worker",
    ("Service", "StateDirectory"): "aios-codex-review-test-worker",
    ("Service", "CacheDirectory"): "aios-codex-review-test-worker",
    ("Service", "UMask"): "0077",
    ("Install", "WantedBy"): "multi-user.target",
}
_SYSTEMD_EXACT_MULTI_VALUES = {
    ("Service", "Environment"): (
        "AIOS_CODEX_WORKER_MODE=shadow",
        "AIOS_TASK_PLANE_EXECUTION_ENABLED=false",
        "PYTHONDONTWRITEBYTECODE=1",
    ),
    ("Service", "ReadOnlyPaths"): (
        "/data/aios-gitops/repo.git",
        "/usr/libexec/aios-codex-review-test-worker",
    ),
    ("Service", "ReadWritePaths"): (
        "/run/aios-codex-review-test-worker",
        "/var/lib/aios-codex-review-test-worker",
        "/var/cache/aios-codex-review-test-worker",
    ),
}


class PathPolicyError(PermissionError):
    pass


@dataclass(frozen=True)
class OperationPolicy:
    exact_paths: tuple[str, ...]
    max_content_bytes: int
    content_profile: str


@dataclass(frozen=True)
class ServicePathPolicy:
    schema_version: int
    policy_id: str
    audience: str
    repository_identity: str
    allowed_branch: str
    allowed_operations: dict[str, OperationPolicy]
    sha256: str

    def operation(self, name: str) -> OperationPolicy:
        try:
            return self.allowed_operations[name]
        except KeyError:
            raise PathPolicyError("isolated_writer_service_operation_rejected") from None

    def require_exact_path(self, operation: str, path: str) -> OperationPolicy:
        selected = self.operation(operation)
        if path not in selected.exact_paths:
            raise PathPolicyError("isolated_writer_service_path_rejected")
        return selected


def _canonical_json(value: Any) -> bytes:
    return (
        json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
        + "\n"
    ).encode("utf-8")


def validate_relative_path_syntax(path: str) -> str:
    if not isinstance(path, str):
        raise PathPolicyError("isolated_writer_path_rejected")
    if not path or path != path.strip() or "\x00" in path or "\\" in path:
        raise PathPolicyError("isolated_writer_path_rejected")
    try:
        decoded = unquote(path, errors="strict")
    except (UnicodeDecodeError, ValueError):
        raise PathPolicyError("isolated_writer_path_rejected") from None
    if decoded != path:
        raise PathPolicyError("isolated_writer_path_rejected")
    posix = PurePosixPath(path)
    windows = PureWindowsPath(path)
    parts = path.split("/")
    if (
        posix.is_absolute()
        or windows.is_absolute()
        or bool(windows.drive)
        or any(part in {"", ".", ".."} for part in parts)
    ):
        raise PathPolicyError("isolated_writer_path_rejected")
    return path


def _require_fields(value: dict[str, Any], allowed: frozenset[str], code: str) -> None:
    if set(value) != allowed:
        raise PathPolicyError(code)


def _parse_policy(document: Any, canonical: bytes) -> ServicePathPolicy:
    if not isinstance(document, dict):
        raise PathPolicyError("isolated_writer_path_policy_object_required")
    _require_fields(document, POLICY_FIELDS, "isolated_writer_path_policy_fields_rejected")
    if document["schema_version"] != 1:
        raise PathPolicyError("isolated_writer_path_policy_schema_rejected")
    text_fields = (
        "policy_id",
        "audience",
        "repository_identity",
        "allowed_branch",
    )
    if any(
        not isinstance(document[field], str) or not document[field].strip()
        for field in text_fields
    ):
        raise PathPolicyError("isolated_writer_path_policy_value_rejected")
    raw_operations = document["allowed_operations"]
    if not isinstance(raw_operations, dict) or not raw_operations:
        raise PathPolicyError("isolated_writer_path_policy_operations_rejected")
    if not set(raw_operations).issubset(SUPPORTED_OPERATIONS):
        raise PathPolicyError("isolated_writer_path_policy_operations_rejected")
    operations: dict[str, OperationPolicy] = {}
    for name, raw in raw_operations.items():
        if not isinstance(raw, dict):
            raise PathPolicyError("isolated_writer_path_policy_operation_rejected")
        _require_fields(
            raw,
            OPERATION_FIELDS,
            "isolated_writer_path_policy_operation_fields_rejected",
        )
        paths = raw["exact_paths"]
        if (
            not isinstance(paths, list)
            or not paths
            or any(not isinstance(item, str) for item in paths)
        ):
            raise PathPolicyError("isolated_writer_path_policy_paths_rejected")
        validated = tuple(validate_relative_path_syntax(item) for item in paths)
        if len(set(validated)) != len(validated):
            raise PathPolicyError("isolated_writer_path_policy_paths_rejected")
        maximum = raw["max_content_bytes"]
        if not isinstance(maximum, int) or isinstance(maximum, bool) or not 1 <= maximum <= 1_048_576:
            raise PathPolicyError("isolated_writer_path_policy_size_rejected")
        profile = raw["content_profile"]
        if profile not in SUPPORTED_CONTENT_PROFILES:
            raise PathPolicyError("isolated_writer_path_policy_profile_rejected")
        operations[name] = OperationPolicy(validated, maximum, profile)
    return ServicePathPolicy(
        schema_version=1,
        policy_id=document["policy_id"],
        audience=document["audience"],
        repository_identity=document["repository_identity"],
        allowed_branch=document["allowed_branch"],
        allowed_operations=operations,
        sha256=hashlib.sha256(canonical).hexdigest(),
    )


def load_path_policy(
    path: Path,
    *,
    expected_owner_uid: int = 0,
    expected_group_gid: int | None = None,
    expected_parent_mode: int = 0o750,
    expected_file_mode: int = 0o640,
) -> ServicePathPolicy:
    lexical = Path(os.path.abspath(str(path)))
    if lexical.is_symlink():
        raise PathPolicyError("isolated_writer_path_policy_symlink_rejected")
    try:
        parent_info = lexical.parent.stat(follow_symlinks=False)
        file_info = lexical.stat(follow_symlinks=False)
    except FileNotFoundError:
        raise PathPolicyError("isolated_writer_path_policy_missing") from None
    if stat.S_ISLNK(parent_info.st_mode) or not stat.S_ISDIR(parent_info.st_mode):
        raise PathPolicyError("isolated_writer_path_policy_parent_rejected")
    if stat.S_ISLNK(file_info.st_mode) or not stat.S_ISREG(file_info.st_mode):
        raise PathPolicyError("isolated_writer_path_policy_file_rejected")
    if lexical.resolve(strict=True) != lexical:
        raise PathPolicyError("isolated_writer_path_policy_symlink_rejected")
    if (
        parent_info.st_uid != expected_owner_uid
        or stat.S_IMODE(parent_info.st_mode) != expected_parent_mode
    ):
        raise PathPolicyError("isolated_writer_path_policy_parent_permissions_rejected")
    if (
        file_info.st_uid != expected_owner_uid
        or stat.S_IMODE(file_info.st_mode) != expected_file_mode
    ):
        raise PathPolicyError("isolated_writer_path_policy_permissions_rejected")
    if expected_group_gid is not None and (
        parent_info.st_gid != expected_group_gid or file_info.st_gid != expected_group_gid
    ):
        raise PathPolicyError("isolated_writer_path_policy_group_rejected")
    if file_info.st_size <= 0 or file_info.st_size > MAX_POLICY_BYTES:
        raise PathPolicyError("isolated_writer_path_policy_file_size_rejected")
    raw = lexical.read_bytes()
    try:
        document = json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        raise PathPolicyError("isolated_writer_path_policy_json_rejected") from None
    canonical = _canonical_json(document)
    if raw != canonical:
        raise PathPolicyError("isolated_writer_path_policy_not_canonical")
    return _parse_policy(document, canonical)


@lru_cache(maxsize=1)
def configured_path_policy() -> ServicePathPolicy | None:
    configured = os.getenv("AIOS_GITOPS_PATH_POLICY_FILE", "").strip()
    if not configured:
        return None
    if not Path(configured).is_absolute():
        raise PathPolicyError("isolated_writer_path_policy_path_rejected")
    try:
        gitops_gid = grp.getgrnam("gitops").gr_gid
    except KeyError:
        raise PathPolicyError("isolated_writer_path_policy_group_rejected") from None
    loaded = load_path_policy(Path(configured), expected_group_gid=gitops_gid)
    if loaded.audience != os.getenv("AIOS_GITOPS_SERVICE_AUDIENCE", ""):
        raise PathPolicyError("isolated_writer_path_policy_audience_rejected")
    if loaded.repository_identity != os.getenv("AIOS_GITOPS_REPOSITORY_IDENTITY", ""):
        raise PathPolicyError("isolated_writer_path_policy_repository_rejected")
    if loaded.allowed_branch != os.getenv("AIOS_GITOPS_ALLOWED_BRANCH", ""):
        raise PathPolicyError("isolated_writer_path_policy_branch_rejected")
    return loaded


def clear_path_policy_cache() -> None:
    configured_path_policy.cache_clear()


def require_service_path(operation: str, path: str) -> OperationPolicy:
    loaded = configured_path_policy()
    if loaded is None:
        raise PathPolicyError("isolated_writer_service_path_policy_required")
    return loaded.require_exact_path(operation, path)


def configured_path_policy_sha256() -> str | None:
    loaded = configured_path_policy()
    return None if loaded is None else loaded.sha256


def validate_content(operation: str, path: str, content: str) -> None:
    selected = require_service_path(operation, path)
    try:
        encoded = content.encode("utf-8", errors="strict")
    except (AttributeError, UnicodeEncodeError):
        raise PathPolicyError("isolated_writer_systemd_profile_rejected") from None
    if len(encoded) > selected.max_content_bytes:
        raise PathPolicyError("isolated_writer_service_content_too_large")
    if selected.content_profile == "systemd-codex-review-worker-v1":
        _validate_systemd_codex_review_worker(content)
        return
    raise PathPolicyError("isolated_writer_path_policy_profile_rejected")


def _validate_systemd_codex_review_worker(content: str) -> None:
    if not isinstance(content, str):
        raise PathPolicyError("isolated_writer_systemd_profile_rejected")
    try:
        content.encode("utf-8", errors="strict")
    except UnicodeEncodeError:
        raise PathPolicyError("isolated_writer_systemd_profile_rejected") from None
    if (
        "\x00" in content
        or "\r" in content
        or _SECRET_RE.search(content)
        or not content.endswith("\n")
    ):
        raise PathPolicyError("isolated_writer_systemd_profile_rejected")

    sections: set[str] = set()
    values: dict[tuple[str, str], list[str]] = {}
    section: str | None = None
    for line in content.splitlines():
        if len(line.encode("utf-8")) > _SYSTEMD_MAX_LINE_BYTES:
            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
        if not line:
            continue
        if line[0].isspace() or line.startswith(("#", ";")):
            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
        if line.startswith("[") and line.endswith("]"):
            candidate = line[1:-1]
            if (
                not _SYSTEMD_NAME_RE.fullmatch(candidate)
                or candidate not in _SYSTEMD_ALLOWED_DIRECTIVES
                or candidate in sections
            ):
                raise PathPolicyError("isolated_writer_systemd_profile_rejected")
            sections.add(candidate)
            section = candidate
            continue
        if section is None or "=" not in line:
            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
        directive, value = line.split("=", 1)
        if (
            not _SYSTEMD_NAME_RE.fullmatch(directive)
            or directive not in _SYSTEMD_ALLOWED_DIRECTIVES[section]
            or not value
            or value != value.strip()
        ):
            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
        key = (section, directive)
        if key in values and key not in _SYSTEMD_MULTI_DIRECTIVES:
            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
        values.setdefault(key, []).append(value)

    if sections != set(_SYSTEMD_ALLOWED_DIRECTIVES):
        raise PathPolicyError("isolated_writer_systemd_profile_rejected")
    if set(values) != set(_SYSTEMD_EXACT_SINGLE_VALUES) | set(
        _SYSTEMD_EXACT_MULTI_VALUES
    ):
        raise PathPolicyError("isolated_writer_systemd_profile_rejected")
    for key, expected in _SYSTEMD_EXACT_SINGLE_VALUES.items():
        if values.get(key) != [expected]:
            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
    for key, expected in _SYSTEMD_EXACT_MULTI_VALUES.items():
        observed = values.get(key, [])
        if len(observed) != len(set(observed)) or tuple(observed) != expected:
            raise PathPolicyError("isolated_writer_systemd_profile_rejected")