headquarter/apps/api/src/services/permission_fixer.py

"""Permission fixer: applies mount permission policies post-start."""

import logging
import subprocess
from typing import Any

logger = logging.getLogger(__name__)


def apply_mount_permissions(
    container_id: str,
    mounts: list[dict],
    timeout: int = 10,
) -> list[dict[str, Any]]:
    """Apply permission policies to mounted directories in a running container.

    Runs `chown`, `chmod`, and file-mode fixes for each mount that declares
    an owner, mode, or file_mode. Requires the container to have a root user.

    Args:
        container_id: Docker container ID or name.
        mounts: List of mount definitions from the manifest.
        timeout: Max seconds per docker exec command.

    Returns:
        List of result dicts: [{mount_name, success, error}]
    """
    results = []

    for mount in mounts:
        name = mount.get("name", "unknown")
        target = mount["target"]
        owner = mount.get("owner")
        mode = mount.get("mode")
        file_mode = mount.get("file_mode")

        result: dict[str, Any] = {
            "mount_name": name,
            "success": True,
            "error": None,
        }

        # Skip if no permission policy defined
        if not owner and not mode and not file_mode:
            results.append(result)
            continue

        try:
            if owner:
                _run_in_container(
                    container_id,
                    ["chown", "-R", f"{owner}:{owner}", target],
                    timeout,
                )
                logger.debug(
                    "Applied owner %s to %s in container %s",
                    owner,
                    target,
                    container_id,
                )

            if mode and result["success"]:
                _run_in_container(
                    container_id,
                    ["chmod", mode, target],
                    timeout,
                )
                logger.debug(
                    "Applied mode %s to %s in container %s",
                    mode,
                    target,
                    container_id,
                )

            if file_mode and result["success"]:
                _run_in_container(
                    container_id,
                    [
                        "sh",
                        "-c",
                        f"find {target} -type f -exec chmod {file_mode} {{}} +",
                    ],
                    timeout,
                )
                logger.debug(
                    "Applied file_mode %s to files in %s in container %s",
                    file_mode,
                    target,
                    container_id,
                )

        except PermissionFixError as exc:
            result["success"] = False
            result["error"] = str(exc)
            logger.warning(
                "Permission fix failed for mount %s (target=%s): %s",
                name,
                target,
                exc,
            )

        results.append(result)

    return results


def apply_ssh_permissions(
    container_id: str,
    ssh_target: str,
    container_user: str,
    timeout: int = 10,
) -> dict[str, Any]:
    """Fix SSH directory ownership and permissions in a running container.

    Runs chown and chmod on the ~/.ssh directory so the container user
    can use the keys (SSH requires the private key to be owned by the
    user with mode 600).

    Args:
        container_id: Docker container ID or name.
        ssh_target: Absolute path to the .ssh directory inside the container.
        container_user: The container user that should own the keys.
        timeout: Max seconds per docker exec command.

    Returns:
        Result dict with keys: success, error.
    """
    result: dict[str, Any] = {"success": True, "error": None}
    try:
        # Ensure directory is owned by the container user
        _run_in_container(
            container_id,
            ["chown", "-R", f"{container_user}:{container_user}", ssh_target],
            timeout,
        )
        # Set directory permissions
        _run_in_container(
            container_id,
            ["chmod", "700", ssh_target],
            timeout,
        )
        # Set private key permissions
        _run_in_container(
            container_id,
            [
                "sh",
                "-c",
                f"find {ssh_target} -name 'id_*' -type f -exec chmod 600 {{}} +",
            ],
            timeout,
        )
        logger.debug(
            "Applied SSH permissions for user %s on %s in container %s",
            container_user,
            ssh_target,
            container_id,
        )
    except PermissionFixError as exc:
        result["success"] = False
        result["error"] = str(exc)
        logger.warning(
            "SSH permission fix failed for container %s (target=%s): %s",
            container_id,
            ssh_target,
            exc,
        )
    return result


class PermissionFixError(Exception):
    """Raised when a permission fix command fails."""

    pass


def _run_in_container(
    container_id: str,
    command: list[str],
    timeout: int,
) -> None:
    """Run a command inside a container as root.

    Args:
        container_id: Docker container ID or name.
        command: Command + args to execute.
        timeout: Max seconds to wait.

    Raises:
        PermissionFixError: If the command fails or times out.
    """
    cmd = ["docker", "exec", "--user", "root", container_id] + command

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired:
        raise PermissionFixError(
            f"Command timed out after {timeout}s: {' '.join(command)}"
        )
    except FileNotFoundError:
        raise PermissionFixError(f"Docker command not found: {' '.join(command)}")

    if result.returncode != 0:
        raise PermissionFixError(
            f"Command failed (rc={result.returncode}): {result.stderr.strip()}"
        )


def check_root_user_available(container_id: str, timeout: int = 5) -> bool:
    """Check if the container has a root user we can exec as.

    Args:
        container_id: Docker container ID or name.
        timeout: Max seconds to wait.

    Returns:
        True if root user exists and is usable.
    """
    try:
        _run_in_container(container_id, ["id", "root"], timeout)
        return True
    except PermissionFixError:
        return False