"""Validation helpers that stand between an untrusted process (a Coder node,
a Validator node, or a candidate deploy instance) and the things that trust
it: the artifact paths it reports back, and the argv used to launch it as a
subprocess.

Both validators raise WorkspaceSecurityError on anything suspicious rather
than silently coercing or ignoring it.

Adapted verbatim from graphloop's workspace_security.py (2026-07-30) --
no functional changes. graphloop's own Phase-10 self-evolution run hit a
real case this class of check is meant to catch (a worker writing files
outside the isolated tree it was supposed to be confined to); keeping this
module's logic unchanged preserves that validated behavior.
"""

from __future__ import annotations

import os
from typing import Any, List


class WorkspaceSecurityError(Exception):
    """Raised by validate_artifacts/validate_subprocess_command on rejection."""


# Shell metacharacters/sequences that must never appear in a subprocess
# argument, since these commands are meant to run with shell=False and any
# of these would only matter if something later reinterpreted them via a
# shell.
_SHELL_METACHARACTERS = (";", "&&", "|", ">", "`", "$(")


def _normalize_relative_posix_parts(artifact: Any) -> List[str]:
    if not isinstance(artifact, str):
        raise WorkspaceSecurityError(f"artifact path is not a string: {artifact!r}")
    if artifact == "":
        raise WorkspaceSecurityError("artifact path is empty")
    if os.path.isabs(artifact):
        raise WorkspaceSecurityError(f"artifact path is absolute: {artifact!r}")

    posix_path = artifact.replace(os.sep, "/")
    if os.altsep:
        posix_path = posix_path.replace(os.altsep, "/")

    parts = [part for part in posix_path.split("/") if part not in ("", ".")]
    if not parts:
        raise WorkspaceSecurityError(f"artifact path is empty: {artifact!r}")
    if any(part == ".." for part in parts):
        raise WorkspaceSecurityError(
            f"artifact path contains a parent traversal segment: {artifact!r}"
        )
    return parts


def validate_artifacts(artifacts: Any, workspace: str) -> List[str]:
    """Validate and normalize artifact paths reported by an untrusted
    child process, ensuring every one lives safely inside `workspace`."""
    if not isinstance(artifacts, (list, tuple)):
        raise WorkspaceSecurityError(
            f"artifacts must be a list of strings, got {type(artifacts).__name__}"
        )

    workspace_real = os.path.realpath(workspace)
    normalized: List[str] = []

    for artifact in artifacts:
        parts = _normalize_relative_posix_parts(artifact)

        current = workspace
        for part in parts:
            current = os.path.join(current, part)
            if os.path.islink(current):
                raise WorkspaceSecurityError(
                    f"artifact path contains a symlink component: {artifact!r}"
                )

        if not os.path.exists(current):
            raise WorkspaceSecurityError(
                f"artifact does not exist under workspace: {artifact!r}"
            )

        real_target = os.path.realpath(current)
        if os.path.commonpath([workspace_real, real_target]) != workspace_real:
            raise WorkspaceSecurityError(
                f"artifact resolves outside workspace: {artifact!r}"
            )

        normalized.append("/".join(parts))

    return normalized


def validate_subprocess_command(command: Any) -> List[str]:
    """Validate an argv list about to be passed to subprocess.run(shell=False).

    Accepts the canonical [python_executable, existing_script.py, ...] shape
    only, with no shell metacharacters in any argument.
    """
    if not isinstance(command, (list, tuple)):
        raise WorkspaceSecurityError(
            f"command must be a list of strings, got {type(command).__name__}"
        )
    if len(command) == 0:
        raise WorkspaceSecurityError("command must not be empty")

    for argument in command:
        if not isinstance(argument, str):
            raise WorkspaceSecurityError(
                f"command argument is not a string: {argument!r}"
            )
        if "\n" in argument or "\r" in argument:
            raise WorkspaceSecurityError(
                f"command argument contains a newline: {argument!r}"
            )
        for metacharacter in _SHELL_METACHARACTERS:
            if metacharacter in argument:
                raise WorkspaceSecurityError(
                    f"command argument contains shell metacharacter "
                    f"{metacharacter!r}: {argument!r}"
                )

    executable = command[0]
    if not os.path.isfile(executable):
        raise WorkspaceSecurityError(
            f"command executable does not exist: {executable!r}"
        )

    if len(command) < 2:
        raise WorkspaceSecurityError(
            f"command is missing a script argument: {list(command)!r}"
        )

    script = command[1]
    if not script.endswith(".py") or not os.path.isfile(script):
        raise WorkspaceSecurityError(
            f"command script argument is not an existing .py file: {script!r}"
        )

    return list(command)