Spaces:
Running
Running
| """Create a small, local-only support bundle without copying private evidence.""" | |
| from __future__ import annotations | |
| import datetime as dt | |
| import json | |
| import math | |
| import os | |
| import platform | |
| import re | |
| import signal | |
| import stat | |
| import subprocess | |
| import sys | |
| from collections.abc import Callable | |
| from pathlib import Path | |
| from typing import Any | |
| from . import __version__ | |
| from .client import request | |
| from .jsonio import JsonInputError, write_new_json | |
| Query = Callable[..., dict[str, Any]] | |
| MAX_DETECTIONS = 100 | |
| TOKEN = re.compile(r"[A-Za-z0-9_.:-]{1,128}\Z") | |
| SUPPORT_BUNDLE_SCHEMA = "lumi-eggcracker.support-bundle.v1" | |
| MAX_VALIDATION_BYTES = 1024 * 1024 | |
| MAX_JSON_DEPTH = 32 | |
| MAX_AGGREGATE_COUNT = 1_000_000_000 | |
| MAX_STATE_KEYS = 4096 | |
| def _token(value: object) -> str | None: | |
| return value if isinstance(value, str) and TOKEN.fullmatch(value) else None | |
| def _systemd_version() -> str | None: | |
| try: | |
| result = subprocess.run( | |
| ["/usr/bin/systemd-run", "--version"], | |
| capture_output=True, | |
| text=True, | |
| check=False, | |
| timeout=5, | |
| ) | |
| except (OSError, subprocess.TimeoutExpired): | |
| return None | |
| line = (result.stdout or "").splitlines() | |
| return line[0][:80] if result.returncode == 0 and line else None | |
| def _host() -> dict[str, Any]: | |
| controllers: list[str] = [] | |
| try: | |
| controllers = sorted( | |
| value | |
| for value in Path("/sys/fs/cgroup/cgroup.controllers") | |
| .read_text(encoding="ascii") | |
| .split() | |
| if value in {"cpu", "memory", "pids"} | |
| ) | |
| except (OSError, UnicodeDecodeError): | |
| pass | |
| return { | |
| "platform": platform.system(), | |
| "kernel": platform.release(), | |
| "machine": platform.machine(), | |
| "python": ".".join(str(value) for value in sys.version_info[:3]), | |
| "cgroup_v2": Path("/sys/fs/cgroup/cgroup.controllers").is_file(), | |
| "controllers": controllers, | |
| "pidfd_open": hasattr(os, "pidfd_open"), | |
| "pidfd_send_signal": hasattr(signal, "pidfd_send_signal"), | |
| "systemd": _systemd_version(), | |
| } | |
| def _doctor(value: dict[str, Any]) -> dict[str, Any]: | |
| discovery = value.get("discovery") if isinstance(value.get("discovery"), dict) else {} | |
| network = value.get("network") if isinstance(value.get("network"), dict) else {} | |
| primitives = network.get("primitives") if isinstance(network.get("primitives"), dict) else {} | |
| return { | |
| "result": _token(value.get("result")), | |
| "backend": _token(value.get("backend")), | |
| "version": _token(value.get("version")), | |
| "workload_uid": value.get("workload_uid"), | |
| "autonomous_discovery": value.get("autonomous_discovery"), | |
| "cgroup_v2": value.get("cgroup_v2"), | |
| "pidfd": value.get("pidfd"), | |
| "execution_boundary": value.get("execution_boundary"), | |
| "installation": value.get("installation"), | |
| "incidents": value.get("incidents"), | |
| "network": { | |
| "mode": _token(network.get("mode")), | |
| "cleanup_healthy": network.get("cleanup_healthy"), | |
| "primitives_supported": primitives.get("supported"), | |
| }, | |
| "discovery": { | |
| "healthy": discovery.get("healthy"), | |
| "consecutive_failures": discovery.get("consecutive_failures"), | |
| "last_scan_duration_ms": discovery.get("last_scan_duration_ms"), | |
| "last_scan_completed": discovery.get("last_scan_completed"), | |
| "receipt_persistence_healthy": discovery.get("receipt_persistence_healthy"), | |
| }, | |
| } | |
| def _detection(value: dict[str, Any]) -> dict[str, Any]: | |
| detector = value.get("detector") if isinstance(value.get("detector"), dict) else {} | |
| trigger_value = value.get("trigger") | |
| trigger = trigger_value.get("kind") if isinstance(trigger_value, dict) else trigger_value | |
| result = { | |
| "event_id": _token(value.get("event_id")), | |
| "result": _token(value.get("result")), | |
| "trigger": _token(trigger), | |
| "version": _token(value.get("version")), | |
| "profile": _token(detector.get("profile")), | |
| } | |
| boundary = value.get("boundary") | |
| if isinstance(boundary, dict): | |
| policy_sha256 = boundary.get("policy_sha256") | |
| result["boundary"] = { | |
| "address_family": _token(boundary.get("address_family")), | |
| "mode": _token(boundary.get("mode")), | |
| "policy_sha256": ( | |
| policy_sha256 | |
| if isinstance(policy_sha256, str) and re.fullmatch(r"[0-9a-f]{64}", policy_sha256) | |
| else None | |
| ), | |
| "violation": _token(boundary.get("violation")), | |
| } | |
| return result | |
| def _workload_health(value: dict[str, Any]) -> dict[str, Any]: | |
| runs = value.get("runs") if isinstance(value.get("runs"), list) else [] | |
| states: dict[str, int] = {} | |
| for item in runs: | |
| if not isinstance(item, dict): | |
| continue | |
| state = _token(item.get("state")) | |
| if state is not None: | |
| states[state] = states.get(state, 0) + 1 | |
| return {"run_count": len(runs), "states": states} | |
| def _incident_health(value: dict[str, Any]) -> dict[str, Any]: | |
| incidents = value.get("incidents") if isinstance(value.get("incidents"), list) else [] | |
| active = 0 | |
| states: dict[str, int] = {} | |
| for item in incidents: | |
| if not isinstance(item, dict): | |
| continue | |
| state = _token(item.get("state")) | |
| if state is not None: | |
| states[state] = states.get(state, 0) + 1 | |
| if state in {"ACTIVE", "ACKNOWLEDGED"}: | |
| active += 1 | |
| return {"count": len(incidents), "active": active, "states": states} | |
| def collect(query: Query = request) -> dict[str, Any]: | |
| """Collect only bounded fields from public read-only supervisor queries.""" | |
| doctor = query("doctor") | |
| detections_raw = query("detections") | |
| list_raw = query("list") | |
| incidents_raw = query("incidents") | |
| detections = detections_raw.get("detections") | |
| if not isinstance(detections, list): | |
| detections = [] | |
| return { | |
| "schema_version": "lumi-eggcracker.support-bundle.v1", | |
| "generated_utc": dt.datetime.now(dt.UTC).isoformat().replace("+00:00", "Z"), | |
| "product": "Lumi Eggcracker", | |
| "version": _token(doctor.get("version", __version__)) or __version__, | |
| "host": _host(), | |
| "health": _doctor(doctor), | |
| "workloads": _workload_health(list_raw), | |
| "incidents": _incident_health(incidents_raw), | |
| "receipts": [_detection(item) for item in detections[:MAX_DETECTIONS] if isinstance(item, dict)], | |
| "privacy": { | |
| "network": "aggregate-boundary-events-only", | |
| "raw_receipts": False, | |
| "argv": False, | |
| "paths": False, | |
| "pids": False, | |
| "model_data": False, | |
| }, | |
| } | |
| def write_bundle(destination: Path, query: Query = request) -> dict[str, Any]: | |
| if destination.exists() or destination.is_symlink() or not destination.parent.is_dir(): | |
| raise JsonInputError("support bundle output must be a new file under an existing directory") | |
| value = collect(query) | |
| write_new_json(destination, value) | |
| return value | |
| def main(destination: Path) -> int: | |
| try: | |
| write_bundle(destination) | |
| except (JsonInputError, OSError) as error: | |
| print(f"eggcracker support-bundle: {error}", file=sys.stderr) | |
| return 4 | |
| print(json.dumps({"result": "WRITTEN", "path": str(destination)}, sort_keys=True)) | |
| return 0 | |
| class _InvalidSupportBundle(ValueError): | |
| """Internal marker for a support bundle outside the bounded v1 contract.""" | |
| def _unique_json_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: | |
| result: dict[str, Any] = {} | |
| for key, value in pairs: | |
| if key in result: | |
| raise _InvalidSupportBundle | |
| result[key] = value | |
| return result | |
| def _reject_json_constant(_value: str) -> None: | |
| raise _InvalidSupportBundle | |
| def _finite_json_float(value: str) -> float: | |
| result = float(value) | |
| if not math.isfinite(result): | |
| raise _InvalidSupportBundle | |
| return result | |
| def _check_json_depth(value: str) -> None: | |
| depth = 0 | |
| quoted = False | |
| escaped = False | |
| for character in value: | |
| if quoted: | |
| if escaped: | |
| escaped = False | |
| elif character == "\\": | |
| escaped = True | |
| elif character == '"': | |
| quoted = False | |
| continue | |
| if character == '"': | |
| quoted = True | |
| elif character in "[{": | |
| depth += 1 | |
| if depth > MAX_JSON_DEPTH: | |
| raise _InvalidSupportBundle | |
| elif character in "]}": | |
| depth -= 1 | |
| if depth < 0: | |
| raise _InvalidSupportBundle | |
| def _read_bounded_regular_file(path: Path) -> bytes: | |
| descriptor: int | None = None | |
| try: | |
| before = path.lstat() | |
| if not stat.S_ISREG(before.st_mode) or before.st_size > MAX_VALIDATION_BYTES: | |
| raise _InvalidSupportBundle | |
| flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | |
| flags |= ( | |
| getattr(os, "O_NOFOLLOW", 0) | |
| | getattr(os, "O_CLOEXEC", 0) | |
| | getattr(os, "O_NONBLOCK", 0) | |
| ) | |
| descriptor = os.open(path, flags) | |
| opened = os.fstat(descriptor) | |
| if ( | |
| not stat.S_ISREG(opened.st_mode) | |
| or opened.st_size > MAX_VALIDATION_BYTES | |
| or opened.st_dev != before.st_dev | |
| or opened.st_ino != before.st_ino | |
| ): | |
| raise _InvalidSupportBundle | |
| with os.fdopen(descriptor, "rb") as handle: | |
| descriptor = None | |
| data = bytearray() | |
| while len(data) <= MAX_VALIDATION_BYTES: | |
| block = handle.read(min(64 * 1024, MAX_VALIDATION_BYTES + 1 - len(data))) | |
| if not block: | |
| break | |
| data.extend(block) | |
| if len(data) > MAX_VALIDATION_BYTES: | |
| raise _InvalidSupportBundle | |
| return bytes(data) | |
| except (OSError, TypeError, ValueError): | |
| raise _InvalidSupportBundle from None | |
| finally: | |
| if descriptor is not None: | |
| os.close(descriptor) | |
| def _require_object(value: object, keys: set[str]) -> dict[str, Any]: | |
| if not isinstance(value, dict) or set(value) != keys: | |
| raise _InvalidSupportBundle | |
| return value | |
| def _require_text(value: object, *, maximum: int = 256) -> None: | |
| if not isinstance(value, str) or len(value) > maximum: | |
| raise _InvalidSupportBundle | |
| try: | |
| if len(value.encode("utf-8", errors="strict")) > maximum * 4: | |
| raise _InvalidSupportBundle | |
| except UnicodeEncodeError: | |
| raise _InvalidSupportBundle from None | |
| def _require_token(value: object, *, nullable: bool = False) -> None: | |
| if nullable and value is None: | |
| return | |
| if _token(value) is None: | |
| raise _InvalidSupportBundle | |
| def _require_bool(value: object, *, nullable: bool = False) -> None: | |
| if nullable and value is None: | |
| return | |
| if type(value) is not bool: | |
| raise _InvalidSupportBundle | |
| def _require_count(value: object, *, nullable: bool = False, maximum: int = MAX_AGGREGATE_COUNT) -> None: | |
| if nullable and value is None: | |
| return | |
| if type(value) is not int or value < 0 or value > maximum: | |
| raise _InvalidSupportBundle | |
| def _require_finite_number(value: object, *, nullable: bool = False) -> None: | |
| if nullable and value is None: | |
| return | |
| if type(value) is int: | |
| if value < 0 or value > MAX_AGGREGATE_COUNT: | |
| raise _InvalidSupportBundle | |
| return | |
| if type(value) is not float or not math.isfinite(value) or value < 0: | |
| raise _InvalidSupportBundle | |
| def _validate_state_counts(value: object, total: int) -> int: | |
| if not isinstance(value, dict) or len(value) > MAX_STATE_KEYS: | |
| raise _InvalidSupportBundle | |
| subtotal = 0 | |
| for state, count in value.items(): | |
| _require_token(state) | |
| _require_count(count, maximum=total) | |
| if count == 0: | |
| raise _InvalidSupportBundle | |
| subtotal += count | |
| if subtotal > total: | |
| raise _InvalidSupportBundle | |
| return len(value) | |
| def _validate_health(value: object) -> None: | |
| health = _require_object( | |
| value, | |
| { | |
| "result", | |
| "backend", | |
| "version", | |
| "workload_uid", | |
| "autonomous_discovery", | |
| "cgroup_v2", | |
| "pidfd", | |
| "execution_boundary", | |
| "installation", | |
| "incidents", | |
| "network", | |
| "discovery", | |
| }, | |
| ) | |
| for name in ("result", "backend", "version"): | |
| _require_token(health[name], nullable=True) | |
| _require_count(health["workload_uid"], nullable=True, maximum=2**32 - 1) | |
| for name in ("autonomous_discovery", "cgroup_v2", "pidfd"): | |
| _require_bool(health[name], nullable=True) | |
| execution_boundary = health["execution_boundary"] | |
| if execution_boundary is not None: | |
| execution = _require_object( | |
| execution_boundary, | |
| {"linux", "architecture", "fcntl", "libc", "user_notification", "supported"}, | |
| ) | |
| for fact in execution.values(): | |
| _require_bool(fact) | |
| installation = health["installation"] | |
| if installation is not None: | |
| if not isinstance(installation, dict) or set(installation) not in ( | |
| {"state", "journal", "files_match"}, | |
| {"state", "journal", "files_match", "manifest_version"}, | |
| ): | |
| raise _InvalidSupportBundle | |
| _require_token(installation["state"]) | |
| _require_bool(installation["journal"]) | |
| _require_bool(installation["files_match"]) | |
| if "manifest_version" in installation: | |
| _require_token(installation["manifest_version"]) | |
| incidents = health["incidents"] | |
| if incidents is not None: | |
| incident_health = _require_object( | |
| incidents, | |
| {"healthy", "count", "active", "lockdown"}, | |
| ) | |
| _require_bool(incident_health["healthy"]) | |
| _require_bool(incident_health["lockdown"]) | |
| _require_count(incident_health["count"]) | |
| _require_count(incident_health["active"], maximum=incident_health["count"]) | |
| network = _require_object( | |
| health["network"], | |
| {"mode", "cleanup_healthy", "primitives_supported"}, | |
| ) | |
| _require_token(network["mode"], nullable=True) | |
| _require_bool(network["cleanup_healthy"], nullable=True) | |
| _require_bool(network["primitives_supported"], nullable=True) | |
| discovery = _require_object( | |
| health["discovery"], | |
| { | |
| "healthy", | |
| "consecutive_failures", | |
| "last_scan_duration_ms", | |
| "last_scan_completed", | |
| "receipt_persistence_healthy", | |
| }, | |
| ) | |
| _require_bool(discovery["healthy"], nullable=True) | |
| _require_count(discovery["consecutive_failures"], nullable=True) | |
| _require_finite_number(discovery["last_scan_duration_ms"], nullable=True) | |
| _require_bool(discovery["last_scan_completed"], nullable=True) | |
| _require_bool(discovery["receipt_persistence_healthy"], nullable=True) | |
| def _validate_support_bundle(value: object) -> dict[str, Any]: | |
| bundle = _require_object( | |
| value, | |
| { | |
| "schema_version", | |
| "generated_utc", | |
| "product", | |
| "version", | |
| "host", | |
| "health", | |
| "workloads", | |
| "incidents", | |
| "receipts", | |
| "privacy", | |
| }, | |
| ) | |
| if bundle["schema_version"] != SUPPORT_BUNDLE_SCHEMA or bundle["product"] != "Lumi Eggcracker": | |
| raise _InvalidSupportBundle | |
| generated = bundle["generated_utc"] | |
| if ( | |
| not isinstance(generated, str) | |
| or not re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?Z", generated) | |
| ): | |
| raise _InvalidSupportBundle | |
| try: | |
| dt.datetime.fromisoformat(generated) | |
| except ValueError: | |
| raise _InvalidSupportBundle from None | |
| _require_token(bundle["version"]) | |
| host = _require_object( | |
| bundle["host"], | |
| { | |
| "platform", | |
| "kernel", | |
| "machine", | |
| "python", | |
| "cgroup_v2", | |
| "controllers", | |
| "pidfd_open", | |
| "pidfd_send_signal", | |
| "systemd", | |
| }, | |
| ) | |
| for name in ("platform", "kernel", "machine", "python"): | |
| _require_text(host[name]) | |
| for name in ("cgroup_v2", "pidfd_open", "pidfd_send_signal"): | |
| _require_bool(host[name]) | |
| controllers = host["controllers"] | |
| if ( | |
| not isinstance(controllers, list) | |
| or len(controllers) > 3 | |
| or any(controller not in {"cpu", "memory", "pids"} for controller in controllers) | |
| or controllers != sorted(set(controllers)) | |
| ): | |
| raise _InvalidSupportBundle | |
| if host["systemd"] is not None: | |
| _require_text(host["systemd"], maximum=80) | |
| _validate_health(bundle["health"]) | |
| workloads = _require_object(bundle["workloads"], {"run_count", "states"}) | |
| _require_count(workloads["run_count"]) | |
| workload_state_count = _validate_state_counts(workloads["states"], workloads["run_count"]) | |
| incidents = _require_object(bundle["incidents"], {"count", "active", "states"}) | |
| _require_count(incidents["count"]) | |
| _require_count(incidents["active"], maximum=incidents["count"]) | |
| incident_state_count = _validate_state_counts(incidents["states"], incidents["count"]) | |
| active_from_states = sum( | |
| count for state, count in incidents["states"].items() if state in {"ACTIVE", "ACKNOWLEDGED"} | |
| ) | |
| if incidents["active"] != active_from_states: | |
| raise _InvalidSupportBundle | |
| receipts = bundle["receipts"] | |
| if not isinstance(receipts, list) or len(receipts) > MAX_DETECTIONS: | |
| raise _InvalidSupportBundle | |
| for receipt in receipts: | |
| if not isinstance(receipt, dict) or set(receipt) not in ( | |
| {"event_id", "result", "trigger", "version", "profile"}, | |
| {"event_id", "result", "trigger", "version", "profile", "boundary"}, | |
| ): | |
| raise _InvalidSupportBundle | |
| for name in ("event_id", "result", "trigger", "version", "profile"): | |
| _require_token(receipt[name], nullable=True) | |
| if "boundary" in receipt: | |
| boundary = _require_object( | |
| receipt["boundary"], | |
| {"address_family", "mode", "policy_sha256", "violation"}, | |
| ) | |
| for name in ("address_family", "mode", "violation"): | |
| _require_token(boundary[name], nullable=True) | |
| digest = boundary["policy_sha256"] | |
| if digest is not None and ( | |
| not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", digest) | |
| ): | |
| raise _InvalidSupportBundle | |
| privacy = _require_object( | |
| bundle["privacy"], | |
| {"network", "raw_receipts", "argv", "paths", "pids", "model_data"}, | |
| ) | |
| if ( | |
| type(privacy["network"]) is not str | |
| or privacy["network"] != "aggregate-boundary-events-only" | |
| or any( | |
| type(privacy[name]) is not bool or privacy[name] is not False | |
| for name in ("raw_receipts", "argv", "paths", "pids", "model_data") | |
| ) | |
| ): | |
| raise _InvalidSupportBundle | |
| return { | |
| "result": "STRUCTURE_VALID", | |
| "schema_version": SUPPORT_BUNDLE_SCHEMA, | |
| "receipts": { | |
| "included": len(receipts), | |
| "maximum_included": MAX_DETECTIONS, | |
| "total_represented": False, | |
| }, | |
| "workloads": { | |
| "run_count": workloads["run_count"], | |
| "reported_state_count": workload_state_count, | |
| }, | |
| "incidents": { | |
| "count": incidents["count"], | |
| "active_count": incidents["active"], | |
| "reported_state_count": incident_state_count, | |
| }, | |
| "limitations": ( | |
| "Structural validation only; not authentication, privacy assurance, completeness, " | |
| "containment, or incident truth." | |
| ), | |
| } | |
| def validate_bundle(path: Path) -> dict[str, Any]: | |
| """Validate a bounded v1 support bundle and return only aggregate counts.""" | |
| try: | |
| content = _read_bounded_regular_file(Path(path)) | |
| text = content.decode("utf-8", errors="strict") | |
| _check_json_depth(text) | |
| value = json.loads( | |
| text, | |
| object_pairs_hook=_unique_json_object, | |
| parse_constant=_reject_json_constant, | |
| parse_float=_finite_json_float, | |
| ) | |
| return _validate_support_bundle(value) | |
| except ( | |
| _InvalidSupportBundle, | |
| UnicodeDecodeError, | |
| json.JSONDecodeError, | |
| RecursionError, | |
| TypeError, | |
| ValueError, | |
| ): | |
| raise JsonInputError("support bundle is invalid or cannot be read") from None | |
| def validate_main(arguments: list[str]) -> int: | |
| if len(arguments) != 1: | |
| print( | |
| "eggcracker validate-support-bundle: expected exactly one FILE argument", | |
| file=sys.stderr, | |
| ) | |
| return 4 | |
| try: | |
| summary = validate_bundle(Path(arguments[0])) | |
| except JsonInputError: | |
| print( | |
| "eggcracker validate-support-bundle: support bundle is invalid or cannot be read", | |
| file=sys.stderr, | |
| ) | |
| return 4 | |
| print(json.dumps(summary, sort_keys=True, separators=(",", ":"))) | |
| return 0 | |