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


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