next-tjwater-agent/scripts/landlock_sandbox.py

#!/usr/bin/env python3
"""Apply a Landlock + seccomp policy, drop privileges, then exec one shell command."""

from __future__ import annotations

import argparse
import ctypes
import ctypes.util
import errno
import json
import os
import platform
import resource
import sys
from pathlib import Path


LANDLOCK_CREATE_RULESET_VERSION = 1
LANDLOCK_RULE_PATH_BENEATH = 1

ACCESS_FS_EXECUTE = 1 << 0
ACCESS_FS_WRITE_FILE = 1 << 1
ACCESS_FS_READ_FILE = 1 << 2
ACCESS_FS_READ_DIR = 1 << 3
ACCESS_FS_REMOVE_DIR = 1 << 4
ACCESS_FS_REMOVE_FILE = 1 << 5
ACCESS_FS_MAKE_CHAR = 1 << 6
ACCESS_FS_MAKE_DIR = 1 << 7
ACCESS_FS_MAKE_REG = 1 << 8
ACCESS_FS_MAKE_SOCK = 1 << 9
ACCESS_FS_MAKE_FIFO = 1 << 10
ACCESS_FS_MAKE_BLOCK = 1 << 11
ACCESS_FS_MAKE_SYM = 1 << 12
ACCESS_FS_REFER = 1 << 13
ACCESS_FS_TRUNCATE = 1 << 14

ACCESS_NET_CONNECT_TCP = 1 << 0
ACCESS_NET_BIND_TCP = 1 << 1

READ_ACCESS = ACCESS_FS_EXECUTE | ACCESS_FS_READ_FILE | ACCESS_FS_READ_DIR
WRITE_ACCESS = (
    ACCESS_FS_WRITE_FILE
    | ACCESS_FS_REMOVE_DIR
    | ACCESS_FS_REMOVE_FILE
    | ACCESS_FS_MAKE_CHAR
    | ACCESS_FS_MAKE_DIR
    | ACCESS_FS_MAKE_REG
    | ACCESS_FS_MAKE_SOCK
    | ACCESS_FS_MAKE_FIFO
    | ACCESS_FS_MAKE_BLOCK
    | ACCESS_FS_MAKE_SYM
    | ACCESS_FS_REFER
    | ACCESS_FS_TRUNCATE
)
HANDLED_FS_ACCESS = READ_ACCESS | WRITE_ACCESS

PR_SET_NO_NEW_PRIVS = 38
AF_INET = 2
AF_INET6 = 10
SCMP_ACT_ALLOW = 0x7FFF0000
SCMP_ACT_ERRNO = 0x00050000 | errno.EPERM
SCMP_CMP_EQ = 4


class RulesetAttr(ctypes.Structure):
    _fields_ = [
        ("handled_access_fs", ctypes.c_uint64),
        ("handled_access_net", ctypes.c_uint64),
    ]


class PathBeneathAttr(ctypes.Structure):
    _fields_ = [
        ("allowed_access", ctypes.c_uint64),
        ("parent_fd", ctypes.c_int32),
    ]


class ScmpArgCmp(ctypes.Structure):
    _fields_ = [
        ("arg", ctypes.c_uint32),
        ("op", ctypes.c_uint32),
        ("datum_a", ctypes.c_uint64),
        ("datum_b", ctypes.c_uint64),
    ]


def syscall_numbers() -> tuple[int, int, int]:
    machine = platform.machine().lower()
    if machine not in {"x86_64", "amd64", "aarch64", "arm64"}:
        raise RuntimeError(f"unsupported architecture for Landlock syscalls: {machine}")
    return 444, 445, 446


def checked_syscall(libc: ctypes.CDLL, number: int, *args: object) -> int:
    result = int(libc.syscall(number, *args))
    if result < 0:
        error_number = ctypes.get_errno()
        raise OSError(error_number, os.strerror(error_number))
    return result


def get_landlock_abi(libc: ctypes.CDLL) -> int:
    create_ruleset, _, _ = syscall_numbers()
    return checked_syscall(
        libc,
        create_ruleset,
        ctypes.c_void_p(),
        ctypes.c_size_t(0),
        ctypes.c_uint32(LANDLOCK_CREATE_RULESET_VERSION),
    )


def load_seccomp() -> ctypes.CDLL:
    library_name = ctypes.util.find_library("seccomp") or "libseccomp.so.2"
    library = ctypes.CDLL(library_name, use_errno=True)
    library.seccomp_init.argtypes = [ctypes.c_uint32]
    library.seccomp_init.restype = ctypes.c_void_p
    library.seccomp_release.argtypes = [ctypes.c_void_p]
    library.seccomp_syscall_resolve_name.argtypes = [ctypes.c_char_p]
    library.seccomp_syscall_resolve_name.restype = ctypes.c_int
    library.seccomp_rule_add_array.argtypes = [
        ctypes.c_void_p,
        ctypes.c_uint32,
        ctypes.c_int,
        ctypes.c_uint,
        ctypes.POINTER(ScmpArgCmp),
    ]
    library.seccomp_rule_add_array.restype = ctypes.c_int
    library.seccomp_load.argtypes = [ctypes.c_void_p]
    library.seccomp_load.restype = ctypes.c_int
    return library


def add_path_rule(
    libc: ctypes.CDLL,
    add_rule_number: int,
    ruleset_fd: int,
    path: str,
    access: int,
) -> None:
    resolved_path = os.path.realpath(path)
    if not os.path.exists(resolved_path):
        return
    path_fd = os.open(resolved_path, os.O_PATH | os.O_CLOEXEC)
    try:
        allowed_access = access
        if not Path(resolved_path).is_dir():
            allowed_access &= ACCESS_FS_EXECUTE | ACCESS_FS_READ_FILE | ACCESS_FS_WRITE_FILE | ACCESS_FS_TRUNCATE
        attribute = PathBeneathAttr(
            allowed_access=allowed_access,
            parent_fd=path_fd,
        )
        checked_syscall(
            libc,
            add_rule_number,
            ctypes.c_int(ruleset_fd),
            ctypes.c_int(LANDLOCK_RULE_PATH_BENEATH),
            ctypes.byref(attribute),
            ctypes.c_uint32(0),
        )
    finally:
        os.close(path_fd)


def build_landlock_ruleset(
    libc: ctypes.CDLL,
    abi: int,
    workspace: str,
    read_only_paths: list[str],
    writable_paths: list[str],
) -> int:
    create_ruleset, add_rule, _ = syscall_numbers()
    handled_network = (
        ACCESS_NET_CONNECT_TCP | ACCESS_NET_BIND_TCP if abi >= 4 else 0
    )
    ruleset_attribute = RulesetAttr(
        handled_access_fs=HANDLED_FS_ACCESS,
        handled_access_net=handled_network,
    )
    ruleset_fd = checked_syscall(
        libc,
        create_ruleset,
        ctypes.byref(ruleset_attribute),
        ctypes.sizeof(ruleset_attribute),
        ctypes.c_uint32(0),
    )
    try:
        add_path_rule(
            libc,
            add_rule,
            ruleset_fd,
            workspace,
            READ_ACCESS,
        )
        for path in writable_paths:
            add_path_rule(libc, add_rule, ruleset_fd, path, HANDLED_FS_ACCESS)
        for path in read_only_paths:
            add_path_rule(libc, add_rule, ruleset_fd, path, READ_ACCESS)
        for device_path, access in (
            ("/dev/null", ACCESS_FS_READ_FILE | ACCESS_FS_WRITE_FILE),
            ("/dev/zero", ACCESS_FS_READ_FILE | ACCESS_FS_WRITE_FILE),
            ("/dev/urandom", ACCESS_FS_READ_FILE),
            ("/dev/random", ACCESS_FS_READ_FILE),
        ):
            add_path_rule(libc, add_rule, ruleset_fd, device_path, access)
    except Exception:
        os.close(ruleset_fd)
        raise
    return ruleset_fd


def install_seccomp_network_filter(library: ctypes.CDLL) -> None:
    context = library.seccomp_init(SCMP_ACT_ALLOW)
    if not context:
        raise RuntimeError("seccomp_init failed")
    try:
        socket_syscall = library.seccomp_syscall_resolve_name(b"socket")
        if socket_syscall < 0:
            raise RuntimeError("could not resolve socket syscall")
        # Deny all socket families, process inspection/signalling, and process
        # group escape. The parent (outside this filter) owns cancellation.
        for syscall_name in (
            b"socket", b"socketpair", b"ptrace", b"process_vm_readv",
            b"process_vm_writev", b"pidfd_getfd", b"pidfd_send_signal",
            b"kill", b"tkill", b"tgkill", b"setsid", b"setpgid",
        ):
            syscall_number = library.seccomp_syscall_resolve_name(syscall_name)
            if syscall_number < 0:
                continue
            result = library.seccomp_rule_add_array(
                context,
                SCMP_ACT_ERRNO,
                syscall_number,
                0,
                None,
            )
            if result != 0:
                raise OSError(-result, os.strerror(-result))
        result = library.seccomp_load(context)
        if result != 0:
            raise OSError(-result, os.strerror(-result))
    finally:
        library.seccomp_release(context)


def set_no_new_privileges(libc: ctypes.CDLL) -> None:
    result = libc.prctl(PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0)
    if result != 0:
        error_number = ctypes.get_errno()
        raise OSError(error_number, os.strerror(error_number))


def restrict_process(
    workspace: str,
    read_only_paths: list[str],
    writable_paths: list[str],
    uid: int,
    gid: int,
) -> int:
    libc = ctypes.CDLL(None, use_errno=True)
    libc.syscall.restype = ctypes.c_long
    libc.prctl.restype = ctypes.c_int
    seccomp = load_seccomp()
    abi = get_landlock_abi(libc)
    if abi < 4:
        raise RuntimeError(f"Landlock ABI 4 or newer is required; detected ABI {abi}")

    ruleset_fd = build_landlock_ruleset(
        libc,
        abi,
        workspace,
        read_only_paths,
        writable_paths,
    )
    try:
        if os.geteuid() == 0:
            os.setgroups([])
            os.setgid(gid)
            os.setuid(uid)
        set_no_new_privileges(libc)
        _, _, restrict_self = syscall_numbers()
        checked_syscall(
            libc,
            restrict_self,
            ctypes.c_int(ruleset_fd),
            ctypes.c_uint32(0),
        )
        install_seccomp_network_filter(seccomp)
    finally:
        os.close(ruleset_fd)
    return abi


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--probe", action="store_true")
    parser.add_argument("--workspace")
    parser.add_argument("--read-only", action="append", default=[])
    parser.add_argument("--writable", action="append", default=[])
    parser.add_argument("--uid", type=int, default=10_001)
    parser.add_argument("--gid", type=int, default=10_001)
    parser.add_argument("--command")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    libc = ctypes.CDLL(None, use_errno=True)
    libc.syscall.restype = ctypes.c_long
    if args.probe:
        abi = get_landlock_abi(libc)
        load_seccomp()
        print(json.dumps({"ok": True, "landlock_abi": abi, "seccomp": True}))
        return 0 if abi >= 4 else 1

    if not args.workspace or args.command is None:
        raise RuntimeError("workspace and command are required")
    workspace = os.path.realpath(args.workspace)
    if not os.path.isdir(workspace):
        raise RuntimeError("workspace must be an existing directory")
    resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
    resource.setrlimit(resource.RLIMIT_FSIZE, (128 * 1024 * 1024, 128 * 1024 * 1024))
    resource.setrlimit(resource.RLIMIT_NOFILE, (256, 256))
    restrict_process(workspace, args.read_only, args.writable, args.uid, args.gid)
    os.execve("/bin/bash", ["bash", "-c", args.command], dict(os.environ))
    return 127


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except Exception as error:
        print(
            json.dumps(
                {
                    "ok": False,
                    "error": "SANDBOX_UNAVAILABLE",
                    "message": str(error),
                }
            ),
            file=sys.stderr,
        )
        raise SystemExit(125)