reproduce.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import time
from pathlib import Path

from google.api import metric_pb2


ROOT = Path(__file__).resolve().parent
ARTIFACTS = ROOT / "artifacts"


def vmhwm_kb() -> int | None:
    with open("/proc/self/status", "r", encoding="utf-8", errors="ignore") as fh:
        for line in fh:
            if line.startswith("VmHWM:"):
                return int(line.split()[1])
    return None


def parse_one(path: Path) -> dict:
    start = time.time()
    before = vmhwm_kb()
    msg = metric_pb2.MetricDescriptor()
    msg.ParseFromString(path.read_bytes())
    after = vmhwm_kb()
    return {
        "path": str(path),
        "elapsed": time.time() - start,
        "vmhwm_before_kb": before,
        "vmhwm_after_kb": after,
        "monitored_resource_types_len": len(msg.monitored_resource_types),
        "display_name_len": len(msg.display_name),
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--control", type=Path, default=ARTIFACTS / "control_one_display_name.pb")
    parser.add_argument("--malicious", type=Path, default=ARTIFACTS / "malicious_monitored_resource_types_6000000.pb")
    parser.add_argument("--json-out", type=Path)
    args = parser.parse_args()

    control = parse_one(args.control)
    malicious = parse_one(args.malicious)
    summary = {
        "format": "Protocol Buffers",
        "protobuf_version": __import__("google.protobuf").protobuf.__version__,
        "entrypoint": "metric_pb2.MetricDescriptor().ParseFromString(...)",
        "control": control,
        "malicious": malicious,
        "delta": {
            "elapsed_ratio": malicious["elapsed"] / max(control["elapsed"], 1e-9),
            "vmhwm_delta_kb": malicious["vmhwm_after_kb"] - control["vmhwm_after_kb"],
            "monitored_resource_types_added": malicious["monitored_resource_types_len"] - control["monitored_resource_types_len"],
            "repro_ok": (
                control["monitored_resource_types_len"] == 0
                and malicious["monitored_resource_types_len"] == 6_000_000
                and malicious["vmhwm_after_kb"] - control["vmhwm_after_kb"] >= 150_000
            ),
        },
    }
    if args.json_out:
        args.json_out.write_text(json.dumps(summary, indent=2) + "\n")
    print(json.dumps(summary, indent=2))
    return 0


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