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."""
    ...

dnsmasq_via_nsenter(container, conf_path, *, binary, pid=None)

Launch the dnsmasq at binary inside a running container's network namespace.

Source code in src/terok_shield/run.py

def dnsmasq_via_nsenter(
    self, container: str, conf_path: str, *, binary: str, pid: str | None = None
) -> str:
    """Launch the dnsmasq at *binary* 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."""
    ...

lookup_all(domain, *, timeout=10)

Resolve domain to both IPv4 and IPv6 addresses.

Source code in src/terok_shield/run.py

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

getent_hosts(domain, *, timeout=10)

Resolve domain via getent hosts (fallback when no lookup tool exists).

Source code in src/terok_shield/run.py

def getent_hosts(self, domain: str, *, timeout: int = 10) -> list[str]:
    """Resolve domain via ``getent hosts`` (fallback when no lookup tool exists)."""
    ...

SubprocessRunner()

Default CommandRunner implementation using subprocess.run.

Checks nft availability at construction time and raises NftNotFoundError immediately if nft is not installed.

Check nft availability, raising NftNotFoundError if missing.

Source code in src/terok_shield/run.py

def __init__(self) -> None:
    """Check nft availability, raising NftNotFoundError if missing."""
    if not find_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(
            [require_host_tool(cmd[0]), *cmd[1:]],
            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 when the current host PATH supplies an executable.

Source code in src/terok_shield/run.py

def has(self, name: str) -> bool:
    """Return True when the current host PATH supplies an executable."""
    return bool(find_host_tool(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(["nft", *args, "-f", "-"], stdin=stdin, check=check)
    return self.run(["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",
        "nft",
    ]
    try:
        cmd[2] = require_host_tool("nsenter")
        cmd[-1] = require_host_tool("nft")
    except FileNotFoundError as exc:
        if check:
            raise ExecError([*cmd, *args], 127, str(exc)) from exc
        return ""
    if stdin is not None:
        return self.run([*cmd, *args, "-f", "-"], stdin=stdin, check=check)
    return self.run([*cmd, *args], check=check)

dnsmasq_via_nsenter(container, conf_path, *, binary, pid=None)

Launch the dnsmasq at binary inside a running container's network namespace.

dnsmasq runs in the host PID namespace but the container's network namespace (like the OCI hook's own launch), so a host-side reload can relaunch it here. dnsmasq daemonizes and writes its own pid-file; the command returns once it has forked into the background.

Source code in src/terok_shield/run.py

def dnsmasq_via_nsenter(
    self, container: str, conf_path: str, *, binary: str, pid: str | None = None
) -> str:
    """Launch the dnsmasq at *binary* inside a running container's network namespace.

    dnsmasq runs in the host PID namespace but the container's network
    namespace (like the OCI hook's own launch), so a host-side reload can
    relaunch it here.  dnsmasq daemonizes and writes its own pid-file; the
    command returns once it has forked into the background.
    """
    if pid is None:
        pid = self.podman_inspect(container, "{{.State.Pid}}")
    cmd = ["podman", "unshare", "nsenter", "-t", pid, "-n", binary, f"--conf-file={conf_path}"]
    try:
        cmd[2] = require_host_tool("nsenter")
    except FileNotFoundError as exc:
        raise ExecError(cmd, 127, str(exc)) from exc
    return self.run(cmd)

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()

lookup_all(domain, *, timeout=10)

Resolve domain to both IPv4 and IPv6 addresses with the host's lookup tool.

Prefers dig (one two-type query) and falls back to drill (ldns, the Arch/Manjaro default; one type per invocation). Validates each output line with ipaddress. Returns an empty list on lookup failure or timeout.

Raises:

Type	Description
LookupToolNotFoundError	If neither tool is installed.

Source code in src/terok_shield/run.py

def lookup_all(self, domain: str, *, timeout: int = 10) -> list[str]:
    """Resolve domain to both IPv4 and IPv6 addresses with the host's lookup tool.

    Prefers ``dig`` (one two-type query) and falls back to ``drill``
    (ldns, the Arch/Manjaro default; one type per invocation).
    Validates each output line with ``ipaddress``.  Returns an empty
    list on lookup failure or timeout.

    Raises:
        LookupToolNotFoundError: If neither tool is installed.
    """
    if self.has("dig"):
        out = self.run(
            ["dig", "+short", domain, "A", domain, "AAAA"],
            check=False,
            timeout=timeout,
        )
    elif self.has("drill"):
        out = "\n".join(
            self.run(["drill", "-Q", domain, rrtype], check=False, timeout=timeout)
            for rrtype in ("A", "AAAA")
        )
    else:
        raise LookupToolNotFoundError(
            "no DNS lookup tool found. Install one:\n"
            "  Debian/Ubuntu: sudo apt install dnsutils\n"
            "  Fedora/RHEL:   sudo dnf install bind-utils\n"
            "  Arch/Manjaro:  sudo pacman -S ldns   (drill) or bind (dig)"
        )
    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, *, timeout=10)

Resolve domain via NSS (fallback when the lookup tool 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 lookup tool crashes -- an allowed literal-IPv4 target then hit the terminal reject as "Host is unreachable" (terok#1119).

timeout bounds each family query; on expiry that family yields nothing (best-effort, matching lookup_all).

Source code in src/terok_shield/run.py

def getent_hosts(self, domain: str, *, timeout: int = 10) -> list[str]:
    """Resolve domain via NSS (fallback when the lookup tool 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 lookup tool
    crashes -- an allowed literal-IPv4 target then hit the terminal
    reject as "Host is unreachable" (terok#1119).

    *timeout* bounds each family query; on expiry that family yields
    nothing (best-effort, matching ``lookup_all``).
    """
    result: list[str] = []
    for database in ("ahostsv4", "ahostsv6"):
        out = self.run(["getent", database, domain], check=False, timeout=timeout)
        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.

LookupToolNotFoundError

Bases: RuntimeError

Raised when neither dig nor drill is found on the host.

ShieldNeedsSetup

Bases: SetupRequiredError

Raised when host configuration cannot support the requested Shield policy.

The message includes the DNS-tier limitation or unavailable configured binary and the corresponding setup remedies.

find_nft()

Locate nft using the same current PATH as every other host tool.

Source code in src/terok_shield/run.py

def find_nft() -> str:
    """Locate nft using the same current PATH as every other host tool."""
    return find_host_tool("nft") or ""