gitea-agent/harness/lib/verification_root.py

#!/usr/bin/env python3
"""Prepare canonical, private verification roots for an Issue run."""

from __future__ import annotations

import argparse
import json
import math
import os
import re
import stat
import sys
import time
from pathlib import Path

if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from harness.lib import managed_lock_broker as broker


class VerificationError(ValueError):
    """The lease, identifiers, or cache layout is not safe to use."""


_PRIVATE_MODE = 0o700
_OWNER_REPOSITORY = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,254}")
_RUN_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}")
_OPEN_DIR_FLAGS = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC


def _identifier(value: object, label: str, pattern: re.Pattern[str]) -> str:
    if not isinstance(value, str) or not pattern.fullmatch(value) or value in {".", ".."}:
        raise VerificationError(f"invalid {label}")
    return value


def _issue_number(value: object) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
        raise VerificationError("issue must be positive")
    return value


def _absolute(value: os.PathLike[str] | str, label: str) -> Path:
    raw = os.fspath(value)
    path = Path(raw)
    if not isinstance(raw, str) or not path.is_absolute() or "\x00" in raw:
        raise VerificationError(f"{label} must be an absolute path")
    if any(part in {"", ".", ".."} for part in raw.split("/")[1:]):
        raise VerificationError(f"invalid {label}")
    return path


def _open_absolute_directory(path: Path, label: str) -> int:
    """Open every component without following symlinks."""
    parts = path.parts
    if not parts or parts[0] != "/":
        raise VerificationError(f"invalid {label}")
    try:
        fd = os.open(parts[0], _OPEN_DIR_FLAGS)
        for part in parts[1:]:
            nxt = os.open(part, _OPEN_DIR_FLAGS, dir_fd=fd)
            os.close(fd)
            fd = nxt
        return fd
    except OSError as exc:
        try:
            os.close(fd)
        except (OSError, UnboundLocalError):
            pass
        raise VerificationError(f"unsafe {label}") from exc


def _require_private_fd(fd: int, label: str) -> None:
    st = os.fstat(fd)
    if (
        not stat.S_ISDIR(st.st_mode)
        or st.st_uid != os.getuid()
        or stat.S_IMODE(st.st_mode) != _PRIVATE_MODE
    ):
        raise VerificationError(f"unsafe {label}")


def _open_private_child(parent_fd: int, name: str, label: str) -> int:
    try:
        fd = os.open(name, _OPEN_DIR_FLAGS, dir_fd=parent_fd)
    except OSError as exc:
        raise VerificationError(f"unsafe {label}") from exc
    try:
        _require_private_fd(fd, label)
        return fd
    except BaseException:
        os.close(fd)
        raise


def _same_inode(left: os.stat_result, right: os.stat_result) -> bool:
    return (left.st_dev, left.st_ino) == (right.st_dev, right.st_ino)


def _remove_created_directory(parent_fd: int, name: str, created: os.stat_result) -> None:
    """Remove only the still-empty directory inode created by this call."""
    try:
        current = os.stat(name, dir_fd=parent_fd, follow_symlinks=False)
        if not _same_inode(current, created):
            return
        os.rmdir(name, dir_fd=parent_fd)
    except OSError:
        pass


def _create_private_child(parent_fd: int, name: str, label: str) -> int:
    try:
        os.mkdir(name, _PRIVATE_MODE, dir_fd=parent_fd)
        created = os.stat(name, dir_fd=parent_fd, follow_symlinks=False)
    except FileExistsError as exc:
        raise VerificationError(f"existing {label}") from exc
    except OSError as exc:
        raise VerificationError(f"cannot create {label}") from exc
    try:
        if not stat.S_ISDIR(created.st_mode):
            raise VerificationError(f"changed {label}")
        fd = _open_private_child(parent_fd, name, label)
        # Bind both the opened descriptor and the current name to the inode
        # observed immediately after mkdir. A cooperating same-UID process is
        # excluded by the authenticated Issue lease; replacement is fail-closed.
        named = os.stat(name, dir_fd=parent_fd, follow_symlinks=False)
        opened = os.fstat(fd)
        if not (_same_inode(created, opened) and _same_inode(created, named)):
            os.close(fd)
            raise VerificationError(f"changed {label}")
        return fd
    except BaseException:
        # Never chmod, replace, or remove an entry substituted after creation.
        _remove_created_directory(parent_fd, name, created)
        raise


def _lease(handle: Path, root: Path, owner: str, repository: str, issue: int) -> None:
    try:
        metadata = broker.read_handle_metadata(handle)
        response = broker.request_handle(handle, "status")
    except Exception as exc:
        raise VerificationError("broker lease unavailable") from exc
    if not isinstance(metadata, dict) or not isinstance(response, dict):
        raise VerificationError("invalid broker response")
    if response.get("error") or response.get("status") not in {"held", "repository-held"}:
        raise VerificationError("broker lease not held")
    deadline = response.get("deadline")
    if (
        isinstance(deadline, bool)
        or not isinstance(deadline, (int, float))
        or not math.isfinite(deadline)
        or deadline <= time.monotonic()
    ):
        raise VerificationError("broker lease expired")
    for source in (metadata, response):
        if (
            source.get("root") != str(root)
            or source.get("owner") != owner
            or source.get("repository") != repository
            or source.get("issue") != issue
        ):
            raise VerificationError("broker lease binding mismatch")
    if metadata.get("role") != response.get("role") or response.get("role") not in broker.ROLES:
        raise VerificationError("broker lease role mismatch")


def prepare(
    root: os.PathLike[str] | str,
    owner: str,
    repository: str,
    issue: int,
    run_id: str,
    handle: os.PathLike[str] | str,
) -> Path:
    """Authenticate a lease and no-clobber-create one private run directory."""
    owner = _identifier(owner, "owner", _OWNER_REPOSITORY)
    repository = _identifier(repository, "repository", _OWNER_REPOSITORY)
    run_id = _identifier(run_id, "run_id", _RUN_ID)
    issue = _issue_number(issue)
    root = _absolute(root, "root")
    handle = _absolute(handle, "handle")
    _lease(handle, root, owner, repository, issue)

    issue_path = root / "repositories" / owner / repository / "worktrees" / f"issue-{issue}"
    issue_fd = _open_absolute_directory(issue_path, "Issue directory")
    verification_fd = None
    run_fd = None
    created_verification_inode = None
    try:
        _require_private_fd(issue_fd, "Issue directory")
        try:
            verification_fd = _open_private_child(issue_fd, "verification", "verification directory")
        except VerificationError as exc:
            try:
                os.stat("verification", dir_fd=issue_fd, follow_symlinks=False)
            except FileNotFoundError:
                verification_fd = _create_private_child(issue_fd, "verification", "verification directory")
                created_verification_inode = os.fstat(verification_fd)
            else:
                raise exc

        try:
            os.stat(run_id, dir_fd=verification_fd, follow_symlinks=False)
        except FileNotFoundError:
            pass
        except OSError as exc:
            raise VerificationError("cannot inspect run directory") from exc
        else:
            raise VerificationError("existing run directory")

        try:
            run_fd = _create_private_child(verification_fd, run_id, "run directory")
        except BaseException:
            if created_verification_inode is not None:
                _remove_created_directory(
                    issue_fd, "verification", created_verification_inode
                )
            raise
        return issue_path / "verification" / run_id
    finally:
        for fd in (run_fd, verification_fd, issue_fd):
            if fd is not None:
                try:
                    os.close(fd)
                except OSError:
                    pass


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="verification_root")
    sub = parser.add_subparsers(dest="command", required=True)
    command = sub.add_parser("prepare")
    command.add_argument("--root", required=True)
    command.add_argument("--owner", required=True)
    command.add_argument("--repository", required=True)
    command.add_argument("--issue", required=True, type=int)
    command.add_argument("--run-id", required=True)
    command.add_argument("--handle", required=True)
    return parser


def main(argv: list[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    try:
        path = prepare(args.root, args.owner, args.repository, args.issue, args.run_id, args.handle)
        print(json.dumps({"path": str(path), "verification_mode": "0700", "run_mode": "0700"}, separators=(",", ":")))
        return 0
    except (VerificationError, OSError, ValueError) as exc:
        print(json.dumps({"error": str(exc)[:200]}, separators=(",", ":")), file=sys.stderr)
        return 1


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