openlatch-client 0.5.3

#!/usr/bin/env python3
"""Fail a Linux release binary that a bare or older host cannot load.

`curl -LsSf https://openlatch.ai/install.sh | sh` promises no runtime, and Mode 2
runs the daemon inside the agent's own container, which is often a `*-slim`
image. The dynamic loader decides whether the binary starts at all, before
`main` and before any fallback in our code can run. It refuses on two things:

  * a `DT_NEEDED` library the host does not have. 0.5.2 linked
    libdbus-1.so.3 (keyring's `sync-secret-service`), absent from
    `ubuntu:24.04`, `debian:*-slim` and most container bases.
  * a `GLIBC_x.y` symbol version newer than the host's glibc. 0.5.2's x86_64
    binary was built natively on ubuntu-latest and needed GLIBC_2.39, so it
    refused Debian 12, Ubuntu 22.04, RHEL 9 and Amazon Linux 2023.

Neither shows on a CI runner, which has both. This reads the ELF itself — the
program headers the loader reads, not `readelf` — so it runs anywhere:

  * every `DT_NEEDED` must be in ALLOWED, the glibc family plus libgcc_s,
    which every glibc distribution ships;
  * the highest `GLIBC_x.y` in `.gnu.version_r` must not exceed `--max-glibc`
    (default 2.28: RHEL 8, Debian 10, Ubuntu 20.04, Amazon Linux 2023).

Exit codes: 0 loadable, 1 violations found, 2 bad invocation or not an ELF.
"""

from __future__ import annotations

import argparse
import re
import struct
import sys
from pathlib import Path

ALLOWED = frozenset({
    "libc.so.6",
    "libm.so.6",
    "libgcc_s.so.1",
    "libdl.so.2",
    "libpthread.so.0",
    "librt.so.1",
    "libutil.so.1",
    "ld-linux-x86-64.so.2",
    "ld-linux-aarch64.so.1",
})

PT_LOAD, PT_DYNAMIC = 1, 2
DT_NULL, DT_NEEDED, DT_STRTAB, DT_STRSZ = 0, 1, 5, 10
DT_VERNEED, DT_VERNEEDNUM = 0x6FFFFFFE, 0x6FFFFFFF
GLIBC_VERSION = re.compile(r"GLIBC_(\d+(?:\.\d+)+)")


class NotElf(ValueError):
    """The file is not a 64-bit little-endian ELF this gate knows how to read."""


def parse_version(text: str) -> tuple[int, ...]:
    return tuple(int(part) for part in text.split("."))


def read_dynamic(data: bytes) -> tuple[list[str], list[tuple[str, str]]]:
    """Return (DT_NEEDED names, [(library, version) needed]) as the loader sees them."""
    if len(data) < 64 or data[:4] != b"\x7fELF":
        raise NotElf("not an ELF file")
    if data[4] != 2 or data[5] != 1:
        raise NotElf("not a 64-bit little-endian ELF")
    phoff, = struct.unpack_from("<Q", data, 0x20)
    phentsize, phnum = struct.unpack_from("<HH", data, 0x36)

    loads, dynamic = [], None
    for i in range(phnum):
        p_type, _, p_offset, p_vaddr, _, p_filesz = struct.unpack_from(
            "<IIQQQQ", data, phoff + i * phentsize
        )
        if p_type == PT_LOAD:
            loads.append((p_vaddr, p_offset, p_filesz))
        elif p_type == PT_DYNAMIC:
            dynamic = (p_offset, p_filesz)
    if dynamic is None:
        return [], []  # static: nothing for the loader to find

    def offset_of(vaddr: int) -> int:
        for start, offset, size in loads:
            if start <= vaddr < start + size:
                return offset + vaddr - start
        raise NotElf(f"address {vaddr:#x} is outside every PT_LOAD segment")

    entries = []
    for pos in range(dynamic[0], dynamic[0] + dynamic[1], 16):
        tag, value = struct.unpack_from("<qQ", data, pos)
        if tag == DT_NULL:
            break
        entries.append((tag, value))
    tags = dict(entries)
    strtab = offset_of(tags[DT_STRTAB])

    def string(index: int) -> str:
        end = data.index(b"\0", strtab + index)
        return data[strtab + index:end].decode("utf-8", "replace")

    needed = [string(value) for tag, value in entries if tag == DT_NEEDED]

    versions = []
    if DT_VERNEED in tags:
        verneed = offset_of(tags[DT_VERNEED])
        for _ in range(tags.get(DT_VERNEEDNUM, 0)):
            _, vn_cnt, vn_file, vn_aux, vn_next = struct.unpack_from("<HHIII", data, verneed)
            aux = verneed + vn_aux
            for _ in range(vn_cnt):
                _, _, _, vna_name, vna_next = struct.unpack_from("<IHHII", data, aux)
                versions.append((string(vn_file), string(vna_name)))
                aux += vna_next
            verneed += vn_next
    return needed, versions


def check(path: Path, ceiling: tuple[int, ...]) -> tuple[list[str], str]:
    """Every reason the loader of an older or slimmer host would refuse `path`, plus a
    one-line account of what it does need."""
    needed, versions = read_dynamic(path.read_bytes())
    errors = [
        f"{path} links {lib}, which bare and slim Linux hosts do not have: the loader "
        f"refuses the binary before main. Find the crate with `cargo tree -i` for the "
        f"Linux target (libdbus-1.so.3 means keyring's `sync-secret-service` is back: "
        f"see the comment on `keyring` in Cargo.toml)."
        for lib in needed
        if lib not in ALLOWED
    ]
    highest: dict[str, tuple[int, ...]] = {}  # per library
    for lib, version in sorted(set(versions)):
        if not version.startswith("GLIBC_"):
            continue  # GCC_x.y from libgcc_s: stable since GCC 4
        match = GLIBC_VERSION.fullmatch(version)
        if match is None:
            errors.append(f"{path} needs {version} from {lib}, a private glibc ABI no other "
                          f"glibc release provides.")
        else:
            highest[lib] = max(highest.get(lib, ()), parse_version(match.group(1)))

    def glibc(version: tuple[int, ...]) -> str:
        return "GLIBC_" + ".".join(map(str, version))

    too_new = [f"{glibc(v)} from {lib}" for lib, v in highest.items() if v > ceiling]
    if too_new:
        errors.append(
            f"{path} needs {', '.join(too_new)}, above the {glibc(ceiling)} ceiling: it will "
            f"not start on older distributions. Build it inside cross's image (the "
            f"`cross: true` matrix rows and Cross.toml), not natively on the runner."
        )
    top = glibc(max(highest.values())) if highest else "no versioned glibc symbol"
    return errors, f"needs {', '.join(needed) or 'no shared library'}; at most {top}"


def main(argv: list[str] | None = None) -> int:
    for stream in (sys.stdout, sys.stderr):
        reconfigure = getattr(stream, "reconfigure", None)
        if reconfigure is not None:
            reconfigure(encoding="utf-8", errors="replace")

    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("binaries", nargs="+", type=Path, help="Linux ELF binaries to check")
    parser.add_argument("--max-glibc", default="2.28", metavar="X.Y",
                        help="highest GLIBC_x.y symbol version allowed (default: %(default)s)")
    args = parser.parse_args(argv)
    if not re.fullmatch(r"\d+(\.\d+)+", args.max_glibc):
        parser.error(f"--max-glibc must look like 2.28, not {args.max_glibc!r}")
    ceiling = parse_version(args.max_glibc)

    failed = False
    for path in args.binaries:
        try:
            errors, summary = check(path, ceiling)
        except (OSError, KeyError, ValueError, struct.error) as exc:  # NotElf is a ValueError
            print(f"::error::{path}: cannot read as a Linux ELF: {exc}")
            return 2
        for error in errors:
            print(f"::error::{error}")
        failed = failed or bool(errors)
        print(f"{path}: {summary} (ceiling GLIBC_{args.max_glibc}).")
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())