diff --git a/for_me.md b/for_me.md index 3962882..aa6bbef 100644 --- a/for_me.md +++ b/for_me.md @@ -36,3 +36,14 @@ xxHash64 (XXH64): относительно простая имплементац ## free free не делает очистку самой структуры, valgrind + +## data + +алфавит для символ библиотек: + +``` +.0123456789@ABCDEFGHIJKLMNOPQRSTUVWXYZ_abcdefghijklmnopqrstuvwxyz +``` + +- предоставленные/определенные символы из библиотек обычно используют @@VERSION +- обязательные/неопределенные символы из двоичных файлов обычно используют @VERSION diff --git a/other/extract_alt_set_symbols.py b/other/extract_alt_set_symbols.py new file mode 100755 index 0000000..83ef5ef --- /dev/null +++ b/other/extract_alt_set_symbols.py @@ -0,0 +1,384 @@ +#!/usr/bin/env python3 +"""Extract ALT Linux set:version strings and optional ELF nm symbol stats. + +The script uses the public rdb.altlinux.org API to read package dependency +metadata. With --download-nm it also downloads binary RPMs, extracts ELF files +with bsdtar, and runs nm on dynamic symbols. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import shutil +import statistics +import subprocess +import sys +import tempfile +import urllib.error +import urllib.parse +import urllib.request +from collections import Counter +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable + +API_BASE = "https://rdb.altlinux.org/api" +DEFAULT_PACKAGES = [ + "glibc-core", + "libcrypto3", + "libssl3", + "zlib", + "libgcc1", + "libcurl", + "libsystemd", + "libqt6-core", + "libgtk+3", + "libsqlite3", + "libxml2", + "libX11", + "libxcb", + "coreutils", + "curl", + "openssl", + "systemd", + "python3-base", +] + +SET_PREFIX = "set:" +SET_ALPHABET = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ" +IDENTISH_SYMBOL_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*(?:@@?[A-Za-z0-9_.]+)?$") + + +@dataclass(frozen=True) +class SetDependency: + package: str + dep_type: str + name: str + set_string: str + + @property + def payload(self) -> str: + return self.set_string[len(SET_PREFIX) :] + + @property + def bpp(self) -> int | None: + return set_char_to_int(self.payload[0]) if self.payload else None + + @property + def mshift(self) -> int | None: + return set_char_to_int(self.payload[1]) if len(self.payload) > 1 else None + + +@dataclass +class NmReport: + package: str + mode: str + files_seen: int + symbols: list[str] + file_examples: list[str] + + +def set_char_to_int(char: str) -> int | None: + try: + return SET_ALPHABET.index(char) + except ValueError: + return None + + +def api_json(path: str, params: dict[str, object] | None = None) -> dict: + url = API_BASE + path + if params: + url += "?" + urllib.parse.urlencode(params, doseq=True) + req = urllib.request.Request(url, headers={"User-Agent": "arsv-set-symbol-extractor/1.0"}) + with urllib.request.urlopen(req, timeout=60) as response: + return json.load(response) + + +def get_pkghash(package: str, branch: str, arch: str) -> str: + data = api_json( + "/site/pkghash_by_binary_name", + {"branch": branch, "name": package, "arch": arch}, + ) + return str(data["pkghash"]) + + +def get_set_dependencies(package: str, branch: str, arch: str) -> tuple[str, list[SetDependency]]: + pkghash = get_pkghash(package, branch, arch) + deps = api_json(f"/dependencies/binary_package_dependencies/{pkghash}")["dependencies"] + set_deps = [] + for dep in deps: + version = dep.get("version") or "" + if version.startswith(SET_PREFIX): + set_deps.append( + SetDependency( + package=package, + dep_type=dep.get("type", ""), + name=dep.get("name", ""), + set_string=version, + ) + ) + return pkghash, set_deps + + +def package_download_url(pkghash: str, branch: str, arch: str) -> str: + data = api_json( + f"/site/package_downloads_bin/{pkghash}", + {"branch": branch, "arch": arch}, + ) + downloads = data.get("downloads") or [] + if not downloads or not downloads[0].get("packages"): + raise RuntimeError(f"no download URL for pkghash={pkghash}") + return downloads[0]["packages"][0]["url"] + + +def download_file(url: str, destination: Path) -> None: + req = urllib.request.Request(url, headers={"User-Agent": "arsv-set-symbol-extractor/1.0"}) + with urllib.request.urlopen(req, timeout=120) as response, destination.open("wb") as out: + shutil.copyfileobj(response, out) + + +def run_text(command: list[str], cwd: Path | None = None, check: bool = True) -> str: + proc = subprocess.run(command, cwd=cwd, text=True, capture_output=True) + if check and proc.returncode != 0: + joined = " ".join(command) + raise RuntimeError(f"{joined} failed with {proc.returncode}: {proc.stderr.strip()}") + return proc.stdout + + +def rpm_members(rpm_path: Path) -> list[str]: + return [line for line in run_text(["bsdtar", "-tf", str(rpm_path)]).splitlines() if line] + + +def select_library_members(members: Iterable[str], max_files: int) -> list[str]: + selected = [ + member + for member in members + if not member.endswith("/") and re.search(r"(^|/)lib[^/]*\.so(?:\.|$)", member) + ] + return selected[:max_files] + + +def select_executable_members(members: Iterable[str], max_files: int) -> list[str]: + prefixes = ("./bin/", "./usr/bin/", "./sbin/", "./usr/sbin/", "./usr/lib/systemd/") + selected = [member for member in members if not member.endswith("/") and member.startswith(prefixes)] + return selected[:max_files] + + +def nm_symbols(path: Path, mode: str) -> list[str]: + flag = "-U" if mode == "defined" else "-u" + proc = subprocess.run( + ["nm", "--dynamic", "-j", flag, str(path)], + text=True, + capture_output=True, + ) + if proc.returncode != 0: + return [] + return [line.strip() for line in proc.stdout.splitlines() if line.strip()] + + +def extract_nm_report( + package: str, + pkghash: str, + branch: str, + arch: str, + mode: str, + max_files: int, + workdir: Path, +) -> NmReport: + rpm_url = package_download_url(pkghash, branch, arch) + rpm_path = workdir / Path(urllib.parse.urlparse(rpm_url).path).name + download_file(rpm_url, rpm_path) + members = rpm_members(rpm_path) + selected = ( + select_library_members(members, max_files) + if mode == "defined" + else select_executable_members(members, max_files) + ) + extract_dir = workdir / f"extract-{package.replace('/', '_').replace('+', '_')}-{mode}" + extract_dir.mkdir(parents=True, exist_ok=True) + if selected: + run_text(["bsdtar", "-xf", str(rpm_path), "-C", str(extract_dir), *selected]) + + symbols: list[str] = [] + for member in selected: + member_path = extract_dir / member + if member_path.exists(): + symbols.extend(nm_symbols(member_path, mode)) + + return NmReport( + package=package, + mode=mode, + files_seen=len(selected), + symbols=symbols, + file_examples=selected[:3], + ) + + +def char_classes(chars: Iterable[str]) -> dict[str, object]: + char_set = sorted(set(chars)) + return { + "alphabet": "".join(char_set), + "upper": sum(ch.isupper() for ch in char_set), + "lower": sum(ch.islower() for ch in char_set), + "digit": sum(ch.isdigit() for ch in char_set), + "underscore": "_" in char_set, + "at": "@" in char_set, + "dot": "." in char_set, + "dollar": "$" in char_set, + "other": "".join(ch for ch in char_set if not (ch.isalnum() or ch in "_.@$")), + } + + +def print_set_report(set_deps: list[SetDependency], prefix_len: int) -> None: + print("# set:version dependency strings") + if not set_deps: + print("No set: dependencies found.") + return + + lengths = [len(dep.payload) for dep in set_deps] + alphabet = "".join(sorted(set("".join(dep.payload for dep in set_deps)))) + print(f"count: {len(set_deps)}") + print(f"payload length min/median/max: {min(lengths)}/{statistics.median(lengths)}/{max(lengths)}") + print(f"observed encoded alphabet: {alphabet}") + print(f"bpp counts: {dict(sorted(Counter(dep.bpp for dep in set_deps).items()))}") + print(f"Mshift counts: {dict(sorted(Counter(dep.mshift for dep in set_deps).items()))}") + print() + print("package\ttype\tbpp\tMshift\tlen\tdependency\tset-prefix") + for dep in sorted(set_deps, key=lambda item: (item.package, item.dep_type, item.name)): + print( + f"{dep.package}\t{dep.dep_type}\t{dep.bpp}\t{dep.mshift}\t" + f"{len(dep.payload)}\t{dep.name}\t{dep.payload[:prefix_len]}" + ) + + +def print_nm_report(reports: list[NmReport], sample_limit: int) -> None: + print("\n# nm dynamic symbol strings") + if not reports: + print("Skipped. Pass --download-nm to download RPMs and run bsdtar/nm.") + return + + for mode in ("defined", "undefined"): + mode_reports = [report for report in reports if report.mode == mode] + if not mode_reports: + continue + all_symbols = [symbol for report in mode_reports for symbol in report.symbols] + print(f"\n## {mode} symbols") + print(f"packages: {len(mode_reports)}") + print(f"symbols total/unique: {len(all_symbols)}/{len(set(all_symbols))}") + if all_symbols: + print(f"max symbol length: {max(len(symbol) for symbol in all_symbols)}") + print(f"character classes: {char_classes(''.join(all_symbols))}") + odd = sorted({symbol for symbol in all_symbols if not IDENTISH_SYMBOL_RE.match(symbol)}) + mangled = sorted({symbol for symbol in all_symbols if symbol.startswith("_Z")}) + print(f"non identifier-ish examples: {odd[:sample_limit]}") + print(f"C++ mangled examples: {mangled[:sample_limit]}") + print("package\tfiles\tsymbols\tfile-examples\tsymbol-examples") + for report in mode_reports: + print( + f"{report.package}\t{report.files_seen}\t{len(report.symbols)}\t" + f"{', '.join(report.file_examples)}\t{', '.join(report.symbols[:sample_limit])}" + ) + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Print ALT package set:version strings and optional nm symbol statistics." + ) + parser.add_argument("packages", nargs="*", default=DEFAULT_PACKAGES, help="binary package names") + parser.add_argument("--branch", default="sisyphus", help="ALT repository branch/packageset") + parser.add_argument("--arch", default="x86_64", help="binary package architecture") + parser.add_argument("--download-nm", action="store_true", help="download RPMs and run nm on ELF files") + parser.add_argument("--nm-defined", action="store_true", help="with --download-nm, inspect defined symbols from libraries") + parser.add_argument("--nm-undefined", action="store_true", help="with --download-nm, inspect undefined symbols from executables") + parser.add_argument("--max-libs", type=int, default=8, help="max library files per package for defined-symbol nm") + parser.add_argument("--max-bins", type=int, default=20, help="max executable files per package for undefined-symbol nm") + parser.add_argument("--prefix-len", type=int, default=48, help="number of set payload chars to print per row") + parser.add_argument("--sample-limit", type=int, default=12, help="number of symbol examples to print") + return parser.parse_args(argv) + + +def main(argv: list[str]) -> int: + args = parse_args(argv) + if args.download_nm and not args.nm_defined and not args.nm_undefined: + args.nm_defined = True + args.nm_undefined = True + + if args.download_nm: + missing = [cmd for cmd in ("bsdtar", "nm") if shutil.which(cmd) is None] + if missing: + print(f"missing required command(s) for --download-nm: {', '.join(missing)}", file=sys.stderr) + return 2 + + package_hashes: dict[str, str] = {} + set_deps: list[SetDependency] = [] + errors: list[str] = [] + + for package in args.packages: + try: + pkghash, deps = get_set_dependencies(package, args.branch, args.arch) + package_hashes[package] = pkghash + set_deps.extend(deps) + except (urllib.error.URLError, urllib.error.HTTPError, KeyError, RuntimeError) as exc: + errors.append(f"{package}: {exc}") + + print(f"branch: {args.branch}") + print(f"arch: {args.arch}") + print(f"packages requested: {', '.join(args.packages)}") + if errors: + print("\n# lookup errors", file=sys.stderr) + for error in errors: + print(error, file=sys.stderr) + print() + print_set_report(set_deps, args.prefix_len) + + nm_reports: list[NmReport] = [] + if args.download_nm: + with tempfile.TemporaryDirectory(prefix="arsv-alt-rpms-") as tmp: + workdir = Path(tmp) + for package, pkghash in package_hashes.items(): + if args.nm_defined: + try: + nm_reports.append( + extract_nm_report( + package, + pkghash, + args.branch, + args.arch, + "defined", + args.max_libs, + workdir, + ) + ) + except Exception as exc: # keep processing other packages + errors.append(f"{package} defined nm: {exc}") + if args.nm_undefined: + try: + nm_reports.append( + extract_nm_report( + package, + pkghash, + args.branch, + args.arch, + "undefined", + args.max_bins, + workdir, + ) + ) + except Exception as exc: # keep processing other packages + errors.append(f"{package} undefined nm: {exc}") + + print_nm_report(nm_reports, args.sample_limit) + + if errors: + print("\n# errors", file=sys.stderr) + for error in errors: + print(error, file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/reimplement/set.py b/reimplement/set.py index fdbfe36..8a84dd3 100644 --- a/reimplement/set.py +++ b/reimplement/set.py @@ -355,11 +355,18 @@ def set_fini(set: Set, bpp: int) -> str | None: set.symbols.sort(key=lambda x: x[1]) # Sort by hash value # warn on hash collisions + j = 0 for i in range(1, set.cnt): if set.symbols[i][1] == set.symbols[i - 1][1]: print( f"Warning: Hash collision detected for symbols '{set.symbols[i][0]}' and '{set.symbols[i - 1][0]}'" ) + else: + j += 1 + set.symbols[j] = set.symbols[i] + + set.symbols = set.symbols[: j + 1] + set.cnt = j + 1 hash_values = [label_hash for _, label_hash in set.symbols] diff --git a/scripts/check_alt_set_impl.py b/scripts/check_alt_set_impl.py new file mode 100755 index 0000000..8f2b91a --- /dev/null +++ b/scripts/check_alt_set_impl.py @@ -0,0 +1,429 @@ +#!/usr/bin/env python3 +"""Check package compatibility using this repo's Python set implementation. + +This script intentionally does *not* compare with ALT's existing set.c-produced +set strings. It uses ALT only as a source of real binary RPMs: + +* Provided labels: `nm --dynamic -j -U ` +* Required labels: `nm --dynamic -j -u ` + +Both sides are encoded with `reimplement/set.py`, then compared with that same +implementation's `rpmsetcmp()`. In other words, it checks whether the current +Python implementation is internally useful for real ALT package symbol labels. +""" + +from __future__ import annotations + +import argparse +import contextlib +import json +import re +import shutil +import subprocess +import sys +import tempfile +import urllib.parse +import urllib.request +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from reimplement import set as rpmset # noqa: E402 + +API_BASE = "https://rdb.altlinux.org/api" +DEFAULT_PROVIDERS = ["glibc-core", "zlib", "libssl3", "libcrypto3"] +DEFAULT_REQUIRERS = ["coreutils", "curl", "openssl"] + + +@dataclass(frozen=True) +class PackageRPM: + name: str + pkghash: str + rpm_path: Path + extract_dir: Path + members: list[str] + + +@dataclass(frozen=True) +class LabelSet: + role: str + package: str + member: str + labels: tuple[str, ...] + set_string: str + + @property + def label_count(self) -> int: + return len(self.labels) + + @property + def set_len(self) -> int: + return len(self.set_string) + + +@dataclass(frozen=True) +class CompatibilityResult: + provider_package: str + provider_member: str + requirer_package: str + requirer_member: str + provider_labels: int + required_labels: int + cmp_result: int + status: str + + +def api_json(path: str, params: dict[str, object] | None = None) -> dict: + url = API_BASE + path + if params: + url += "?" + urllib.parse.urlencode(params, doseq=True) + req = urllib.request.Request(url, headers={"User-Agent": "arsv-alt-set-compat/1.0"}) + with urllib.request.urlopen(req, timeout=60) as response: + return json.load(response) + + +def get_pkghash(package: str, branch: str, arch: str) -> str: + data = api_json( + "/site/pkghash_by_binary_name", + {"branch": branch, "name": package, "arch": arch}, + ) + return str(data["pkghash"]) + + +def package_download_url(pkghash: str, branch: str, arch: str) -> str: + data = api_json( + f"/site/package_downloads_bin/{pkghash}", + {"branch": branch, "arch": arch}, + ) + downloads = data.get("downloads") or [] + if not downloads or not downloads[0].get("packages"): + raise RuntimeError(f"no download URL for pkghash={pkghash}") + return downloads[0]["packages"][0]["url"] + + +def run_text(command: list[str], cwd: Path | None = None, check: bool = True) -> str: + proc = subprocess.run(command, cwd=cwd, text=True, capture_output=True) + if check and proc.returncode != 0: + raise RuntimeError(f"{' '.join(command)} failed with {proc.returncode}: {proc.stderr.strip()}") + return proc.stdout + + +def download_file(url: str, destination: Path) -> None: + req = urllib.request.Request(url, headers={"User-Agent": "arsv-alt-set-compat/1.0"}) + with urllib.request.urlopen(req, timeout=120) as response, destination.open("wb") as out: + shutil.copyfileobj(response, out) + + +def rpm_members(rpm_path: Path) -> list[str]: + return [line for line in run_text(["bsdtar", "-tf", str(rpm_path)]).splitlines() if line] + + +def is_shared_library_member(member: str) -> bool: + name = Path(member).name + return not member.endswith("/") and re.search(r"(?:^|/)lib[^/]*\.so(?:\.|$)", member) is not None and ".debug" not in name + + +def select_provider_members(members: Iterable[str]) -> list[str]: + """Files whose defined dynamic symbols are Provided labels.""" + return [member for member in members if is_shared_library_member(member)] + + +def select_requirer_members(members: Iterable[str]) -> list[str]: + """Files whose undefined dynamic symbols are Required labels.""" + executable_prefixes = ("./bin/", "./usr/bin/", "./sbin/", "./usr/sbin/", "./usr/lib/systemd/") + selected = [] + for member in members: + if member.endswith("/"): + continue + if is_shared_library_member(member) or member.startswith(executable_prefixes): + selected.append(member) + return selected + + +def extract_members(rpm_path: Path, extract_dir: Path, members: Iterable[str]) -> None: + unique_members = sorted(set(members)) + if unique_members: + run_text(["bsdtar", "-xf", str(rpm_path), "-C", str(extract_dir), *unique_members]) + + +def nm_symbols(path: Path, mode: str) -> list[str]: + command = ["nm", "--dynamic", "-j", "-U", str(path)] if mode == "provided" else ["nm", "--dynamic", "-u", str(path)] + proc = subprocess.run(command, text=True, capture_output=True) + if proc.returncode != 0: + return [] + if mode == "required": + return parse_required_nm_output(proc.stdout) + return [line.strip() for line in proc.stdout.splitlines() if line.strip()] + + +def parse_required_nm_output(output: str) -> list[str]: + """Parse `nm --dynamic -u` output, ignoring weak undefined references. + + Plain `nm -j -u` discards the symbol type, but real ALT RPMs contain weak + undefined hooks like `__gmon_start__` and `_ITM_*`. Those are optional ELF + references, not hard Required labels, so keep only strong `U` entries. + """ + symbols = [] + for line in output.splitlines(): + parts = line.split() + if len(parts) < 2: + continue + symbol_type, symbol = parts[-2], parts[-1] + if symbol_type == "U": + symbols.append(symbol) + return symbols + + +def normalize_required_symbol(symbol: str) -> str: + """Normalize `foo@VER` from undefined nm output to provider-like `foo@@VER`. + + `nm -u` prints required versioned symbols with a single `@`, while defined + default-version symbols commonly use `@@`. This normalization is for this + script's compatibility model only; it is not a set.c compatibility shim. + """ + if "@@" in symbol or "@" not in symbol: + return symbol + name, version = symbol.split("@", 1) + if not name or not version: + return symbol + return f"{name}@@{version}" + + +def labels_to_set_string(labels: Iterable[str], bpp: int) -> str | None: + item_set = rpmset.set_new() + for label in labels: + rpmset.set_add(item_set, label) + # reimplement/set.py prints collision warnings to stdout; keep this script's + # stdout machine-readable and route those warnings to stderr instead. + with contextlib.redirect_stdout(sys.stderr): + return rpmset.set_fini(item_set, bpp) + + +def generate_label_set(role: str, member: str, labels: Iterable[str], bpp: int, package: str = "") -> LabelSet: + unique_labels = tuple(sorted(set(label for label in labels if label))) + set_string = labels_to_set_string(unique_labels, bpp) + if set_string is None: + raise ValueError(f"no labels for {role} {package}:{member}") + return LabelSet(role=role, package=package, member=member, labels=unique_labels, set_string=set_string) + + +def compare_status(cmp_result: int) -> str: + # provider is first argument; compatible means provider is equal or superset. + if cmp_result in (0, 1): + return "compatible" + return "incompatible" + + +def compare_label_sets(provider: LabelSet, requirer: LabelSet) -> CompatibilityResult: + cmp_result = rpmset.rpmsetcmp(provider.set_string, requirer.set_string) + return CompatibilityResult( + provider_package=provider.package, + provider_member=provider.member, + requirer_package=requirer.package, + requirer_member=requirer.member, + provider_labels=provider.label_count, + required_labels=requirer.label_count, + cmp_result=cmp_result, + status=compare_status(cmp_result), + ) + + +def build_dependency_results(provider_sets: list[LabelSet], requirer_sets: list[LabelSet], bpp: int) -> list[CompatibilityResult]: + """Compare each library only with the symbols actually required from it. + + A package executable/library has one undefined-symbol list containing symbols + required from all of its DT_NEEDED libraries. Comparing that whole list with + one provider library gives false ``-2`` results: e.g. ``/usr/bin/curl`` needs + symbols from libc, libssl, libcrypto, zlib, etc., and no single library is + supposed to provide all of them. + + For this local set.py check, keep the symbol-level ground truth around and + split every requirer's labels by provider library: provider labels ∩ required + labels. Each non-empty subset is then encoded as the requirement for exactly + that provider and compared with ``rpmsetcmp(provider, required_subset)``. + """ + results: list[CompatibilityResult] = [] + for requirer in requirer_sets: + required_labels = set(requirer.labels) + for provider in provider_sets: + required_from_provider = sorted(required_labels.intersection(provider.labels)) + if not required_from_provider: + continue + split_requirer = generate_label_set( + "required", + requirer.member, + required_from_provider, + bpp, + package=requirer.package, + ) + results.append(compare_label_sets(provider, split_requirer)) + return results + + +def fetch_package_rpm(package: str, branch: str, arch: str, workdir: Path) -> PackageRPM: + pkghash = get_pkghash(package, branch, arch) + rpm_url = package_download_url(pkghash, branch, arch) + rpm_path = workdir / Path(urllib.parse.urlparse(rpm_url).path).name + if not rpm_path.exists(): + download_file(rpm_url, rpm_path) + members = rpm_members(rpm_path) + extract_dir = workdir / f"extract-{package.replace('/', '_').replace('+', '_')}" + extract_dir.mkdir(parents=True, exist_ok=True) + return PackageRPM(package, pkghash, rpm_path, extract_dir, members) + + +def build_provider_sets(package_rpm: PackageRPM, bpp: int, max_files: int) -> list[LabelSet]: + members = select_provider_members(package_rpm.members)[:max_files] + extract_members(package_rpm.rpm_path, package_rpm.extract_dir, members) + sets = [] + for member in members: + labels = nm_symbols(package_rpm.extract_dir / member, "provided") + if labels: + sets.append(generate_label_set("provided", member, labels, bpp, package_rpm.name)) + return sets + + +def build_requirer_sets( + package_rpm: PackageRPM, + bpp: int, + max_files: int, + normalize_versions: bool, +) -> list[LabelSet]: + members = select_requirer_members(package_rpm.members)[:max_files] + extract_members(package_rpm.rpm_path, package_rpm.extract_dir, members) + sets = [] + for member in members: + labels = nm_symbols(package_rpm.extract_dir / member, "required") + if normalize_versions: + labels = [normalize_required_symbol(label) for label in labels] + if labels: + sets.append(generate_label_set("required", member, labels, bpp, package_rpm.name)) + return sets + + +def aggregate_label_sets(role: str, package_names: list[str], label_sets: list[LabelSet], bpp: int) -> LabelSet: + labels = [label for label_set in label_sets for label in label_set.labels] + return generate_label_set(role, "+".join(package_names), labels, bpp, package="aggregate") + + +def print_label_sets(title: str, label_sets: list[LabelSet]) -> None: + print(f"\n# {title}") + print("role\tpackage\tmember\tlabels\tset_len\tset_prefix") + for label_set in label_sets: + print( + f"{label_set.role}\t{label_set.package}\t{label_set.member}\t" + f"{label_set.label_count}\t{label_set.set_len}\t{label_set.set_string[:48]}" + ) + + +def print_results(results: list[CompatibilityResult]) -> None: + print("\n# compatibility") + print("status\tcmp\tprovider_pkg\tprovider_member\tprovider_labels\trequirer_pkg\trequirer_member\trequired_labels") + for result in results: + print( + f"{result.status}\t{result.cmp_result}\t{result.provider_package}\t{result.provider_member}\t" + f"{result.provider_labels}\t{result.requirer_package}\t{result.requirer_member}\t{result.required_labels}" + ) + summary = {status: sum(1 for result in results if result.status == status) for status in sorted({r.status for r in results})} + print(f"summary: {summary}") + + +def parse_package_list(values: list[str] | None) -> list[str]: + packages = [] + for value in values or []: + packages.extend(part for part in value.split(",") if part) + return packages + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Generate Provided/Required set strings from ALT RPM labels with reimplement/set.py and compare them." + ) + parser.add_argument("packages", nargs="*", help="packages used as both providers and requirers if explicit lists are omitted") + parser.add_argument("--provider", action="append", help="provider package; can be repeated or comma-separated") + parser.add_argument("--requirer", action="append", help="requirer package; can be repeated or comma-separated") + parser.add_argument("--branch", default="sisyphus", help="ALT repository branch/packageset") + parser.add_argument("--arch", default="x86_64", help="binary package architecture") + parser.add_argument("--bpp", type=int, default=32, help="bits per hash used by local set.py") + parser.add_argument("--max-provider-files", type=int, default=64, help="max provider ELF files per package") + parser.add_argument("--max-requirer-files", type=int, default=64, help="max requirer ELF files per package") + parser.add_argument("--all-pairs", action="store_true", help="compare every provider file set with every requirer file set") + parser.add_argument( + "--no-normalize-required-version", + action="store_true", + help="keep nm -u single-@ required symbols unchanged instead of converting foo@VER to foo@@VER", + ) + parser.add_argument("--keep-workdir", action="store_true", help="keep downloaded RPMs/extracted files") + parser.add_argument("--workdir", type=Path, help="directory for downloads/extraction") + return parser.parse_args(argv) + + +def main(argv: list[str]) -> int: + args = parse_args(argv) + missing = [cmd for cmd in ("bsdtar", "nm") if shutil.which(cmd) is None] + if missing: + print(f"missing required command(s): {', '.join(missing)}", file=sys.stderr) + return 2 + + positional = args.packages or [] + providers = parse_package_list(args.provider) or positional or DEFAULT_PROVIDERS + requirers = parse_package_list(args.requirer) or positional or DEFAULT_REQUIRERS + + cleanup = False + if args.workdir: + workdir = args.workdir + workdir.mkdir(parents=True, exist_ok=True) + else: + workdir = Path(tempfile.mkdtemp(prefix="arsv-alt-set-compat-")) + cleanup = not args.keep_workdir + + try: + provider_sets: list[LabelSet] = [] + requirer_sets: list[LabelSet] = [] + for package in providers: + provider_sets.extend( + build_provider_sets(fetch_package_rpm(package, args.branch, args.arch, workdir), args.bpp, args.max_provider_files) + ) + for package in requirers: + requirer_sets.extend( + build_requirer_sets( + fetch_package_rpm(package, args.branch, args.arch, workdir), + args.bpp, + args.max_requirer_files, + normalize_versions=not args.no_normalize_required_version, + ) + ) + + print(f"branch: {args.branch}") + print(f"arch: {args.arch}") + print(f"bpp: {args.bpp}") + print(f"providers: {', '.join(providers)}") + print(f"requirers: {', '.join(requirers)}") + print(f"required symbol version normalization: {not args.no_normalize_required_version}") + print_label_sets("generated Provided sets", provider_sets) + print_label_sets("generated Required sets", requirer_sets) + + if not provider_sets or not requirer_sets: + print("\nNo comparable sets generated.", file=sys.stderr) + return 1 + + if args.all_pairs: + results = [compare_label_sets(provider, requirer) for provider in provider_sets for requirer in requirer_sets] + else: + results = build_dependency_results(provider_sets, requirer_sets, args.bpp) + print_results(results) + finally: + if cleanup: + shutil.rmtree(workdir, ignore_errors=True) + else: + print(f"workdir: {workdir}") + + return 0 if all(result.status == "compatible" for result in results) else 1 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/scripts/check_arch_set_impl.py b/scripts/check_arch_set_impl.py new file mode 100755 index 0000000..86c55b6 --- /dev/null +++ b/scripts/check_arch_set_impl.py @@ -0,0 +1,172 @@ +#!/usr/bin/env python3 +"""Check reimplement/set.py against ELF libraries installed on this Arch Linux system. + +This is the local-system analogue of check_alt_set_impl.py. It treats shared +libraries as providers (defined dynamic symbols) and executables/shared objects +as requirers (undefined dynamic symbols). Required labels are split by provider +library before comparing, so each library is checked only against symbols that +it actually exports. +""" + +from __future__ import annotations + +import argparse +import os +import shutil +import sys +from pathlib import Path + +SCRIPT_DIR = Path(__file__).resolve().parent +if str(SCRIPT_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPT_DIR)) + +import check_alt_set_impl as compat # noqa: E402 + +DEFAULT_PROVIDER_DIRS = [Path("/usr/lib")] +DEFAULT_REQUIRER_DIRS = [Path("/usr/bin"), Path("/usr/lib")] + + +def parse_path_list(values: list[str] | None, defaults: list[Path]) -> list[Path]: + paths: list[Path] = [] + for value in values or []: + paths.extend(Path(part) for part in value.split(",") if part) + return paths or defaults + + +def is_shared_library_path(path: Path) -> bool: + name = path.name + return ".debug" not in name and name.startswith("lib") and ".so" in name + + +def iter_files(paths: list[Path], recursive: bool) -> list[Path]: + files: list[Path] = [] + seen: set[Path] = set() + for root in paths: + if root.is_file(): + candidates = [root] + elif recursive: + candidates = (path for path in root.rglob("*") if path.is_file()) + else: + candidates = (path for path in root.iterdir() if path.is_file()) if root.is_dir() else [] + for path in candidates: + try: + resolved = path.resolve() + except OSError: + continue + if resolved in seen: + continue + seen.add(resolved) + files.append(path) + return sorted(files) + + +def executable_or_library(path: Path) -> bool: + return is_shared_library_path(path) or os.access(path, os.X_OK) + + +def build_local_provider_sets(paths: list[Path], bpp: int, recursive: bool, max_files: int) -> list[compat.LabelSet]: + provider_files = [path for path in iter_files(paths, recursive) if is_shared_library_path(path)][:max_files] + sets: list[compat.LabelSet] = [] + for path in provider_files: + labels = compat.nm_symbols(path, "provided") + if labels: + sets.append(compat.generate_label_set("provided", str(path), labels, bpp, package="arch-local")) + return sets + + +def build_local_requirer_sets( + paths: list[Path], + bpp: int, + recursive: bool, + max_files: int, + normalize_versions: bool, +) -> list[compat.LabelSet]: + requirer_files = [path for path in iter_files(paths, recursive) if executable_or_library(path)][:max_files] + sets: list[compat.LabelSet] = [] + for path in requirer_files: + labels = compat.nm_symbols(path, "required") + if normalize_versions: + labels = [compat.normalize_required_symbol(label) for label in labels] + if labels: + sets.append(compat.generate_label_set("required", str(path), labels, bpp, package="arch-local")) + return sets + + +def print_unmatched_required_labels(provider_sets: list[compat.LabelSet], requirer_sets: list[compat.LabelSet], limit: int) -> None: + provided = {label for provider in provider_sets for label in provider.labels} + rows: list[tuple[str, int, list[str]]] = [] + for requirer in requirer_sets: + missing = sorted(set(requirer.labels).difference(provided)) + if missing: + rows.append((requirer.member, len(missing), missing[:limit])) + + print("\n# required labels not exported by scanned provider libraries") + print("requirer\tmissing_labels\texamples") + for member, count, examples in rows[:limit]: + print(f"{member}\t{count}\t{', '.join(examples)}") + if len(rows) > limit: + print(f"... {len(rows) - limit} more requirer files with unmatched labels") + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Generate Provided/Required set strings from local Arch Linux ELF files and compare them with reimplement/set.py." + ) + parser.add_argument("--provider-dir", action="append", help="directory/file containing provider libraries; repeat or comma-separate") + parser.add_argument("--requirer-dir", action="append", help="directory/file containing requirer ELF files; repeat or comma-separate") + parser.add_argument("--bpp", type=int, default=32, help="bits per hash used by local set.py") + parser.add_argument("--max-provider-files", type=int, default=256, help="max provider libraries to inspect") + parser.add_argument("--max-requirer-files", type=int, default=256, help="max requirer files to inspect") + parser.add_argument("--recursive", action="store_true", help="scan directories recursively") + parser.add_argument("--all-pairs", action="store_true", help="compare every provider set with every full requirer set") + parser.add_argument( + "--no-normalize-required-version", + action="store_true", + help="keep nm -u single-@ required symbols unchanged instead of converting foo@VER to foo@@VER", + ) + parser.add_argument("--unmatched-limit", type=int, default=20, help="max unmatched-label rows/examples to print") + return parser.parse_args(argv) + + +def main(argv: list[str]) -> int: + args = parse_args(argv) + if shutil.which("nm") is None: + print("missing required command: nm", file=sys.stderr) + return 2 + + provider_paths = parse_path_list(args.provider_dir, DEFAULT_PROVIDER_DIRS) + requirer_paths = parse_path_list(args.requirer_dir, DEFAULT_REQUIRER_DIRS) + + provider_sets = build_local_provider_sets(provider_paths, args.bpp, args.recursive, args.max_provider_files) + requirer_sets = build_local_requirer_sets( + requirer_paths, + args.bpp, + args.recursive, + args.max_requirer_files, + normalize_versions=not args.no_normalize_required_version, + ) + + print("system: arch-local") + print(f"bpp: {args.bpp}") + print(f"provider paths: {', '.join(str(path) for path in provider_paths)}") + print(f"requirer paths: {', '.join(str(path) for path in requirer_paths)}") + print(f"required symbol version normalization: {not args.no_normalize_required_version}") + compat.print_label_sets("generated Provided sets", provider_sets) + compat.print_label_sets("generated Required sets", requirer_sets) + + if not provider_sets or not requirer_sets: + print("\nNo comparable sets generated.", file=sys.stderr) + return 1 + + if args.all_pairs: + results = [compat.compare_label_sets(provider, requirer) for provider in provider_sets for requirer in requirer_sets] + else: + results = compat.build_dependency_results(provider_sets, requirer_sets, args.bpp) + compat.print_results(results) + print_unmatched_required_labels(provider_sets, requirer_sets, args.unmatched_limit) + + return 0 if results and all(result.status == "compatible" for result in results) else 1 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/scripts/compare_binary_with_lib_sets.py b/scripts/compare_binary_with_lib_sets.py new file mode 100755 index 0000000..181aecf --- /dev/null +++ b/scripts/compare_binary_with_lib_sets.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +"""Compare one binary's required set string with a few provider libraries. + +Usage: + scripts/compare_binary_with_lib_sets.py [--bpp N] BINARY LIB.so [LIB.so ...] + +For every library, this script compares: + set(defined dynamic symbols from LIB) vs + set(required dynamic symbols from BINARY that LIB provides) +""" + +from __future__ import annotations + +import argparse +import contextlib +import shutil +import subprocess +import sys +from pathlib import Path +from typing import Iterable + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from reimplement import set as rpmset # noqa: E402 + + +def run_nm(command: list[str]) -> str: + proc = subprocess.run(command, text=True, capture_output=True) + if proc.returncode != 0: + raise RuntimeError(f"{' '.join(command)} failed with {proc.returncode}: {proc.stderr.strip()}") + return proc.stdout + + +def normalize_required_symbol(symbol: str) -> str: + """Normalize nm's required `foo@VER` form to provider-like `foo@@VER`.""" + if "@@" in symbol or "@" not in symbol: + return symbol + name, version = symbol.split("@", 1) + if not name or not version: + return symbol + return f"{name}@@{version}" + + +def required_symbols(path: Path, normalize_versions: bool) -> set[str]: + """Return strong undefined dynamic symbols required by one ELF file.""" + output = run_nm(["nm", "--dynamic", "-u", str(path)]) + symbols: set[str] = set() + for line in output.splitlines(): + parts = line.split() + if len(parts) < 2 or parts[-2] != "U": + continue + symbol = parts[-1] + symbols.add(normalize_required_symbol(symbol) if normalize_versions else symbol) + return symbols + + +def provided_symbols(path: Path) -> set[str]: + """Return defined dynamic symbols provided by one ELF shared library.""" + output = run_nm(["nm", "--dynamic", "-j", "-U", str(path)]) + return {line.strip() for line in output.splitlines() if line.strip()} + + +def labels_to_set_string(labels: Iterable[str], bpp: int) -> str | None: + item_set = rpmset.set_new() + for label in sorted(set(labels)): + rpmset.set_add(item_set, label) + # set.py may print hash-collision warnings; keep stdout tabular. + with contextlib.redirect_stdout(sys.stderr): + return rpmset.set_fini(item_set, bpp) + + +def compare_library(binary_required: set[str], library: Path, bpp: int) -> tuple[str, str, int, int, str, str]: + provided = provided_symbols(library) + required_from_library = binary_required.intersection(provided) + provider_set = labels_to_set_string(provided, bpp) + required_set = labels_to_set_string(required_from_library, bpp) + + if not provided: + return "no-provided-symbols", "", 0, 0, "-", "-" + if not required_from_library: + return "not-required", "", len(provided), 0, provider_set or "-", "-" + + assert provider_set is not None + assert required_set is not None + cmp_result = rpmset.rpmsetcmp(provider_set, required_set) + status = "compatible" if cmp_result in (0, 1) else "incompatible" + return status, str(cmp_result), len(provided), len(required_from_library), provider_set, required_set + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Compare one binary against a few libraries using local set.py set strings.") + parser.add_argument("binary", type=Path, help="ELF executable/shared object with required dynamic symbols") + parser.add_argument("libraries", nargs="+", type=Path, help="provider shared libraries to compare against") + parser.add_argument("--bpp", type=int, default=32, help="bits per hash used by local set.py") + parser.add_argument( + "--no-normalize-required-version", + action="store_true", + help="keep nm -u single-@ required symbols unchanged instead of converting foo@VER to foo@@VER", + ) + return parser.parse_args(argv) + + +def main(argv: list[str]) -> int: + args = parse_args(argv) + if shutil.which("nm") is None: + print("missing required command: nm", file=sys.stderr) + return 2 + + required = required_symbols(args.binary, normalize_versions=not args.no_normalize_required_version) + if not required: + print(f"no strong dynamic required symbols found in {args.binary}", file=sys.stderr) + return 1 + + print(f"binary\t{args.binary}") + print(f"bpp\t{args.bpp}") + print(f"required_symbols\t{len(required)}") + print("status\tcmp\tlib\tprovided\trequired_from_lib\tprovider_set\trequired_set") + + failed = False + for library in args.libraries: + try: + status, cmp_result, provided_count, required_count, provider_set, required_set = compare_library( + required, library, args.bpp + ) + except RuntimeError as exc: + print(f"error\t\t{library}\t0\t0\t-\t-", flush=True) + print(exc, file=sys.stderr) + failed = True + continue + + print( + f"{status}\t{cmp_result}\t{library}\t{provided_count}\t{required_count}\t{provider_set}\t{required_set}" + ) + failed = failed or status in {"incompatible", "no-provided-symbols"} + + return 1 if failed else 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/tests/test_check_alt_set_impl.py b/tests/test_check_alt_set_impl.py new file mode 100644 index 0000000..9a0be6e --- /dev/null +++ b/tests/test_check_alt_set_impl.py @@ -0,0 +1,100 @@ +import importlib.util +import sys +from pathlib import Path + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "check_alt_set_impl.py" + + +def load_script(): + spec = importlib.util.spec_from_file_location("check_alt_set_impl", SCRIPT) + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def test_select_elf_members_separates_providers_and_requirers(): + check = load_script() + members = [ + "./usr/lib64/libfoo.so.1", + "./usr/lib64/libfoo.so.1.2.3", + "./usr/lib64/libfoo.a", + "./usr/bin/tool", + "./usr/share/doc/readme", + ] + + assert check.select_provider_members(members) == [ + "./usr/lib64/libfoo.so.1", + "./usr/lib64/libfoo.so.1.2.3", + ] + assert check.select_requirer_members(members) == [ + "./usr/lib64/libfoo.so.1", + "./usr/lib64/libfoo.so.1.2.3", + "./usr/bin/tool", + ] + + +def test_generate_label_set_and_compare_uses_only_local_set_py(): + check = load_script() + provided = check.generate_label_set("provider", "libfoo.so.1", ["foo", "bar", "baz"], bpp=16) + required = check.generate_label_set("requirer", "tool", ["foo", "bar"], bpp=16) + missing = check.generate_label_set("requirer", "badtool", ["foo", "quux"], bpp=16) + + ok = check.compare_label_sets(provided, required) + bad = check.compare_label_sets(provided, missing) + + assert ok.status == "compatible" + assert ok.cmp_result == 1 + assert bad.status == "incompatible" + assert bad.cmp_result == -2 + + +def test_normalize_required_symbol_version_can_match_provided_symbol_version(): + check = load_script() + + assert check.normalize_required_symbol("foo@LIB_1") == "foo@@LIB_1" + assert check.normalize_required_symbol("foo@@LIB_1") == "foo@@LIB_1" + assert check.normalize_required_symbol("foo") == "foo" + + +def test_parse_required_nm_output_filters_weak_undefined_symbols(): + check = load_script() + nm_output = """ + w __gmon_start__ + w _ITM_deregisterTMCloneTable + U close@GLIBC_2.2.5 + U memcpy@GLIBC_2.14 + W optional_hook + """ + + assert check.parse_required_nm_output(nm_output) == [ + "close@GLIBC_2.2.5", + "memcpy@GLIBC_2.14", + ] + + +def test_default_bpp_is_large_enough_for_real_alt_symbol_sets(): + check = load_script() + + assert check.parse_args([]).bpp == 32 + + +def test_build_dependency_results_splits_required_labels_by_provider_library(): + check = load_script() + libc = check.generate_label_set( + "provided", "libc.so.6", ["close@@GLIBC_2.2.5", "read@@GLIBC_2.2.5"], bpp=16, package="glibc-core" + ) + libz = check.generate_label_set( + "provided", "libz.so.1", ["inflate", "deflate"], bpp=16, package="zlib" + ) + tool = check.generate_label_set( + "required", "tool", ["close@@GLIBC_2.2.5", "inflate"], bpp=16, package="consumer" + ) + + results = check.build_dependency_results([libc, libz], [tool], bpp=16) + + assert [(result.provider_member, result.required_labels, result.status) for result in results] == [ + ("libc.so.6", 1, "compatible"), + ("libz.so.1", 1, "compatible"), + ] diff --git a/tests/test_reimplement_set.py b/tests/test_reimplement_set.py index 5f60867..14686c7 100644 --- a/tests/test_reimplement_set.py +++ b/tests/test_reimplement_set.py @@ -52,11 +52,13 @@ class SetStringTest(unittest.TestCase): encoded = rpmset.set_fini(item_set, bpp=8) self.assertIsNotNone(encoded) self.assertEqual(item_set.cnt, 2) - self.assertEqual(item_set.labels, sorted(item_set.labels, key=lambda item: item[1])) + self.assertEqual( + item_set.symbols, sorted(item_set.symbols, key=lambda item: item[1]) + ) rpmset.set_free(item_set) self.assertEqual(item_set.cnt, 0) - self.assertEqual(item_set.labels, []) + self.assertEqual(item_set.symbols, []) def test_hash_is_stable_64_bit_ascii_integer(self): self.assertEqual(rpmset.hash("ascii_symbol"), 10827468943333989194) @@ -69,7 +71,9 @@ class SetStringTest(unittest.TestCase): class DownsampleSetTest(unittest.TestCase): def test_masks_high_half_and_keeps_sorted_unique_values(self): - self.assertEqual(rpmset.downsample_set([1, 3, 6, 8, 10, 14], 3), [0, 1, 2, 3, 6]) + self.assertEqual( + rpmset.downsample_set([1, 3, 6, 8, 10, 14], 3), [0, 1, 2, 3, 6] + ) def test_removes_duplicates_created_by_masking(self): self.assertEqual(rpmset.downsample_set([1, 6, 14], 3), [1, 6])