run

run

Subprocess execution boundary for all external commands.

Every shell-out in terok-shield flows through the CommandRunner protocol. Production code uses SubprocessRunner; tests inject fakes. This keeps external dependencies auditable and mockable in one place.

CommandRunner

Bases: Protocol

Protocol for executing external commands.

Decouples all subprocess calls behind a testable interface.

run(cmd, *, check=True, stdin=None, timeout=None)

Run a command, return stdout.

Source code in src/terok_shield/run.py

def run(
    self,
    cmd: list[str],
    *,
    check: bool = True,
    stdin: str | None = None,
    timeout: int | None = None,
) -> str:
    """Run a command, return stdout."""
    ...

has(name)

Return True if an executable is on PATH.

Source code in src/terok_shield/run.py

def has(self, name: str) -> bool:
    """Return True if an executable is on PATH."""
    ...

nft(*args, stdin=None, check=True)

Run nft command directly (inside container netns).

Source code in src/terok_shield/run.py

def nft(self, *args: str, stdin: str | None = None, check: bool = True) -> str:
    """Run nft command directly (inside container netns)."""
    ...

nft_via_nsenter(container, *args, pid=None, stdin=None, check=True)

Run nft inside a running container's network namespace.

Source code in src/terok_shield/run.py

def nft_via_nsenter(
    self,
    container: str,
    *args: str,
    pid: str | None = None,
    stdin: str | None = None,
    check: bool = True,
) -> str:
    """Run nft inside a running container's network namespace."""
    ...

podman_inspect(container, fmt)

Inspect a container attribute via podman.

Source code in src/terok_shield/run.py

def podman_inspect(self, container: str, fmt: str) -> str:
    """Inspect a container attribute via podman."""
    ...

dig_all(domain, *, timeout=10)

Resolve domain to both IPv4 and IPv6 addresses.

Source code in src/terok_shield/run.py

def dig_all(self, domain: str, *, timeout: int = 10) -> list[str]:
    """Resolve domain to both IPv4 and IPv6 addresses."""
    ...

getent_hosts(domain)

Resolve domain via getent hosts (fallback when dig is missing).

Source code in src/terok_shield/run.py

def getent_hosts(self, domain: str) -> list[str]:
    """Resolve domain via ``getent hosts`` (fallback when dig is missing)."""
    ...

SubprocessRunner()

Default CommandRunner implementation using subprocess.run.

Resolves the nft binary path at construction time and raises NftNotFoundError immediately if nft is not installed.

Resolve the nft binary path, raising NftNotFoundError if missing.

Source code in src/terok_shield/run.py

def __init__(self) -> None:
    """Resolve the nft binary path, raising NftNotFoundError if missing."""
    self._has_cache: dict[str, bool] = {}
    self._nft = find_nft()
    if not self._nft:
        raise NftNotFoundError(
            "nft binary not found. Install nftables:\n"
            "  Debian/Ubuntu: sudo apt install nftables\n"
            "  Fedora/RHEL:   sudo dnf install nftables\n"
            "  Arch:          sudo pacman -S nftables"
        )

run(cmd, *, check=True, stdin=None, timeout=None)

Run a command, return stdout. Raise ExecError on failure when check=True.

Source code in src/terok_shield/run.py

def run(
    self,
    cmd: list[str],
    *,
    check: bool = True,
    stdin: str | None = None,
    timeout: int | None = None,
) -> str:
    """Run a command, return stdout.  Raise ExecError on failure when check=True."""
    try:
        # Explicit argv list with shell=False — auditable and testable
        r = subprocess.run(
            cmd,
            input=stdin,
            capture_output=True,
            text=True,
            timeout=timeout,
            shell=False,  # nosec B603
        )
    except FileNotFoundError as e:
        if check:
            raise ExecError(cmd, 127, str(e)) from e
        return ""
    except subprocess.TimeoutExpired as e:
        if check:
            raise ExecError(cmd, -1, f"timed out after {timeout}s") from e
        return ""
    if check and r.returncode != 0:
        raise ExecError(cmd, r.returncode, r.stderr or "")
    return r.stdout or ""

has(name)

Return True if an executable is on PATH or a sbin dir (cached).

sbin-aware for the same reason as find_nft: Debian-family login shells omit /usr/sbin, and a plain which() there would silently downgrade the DNS tier by "missing" dnsmasq.

Source code in src/terok_shield/run.py

def has(self, name: str) -> bool:
    """Return True if an executable is on PATH or a sbin dir (cached).

    sbin-aware for the same reason as [`find_nft`][terok_shield.run.find_nft]: Debian-family login
    shells omit /usr/sbin, and a plain which() there would silently
    downgrade the DNS tier by "missing" dnsmasq.
    """
    if name not in self._has_cache:
        self._has_cache[name] = bool(which_sbin_aware(name))
    return self._has_cache[name]

nft(*args, stdin=None, check=True)

Run nft command directly (hook mode, inside container netns).

Source code in src/terok_shield/run.py

def nft(self, *args: str, stdin: str | None = None, check: bool = True) -> str:
    """Run nft command directly (hook mode, inside container netns)."""
    if stdin is not None:
        return self.run([self._nft, *args, "-f", "-"], stdin=stdin, check=check)
    return self.run([self._nft, *args], check=check)

nft_via_nsenter(container, *args, pid=None, stdin=None, check=True)

Run nft inside a running container's network namespace.

Source code in src/terok_shield/run.py

def nft_via_nsenter(
    self,
    container: str,
    *args: str,
    pid: str | None = None,
    stdin: str | None = None,
    check: bool = True,
) -> str:
    """Run nft inside a running container's network namespace."""
    if pid is None:
        pid = self.podman_inspect(container, "{{.State.Pid}}")
    cmd = ["podman", "unshare", "nsenter", "-t", pid, "-n", self._nft]
    if stdin is not None:
        return self.run([*cmd, *args, "-f", "-"], stdin=stdin, check=check)
    return self.run([*cmd, *args], check=check)

podman_inspect(container, fmt)

Inspect a container attribute via podman.

Source code in src/terok_shield/run.py

def podman_inspect(self, container: str, fmt: str) -> str:
    """Inspect a container attribute via podman."""
    return self.run(["podman", "inspect", "--format", fmt, container]).strip()

dig_all(domain, *, timeout=10)

Resolve domain to both IPv4 and IPv6 addresses in a single query.

Runs dig +short domain A domain AAAA and validates each line with ipaddress. Returns empty list on lookup failure or timeout.

Raises:

Type	Description
DigNotFoundError	If dig is not installed.

Source code in src/terok_shield/run.py

def dig_all(self, domain: str, *, timeout: int = 10) -> list[str]:
    """Resolve domain to both IPv4 and IPv6 addresses in a single query.

    Runs ``dig +short domain A domain AAAA`` and validates each line
    with ``ipaddress``.  Returns empty list on lookup failure or
    timeout.

    Raises:
        DigNotFoundError: If ``dig`` is not installed.
    """
    if not self.has("dig"):
        raise DigNotFoundError(
            "dig binary not found. Install DNS utilities:\n"
            "  Debian/Ubuntu: sudo apt install dnsutils\n"
            "  Fedora/RHEL:   sudo dnf install bind-utils\n"
            "  Arch:          sudo pacman -S bind"
        )
    out = self.run(
        ["dig", "+short", domain, "A", domain, "AAAA"],
        check=False,
        timeout=timeout,
    )
    result: list[str] = []
    for line in out.splitlines():
        addr = line.strip()
        if not addr:
            continue
        try:
            _ipaddress.ip_address(addr)
            result.append(addr)
        except ValueError:
            continue
    return result

getent_hosts(domain)

Resolve domain via NSS (fallback when dig is missing or broken).

Queries both address families explicitly: plain getent hosts stops at the first family glibc resolves (AAAA for dual-stack names), which left allow_v4 empty on the one host whose dig crashes -- an allowed literal-IPv4 target then hit the terminal reject as "Host is unreachable" (terok#1119).

Source code in src/terok_shield/run.py

def getent_hosts(self, domain: str) -> list[str]:
    """Resolve domain via NSS (fallback when dig is missing or broken).

    Queries both address families explicitly: plain ``getent hosts``
    stops at the first family glibc resolves (AAAA for dual-stack
    names), which left ``allow_v4`` empty on the one host whose dig
    crashes -- an allowed literal-IPv4 target then hit the terminal
    reject as "Host is unreachable" (terok#1119).
    """
    result: list[str] = []
    for database in ("ahostsv4", "ahostsv6"):
        out = self.run(["getent", database, domain], check=False, timeout=10)
        for line in out.splitlines():
            parts = line.strip().split()
            if len(parts) < 2 or parts[1] != "STREAM":
                continue
            try:
                _ipaddress.ip_address(parts[0])
            except ValueError:
                continue
            if parts[0] not in result:
                result.append(parts[0])
    return result

ExecError(cmd, rc, stderr)

Bases: Exception

Raised when a subprocess fails.

Store command details and format the error message.

Source code in src/terok_shield/run.py

def __init__(self, cmd: list[str], rc: int, stderr: str) -> None:
    """Store command details and format the error message."""
    self.cmd = cmd
    self.rc = rc
    self.stderr = stderr
    super().__init__(f"{cmd!r} failed (rc={rc}): {stderr.strip()}")

cmd = cmd instance-attribute

rc = rc instance-attribute

stderr = stderr instance-attribute

NftNotFoundError

Bases: RuntimeError

Raised when the nft binary is not found on the host.

DigNotFoundError

Bases: RuntimeError

Raised when the dig binary is not found on the host.

DNS resolution requires dig (from bind-utils / dnsutils).

ShieldNeedsSetup

Bases: RuntimeError

Raised when global OCI hooks are not installed.

Per-container --hooks-dir does not persist across container restarts, so global hooks are required. The message includes system-specific setup hints.

which_sbin_aware(name)

Resolve name like shutil.which, falling back to the sbin dirs.

Login shells on the Debian family exclude /usr/sbin from PATH, so a plain which() misses sbin-installed daemons (dnsmasq) — and the DNS tier silently downgrades to dig. shutil.which with an explicit path= keeps the executability check identical to PATH resolution.

Source code in src/terok_shield/run.py

def which_sbin_aware(name: str) -> str:
    """Resolve *name* like ``shutil.which``, falling back to the sbin dirs.

    Login shells on the Debian family exclude ``/usr/sbin`` from PATH, so
    a plain which() misses sbin-installed daemons (dnsmasq) — and the DNS
    tier silently downgrades to dig.  ``shutil.which`` with an explicit
    ``path=`` keeps the executability check identical to PATH resolution.
    """
    for search_path in (None, *_SBIN_DIRS):
        found = shutil.which(name, path=search_path)
        if found:
            return found
    return ""

find_nft()

Locate the nft binary, checking PATH then common sbin directories.

sbin directories are checked explicitly because rootless users often lack them in PATH. Returns empty string if not found.

Source code in src/terok_shield/run.py

def find_nft() -> str:
    """Locate the nft binary, checking PATH then common sbin directories.

    sbin directories are checked explicitly because rootless users often
    lack them in PATH.  Returns empty string if not found.
    """
    return which_sbin_aware("nft")