govoplan/tools/checks/security-audit/resolve_workspace_mount.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import posixpath
from pathlib import Path
import re
import sys
from typing import Any


VOLUME_NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]*")
UNSAFE_BIND_SOURCES = {
    Path("/"),
    Path("/bin"),
    Path("/boot"),
    Path("/dev"),
    Path("/etc"),
    Path("/home"),
    Path("/lib"),
    Path("/lib64"),
    Path("/mnt"),
    Path("/opt"),
    Path("/proc"),
    Path("/root"),
    Path("/run"),
    Path("/sbin"),
    Path("/srv"),
    Path("/sys"),
    Path("/tmp"),
    Path("/usr"),
    Path("/var"),
    Path("/var/run"),
}
SENSITIVE_BIND_ROOTS = {
    Path("/dev"),
    Path("/etc"),
    Path("/proc"),
    Path("/root"),
    Path("/run"),
    Path("/sys"),
    Path("/var/run"),
}


class MountResolutionError(ValueError):
    pass


def normalized_absolute_path(
    value: str,
    *,
    field: str,
    resolve_symlinks: bool = True,
) -> Path:
    if not value.startswith("/") or value.startswith("//"):
        raise MountResolutionError(f"{field} must be an absolute path")
    path = Path(posixpath.normpath(value))
    return path.resolve(strict=False) if resolve_symlinks else path


def contains_path(container: Path, path: Path) -> bool:
    return path == container or container in path.parents


def mount_option_value(value: str, *, field: str) -> str:
    if not value or any(character in value for character in (",", "\n", "\r")):
        raise MountResolutionError(f"{field} cannot be represented safely")
    return value


def resolve_workspace_mount(
    mounts: Any,
    *,
    root: str,
    scope: str,
    reports_dir: str,
) -> str:
    if scope not in {"current", "govoplan"}:
        raise MountResolutionError(f"unsupported audit scope: {scope}")
    if not isinstance(mounts, list):
        raise MountResolutionError("Docker mount metadata must be a list")

    root_path = normalized_absolute_path(root, field="audit root")
    scan_root = root_path if scope == "current" else root_path.parent
    reports_path = Path(reports_dir)
    if not reports_path.is_absolute():
        reports_path = root_path / reports_path
    reports_path = reports_path.resolve(strict=False)

    parsed_mounts: list[tuple[Path, dict[str, Any]]] = []
    for mount in mounts:
        if not isinstance(mount, dict) or not isinstance(mount.get("Destination"), str):
            continue
        destination = normalized_absolute_path(
            mount["Destination"],
            field="mount destination",
        )
        parsed_mounts.append((destination, mount))

    candidates = [
        (destination, mount)
        for destination, mount in parsed_mounts
        if destination != Path("/") and contains_path(destination, scan_root)
    ]

    if not candidates:
        raise MountResolutionError(
            f"no job-container mount covers audit scope root {scan_root}"
        )
    longest_depth = max(len(destination.parts) for destination, _ in candidates)
    selected = [
        candidate
        for candidate in candidates
        if len(candidate[0].parts) == longest_depth
    ]
    if len(selected) != 1:
        raise MountResolutionError("job-container workspace mount is ambiguous")

    destination, mount = selected[0]
    nested_mounts = [
        nested_destination
        for nested_destination, _ in parsed_mounts
        if nested_destination != destination
        and contains_path(scan_root, nested_destination)
    ]
    if nested_mounts:
        raise MountResolutionError(
            "nested job-container mounts inside the audit scope cannot be "
            "reproduced safely"
        )
    if mount.get("RW") is not True:
        raise MountResolutionError("job-container workspace mount is not writable")
    if not contains_path(destination, reports_path):
        raise MountResolutionError(
            f"audit reports path {reports_path} is outside the workspace mount"
        )

    destination_value = mount_option_value(
        str(destination),
        field="mount destination",
    )
    mount_type = mount.get("Type")
    if mount_type == "volume":
        name = mount.get("Name")
        if not isinstance(name, str) or VOLUME_NAME.fullmatch(name) is None:
            raise MountResolutionError("workspace volume name is missing or malformed")
        source_value = name
    elif mount_type == "bind":
        source = mount.get("Source")
        if not isinstance(source, str):
            raise MountResolutionError("workspace bind source is missing")
        source_path = normalized_absolute_path(
            source,
            field="mount source",
            resolve_symlinks=False,
        )
        if source_path in UNSAFE_BIND_SOURCES or any(
            contains_path(sensitive_root, source_path)
            for sensitive_root in SENSITIVE_BIND_ROOTS
        ):
            raise MountResolutionError(
                "workspace bind source is too broad or sensitive"
            )
        source_value = mount_option_value(str(source_path), field="mount source")
    else:
        raise MountResolutionError(f"unsupported workspace mount type: {mount_type!r}")

    return f"type={mount_type},source={source_value},target={destination_value}"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Resolve one safe Gitea Actions workspace mount for an audit container."
    )
    parser.add_argument("--root", required=True)
    parser.add_argument("--scope", choices=("current", "govoplan"), required=True)
    parser.add_argument("--reports-dir", required=True)
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    try:
        mounts = json.load(sys.stdin)
        print(
            resolve_workspace_mount(
                mounts,
                root=args.root,
                scope=args.scope,
                reports_dir=args.reports_dir,
            )
        )
    except (json.JSONDecodeError, MountResolutionError) as exc:
        print(f"workspace mount error: {exc}", file=sys.stderr)
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())