plasma-next/tools/lib/audit.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Audit accepted .debs for runtime ELF libraries accidentally shipped in -dev."""

import hashlib
import json
import os
import re
import shutil
import subprocess
import tarfile
import tempfile
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path

from project import ROOT as PROJECT
from project import WORK


def main():
    artifacts = {}
    records = {}
    states = list((WORK / "state").glob("*.json"))
    if not states:
        raise ValueError("No accepted builds. Run build first.")
    for path in states:
        record = json.loads(path.read_text())
        records[path.name] = hashlib.sha256(path.read_bytes()).hexdigest()
        for name, digest in record["artifacts"].items():
            if name.endswith(".deb") and "-dbgsym_" not in name:
                assert artifacts.get(name, digest) == digest
                artifacts[name] = digest

    elfs = []
    dev_symlinks = 0
    for index, (name, expected) in enumerate(sorted(artifacts.items()), 1):
        archive = WORK / "results" / name
        with archive.open("rb") as stream:
            assert hashlib.file_digest(stream, "sha256").hexdigest() == expected, name
        package = name.split("_")[0]
        development = package.endswith("-dev")
        with tempfile.TemporaryDirectory(prefix="package-elf-", dir=WORK) as temporary:
            paths = {}
            process = subprocess.Popen(
                ["dpkg-deb", "--fsys-tarfile", str(archive)], stdout=subprocess.PIPE
            )
            try:
                with tarfile.open(fileobj=process.stdout, mode="r|") as tar:
                    for member in tar:
                        path = member.name.removeprefix("./")
                        if development and member.issym() and path.endswith(".so"):
                            dev_symlinks += 1
                        if not member.isfile():
                            continue
                        stream = tar.extractfile(member)
                        head = stream.read(4)
                        if head != b"\x7fELF":
                            continue
                        local = Path(temporary) / str(len(paths))
                        with local.open("wb") as output:
                            output.write(head)
                            shutil.copyfileobj(stream, output)
                        paths[str(local)] = {
                            "package": package,
                            "path": path,
                            "development": development,
                        }
                assert process.wait() == 0, name
            finally:
                process.stdout.close()
                if process.poll() is None:
                    process.terminate()
                    process.wait()
            filenames = list(paths)
            for offset in range(0, len(filenames), 100):
                batch = filenames[offset : offset + 100]
                text = subprocess.check_output(
                    ["readelf", "--wide", "--dynamic", *batch],
                    text=True,
                    env={**os.environ, "LC_ALL": "C"},
                )
                sections = re.split(r"(?m)^File: (.+)\n", text)
                entries = (
                    [(batch[0], text)]
                    if len(batch) == 1
                    else list(zip(sections[1::2], sections[2::2]))
                )
                assert len(entries) == len(batch) and {
                    filename for filename, _ in entries
                } == set(batch), name
                for filename, section in entries:
                    entry = paths[filename]
                    entry["needed"] = re.findall(r"\(NEEDED\).*?\[(.*?)\]", section)
                    entry["soname"] = re.findall(r"\(SONAME\).*?\[(.*?)\]", section)
                    elfs.append(entry)
        if index % 50 == 0:
            print(f"Checked {index}/{len(artifacts)} packages", flush=True)

    providers = defaultdict(list)
    candidates = []
    for entry in elfs:
        path = Path(entry["path"])
        library = re.search(r"\.so(?:\..+)?$", path.name) and "lib" in path.parts
        if not library:
            continue
        for key in set(entry["soname"] + [path.name]):
            providers[key].append(entry)
        if entry["development"]:
            candidates.append(entry)
    violations = []
    for entry in elfs:
        if entry["development"] or entry["package"] == "plasma-next-rust-build-tools":
            continue
        for needed in entry["needed"]:
            matches = providers.get(needed, [])
            if matches and all(provider["development"] for provider in matches):
                violations.append(
                    {
                        "consumer_package": entry["package"],
                        "consumer": entry["path"],
                        "needed": needed,
                        "providers": matches,
                    }
                )
    report = {
        "checked_at": datetime.now(timezone.utc).isoformat(),
        "status": "findings" if candidates or violations else "passed",
        "packages_checked": len(artifacts),
        "elf_files_checked": len(elfs),
        "development_so_symlinks": dev_symlinks,
        "records": records,
        "verified_artifacts": artifacts,
        "real_libraries_in_dev": candidates,
        "runtime_needs_dev_library": violations,
        "limits": "Static DT_NEEDED and regular ELF library placement. Does not verify all dlopen/QML/script dependencies or loader search paths.",
    }
    target = WORK / "records/package-library-split-audit.json"
    target.write_text(json.dumps(report, indent=2) + "\n")
    print(
        f"{len(artifacts)} packages, {len(elfs)} ELF files, {dev_symlinks} development .so symlinks."
    )
    for entry in candidates:
        print(f"REVIEW {entry['package']}: {entry['path']}")
    print(
        f"{len(violations)} runtime DT_NEEDED references resolve only to -dev. Report: {target.relative_to(PROJECT)}"
    )
    return 1 if candidates or violations else 0


if __name__ == "__main__":
    raise SystemExit(main())