"""Public CLI for protected workloads and autonomous AI-runtime enforcement.""" from __future__ import annotations import argparse import json import math import os import re import sys import time from pathlib import Path from . import __version__ from .brokered.compare_detection_receipts import main as compare_detection_receipts_main from .client import RequestDeadlineExceeded, request, request_before from .gate import main as gate_main from .jsonio import JsonInputError, write_new_json from .kill_receipt import main as validate_kill_receipt_main from .kill_receipt import prepare_handoff_main from .supervisor import main as supervisor_main from .support_bundle import main as support_bundle_main from .support_bundle import validate_main as validate_support_bundle_main from .watchdog import main as watchdog_main _DOCTOR_SUMMARY_SCHEMA = "lumi-eggcracker.doctor-summary.v1" _DOCTOR_SUMMARY_ERROR = "eggcracker: doctor summary unavailable" _CHECK_START_SCHEMA = "lumi-eggcracker.check-start.v1" _CHECK_START_ERROR = "eggcracker: check-start response unavailable" _CHECK_START_REASONS = frozenset( { "APPROVAL_MISSING", "EXACT_APPROVAL_MATCHED", "EXECUTION_POLICY_INVALID", "INVALID_ARGUMENTS", "LOCKDOWN", "NAME_UNAVAILABLE", "STATE_UNAVAILABLE", } ) _CHECK_START_NEXT_ACTIONS = { "EXACT_APPROVAL_MATCHED": ( "run start --require-approval immediately; admission is revalidated" ), "APPROVAL_MISSING": ( "create or restore the exact root approval for this tuple, then check again" ), "EXECUTION_POLICY_INVALID": ( "use a valid root-created execution policy or omit it, then check again" ), "INVALID_ARGUMENTS": "correct the launch tuple, then check again", "LOCKDOWN": "clear the exact local lockdown through root recovery, then check again", "NAME_UNAVAILABLE": "choose an unused workload name, then check again", "STATE_UNAVAILABLE": "repair unavailable supervisor state, then check again", } _CHECK_START_RESULTS = frozenset({"READY_TO_ATTEMPT", "NOT_READY"}) _DOCTOR_SUMMARY_INSTALLATION_STATES = frozenset( {"HEALTHY", "DRIFT", "RECOVERY_REQUIRED", "NOT_INSTALLED"} ) _DOCTOR_SUMMARY_NETWORK_MODES = frozenset({"offline", "unsupported"}) _DOCTOR_RECOVERY_PLAN_SCHEMA = "lumi-eggcracker.doctor-recovery-plan.v1" _DOCTOR_RECOVERY_PLAN_ERROR = "eggcracker: doctor recovery plan unavailable" _DOCTOR_RECOVERY_PLAN_ACTIONS = { "cgroup_v2": "MANUAL_HOST_REVIEW", "discovery": "MANUAL_SUPERVISOR_REVIEW", "execution_boundary": "MANUAL_POLICY_REVIEW", "incidents": "MANUAL_INCIDENT_REVIEW", "network_cleanup": "MANUAL_SUPERVISOR_REVIEW", "network_mode": "MANUAL_POLICY_REVIEW", "network_primitives": "MANUAL_HOST_REVIEW", "pidfd": "MANUAL_HOST_REVIEW", "receipt_persistence": "MANUAL_SUPERVISOR_REVIEW", "supervisor_readiness": "MANUAL_SUPERVISOR_REVIEW", "workload_identity": "MANUAL_IDENTITY_REVIEW", } _DOCTOR_RECOVERY_PLAN_INSTALLATION_ACTIONS = { "NOT_INSTALLED": "MANUAL_INSTALLATION_REQUIRED", "DRIFT": "MANUAL_INSTALLATION_REPAIR", "RECOVERY_REQUIRED": "MANUAL_INSTALLATION_RECOVERY", } _WAIT_PUBLIC_STATUS_KEYS = frozenset( { "exec_policy_digest", "exec_policy_id", "incident_lockdown", "name", "network_mode", "run_id", "state", "unit", "workload_uid", } ) _WAIT_ACTIVE_STATES = frozenset({"STARTING", "RUNNING"}) _WAIT_TERMINAL_STATES = frozenset( { "COMPLETED_ALLOWED", "CONTAINED_RECEIPT_FAILED", "CONTAINMENT_FAILED", "TERMINATED", } ) _WAIT_NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}\Z") _WAIT_RUN_ID = re.compile(r"[0-9a-f]{24}\Z") _WAIT_UNIT_PREFIX = "lumi-eggcracker-workload-" _WAIT_HEX = re.compile(r"[0-9a-f]{64}\Z") _WAIT_POLL_CADENCE_SECONDS = 0.1 _WAIT_ERROR = "ERROR" _WAIT_TERMINAL = "TERMINAL" _WAIT_TIMED_OUT = "TIMED_OUT" _WAIT_EXPECTATION_MISMATCH = 7 class _DoctorSummaryInvalid(ValueError): """The supervisor response cannot be safely reduced to a doctor summary.""" def _doctor_summary(value: object) -> dict[str, object]: """Reduce a doctor response to a fixed, nonidentifying health object.""" if type(value) is not dict: raise _DoctorSummaryInvalid result = value.get("result") if type(result) is not str or result not in {"PASS", "UNSUPPORTED"}: raise _DoctorSummaryInvalid def required_bool(container: object, key: str) -> bool: if type(container) is not dict or key not in container: raise _DoctorSummaryInvalid item = container[key] if type(item) is not bool: raise _DoctorSummaryInvalid return item autonomous_discovery = required_bool(value, "autonomous_discovery") cgroup_v2 = required_bool(value, "cgroup_v2") pidfd = required_bool(value, "pidfd") network = value.get("network") if type(network) is not dict: raise _DoctorSummaryInvalid network_mode = network.get("mode") if type(network_mode) is not str or network_mode not in _DOCTOR_SUMMARY_NETWORK_MODES: raise _DoctorSummaryInvalid network_primitives = network.get("primitives") primitives_supported = required_bool(network_primitives, "supported") network_cleanup = required_bool(network, "cleanup_healthy") execution_boundary = value.get("execution_boundary") execution_supported = required_bool(execution_boundary, "supported") discovery = value.get("discovery") discovery_healthy = required_bool(discovery, "healthy") receipt_persistence = required_bool(discovery, "receipt_persistence_healthy") incidents = value.get("incidents") incidents_healthy = required_bool(incidents, "healthy") incidents_lockdown = required_bool(incidents, "lockdown") if not incidents_healthy and not incidents_lockdown: raise _DoctorSummaryInvalid installation = value.get("installation") if type(installation) is not dict or "state" not in installation: raise _DoctorSummaryInvalid installation_state = installation["state"] if ( type(installation_state) is not str or installation_state not in _DOCTOR_SUMMARY_INSTALLATION_STATES ): raise _DoctorSummaryInvalid workload_identity = value.get("workload_identity") workload_identity_healthy = required_bool(workload_identity, "healthy") failed_checks: set[str] = set() if not cgroup_v2: failed_checks.add("cgroup_v2") if not pidfd: failed_checks.add("pidfd") if network_mode != "offline": failed_checks.add("network_mode") if not primitives_supported: failed_checks.add("network_primitives") if not network_cleanup: failed_checks.add("network_cleanup") if not execution_supported: failed_checks.add("execution_boundary") if not discovery_healthy: failed_checks.add("discovery") if not receipt_persistence: failed_checks.add("receipt_persistence") if not incidents_healthy: failed_checks.add("incidents") if installation_state != "HEALTHY": failed_checks.add("installation") if not workload_identity_healthy: failed_checks.add("workload_identity") if result == "PASS": if not autonomous_discovery or failed_checks: raise _DoctorSummaryInvalid summary_result = "READY" else: if autonomous_discovery: raise _DoctorSummaryInvalid if not failed_checks: # The supervisor also gates readiness on quarantine_root, which is # deliberately not exposed by its public doctor response. failed_checks.add("supervisor_readiness") summary_result = "NOT_READY" return { "failed_checks": sorted(failed_checks), "next_action": ( "NONE" if summary_result == "READY" else "REVIEW_SUPERVISOR_READINESS" ), "result": summary_result, "schema": _DOCTOR_SUMMARY_SCHEMA, } def _doctor_recovery_plan(value: object) -> dict[str, object]: """Reduce a doctor response to a fixed, nonidentifying recovery plan.""" summary = _doctor_summary(value) installation = value["installation"] installation_state = installation["state"] failed_checks = summary["failed_checks"] actions = { _DOCTOR_RECOVERY_PLAN_ACTIONS[check] for check in failed_checks if check != "installation" } if installation_state != "HEALTHY": actions.add(_DOCTOR_RECOVERY_PLAN_INSTALLATION_ACTIONS[installation_state]) return { "actions": sorted(actions), "failed_checks": failed_checks, "installation_state": installation_state, "result": summary["result"], "schema": _DOCTOR_RECOVERY_PLAN_SCHEMA, } def _check_start_projection(value: object) -> dict[str, str]: """Accept only the supervisor's fixed, nonidentifying check result.""" if type(value) is not dict or set(value) != { "next_action", "reason_code", "result", "schema", }: raise ValueError schema = value["schema"] reason_code = value["reason_code"] result = value["result"] next_action = value["next_action"] if ( type(schema) is not str or schema != _CHECK_START_SCHEMA or type(reason_code) is not str or reason_code not in _CHECK_START_REASONS or type(result) is not str or result not in _CHECK_START_RESULTS or type(next_action) is not str or next_action != _CHECK_START_NEXT_ACTIONS[reason_code] or (result == "READY_TO_ATTEMPT") != (reason_code == "EXACT_APPROVAL_MATCHED") ): raise ValueError return { "next_action": next_action, "reason_code": reason_code, "result": result, "schema": schema, } def _wait_status_projection(value: object, *, name: str) -> dict[str, object]: """Validate and close the supervisor's public run projection.""" if type(value) is not dict or set(value) != _WAIT_PUBLIC_STATUS_KEYS: raise ValueError if ( type(value["name"]) is not str or value["name"] != name or _WAIT_NAME.fullmatch(value["name"]) is None ): raise ValueError run_id = value["run_id"] if type(run_id) is not str or _WAIT_RUN_ID.fullmatch(run_id) is None: raise ValueError state = value["state"] if type(state) is not str or state not in _WAIT_ACTIVE_STATES | _WAIT_TERMINAL_STATES: raise ValueError unit = value["unit"] if type(unit) is not str or unit != f"{_WAIT_UNIT_PREFIX}{run_id}.service": raise ValueError network_mode = value["network_mode"] if type(network_mode) is not str or network_mode not in {"offline", "none"}: raise ValueError exec_policy_id = value["exec_policy_id"] if type(exec_policy_id) is not str or _WAIT_NAME.fullmatch(exec_policy_id) is None: raise ValueError exec_policy_digest = value["exec_policy_digest"] if type(exec_policy_digest) is not str or _WAIT_HEX.fullmatch(exec_policy_digest) is None: raise ValueError if type(value["incident_lockdown"]) is not bool: raise ValueError workload_uid = value["workload_uid"] if type(workload_uid) is not int or workload_uid < 0: raise ValueError return { "exec_policy_digest": exec_policy_digest, "exec_policy_id": exec_policy_id, "incident_lockdown": value["incident_lockdown"], "name": value["name"], "network_mode": network_mode, "run_id": run_id, "state": state, "unit": unit, "workload_uid": workload_uid, } class _WaitClock: """Read one monotonic clock and reject non-monotonic or invalid samples.""" def __init__(self) -> None: self._last: float | None = None def now(self) -> float: value = time.monotonic() if ( isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or self._last is not None and value < self._last ): raise ValueError self._last = float(value) return self._last def _wait_output(marker: str, state: dict[str, object] | None = None) -> None: print(marker) if state is not None: print(json.dumps(state, sort_keys=True)) def _wait_error() -> int: print(_WAIT_ERROR, file=sys.stderr) return 4 def _wait_main( *, name: str, timeout_seconds: int, expect_state: str | None = None ) -> int: clock = _WaitClock() try: started = clock.now() deadline = started + timeout_seconds if not math.isfinite(deadline): raise ValueError last_state: dict[str, object] | None = None pinned_run_id: str | None = None while True: remaining = deadline - clock.now() if remaining <= 0: _wait_output(_WAIT_TIMED_OUT, last_state) return 6 try: value = request_before( "status", deadline_monotonic=deadline, name=name ) except (RequestDeadlineExceeded, TimeoutError): _wait_output(_WAIT_TIMED_OUT, last_state) return 6 except Exception: # noqa: BLE001 - wait failures are constant and closed return _wait_error() # A response completing at or after the absolute deadline cannot be # treated as a validated observation for this bounded wait. if clock.now() >= deadline: _wait_output(_WAIT_TIMED_OUT, last_state) return 6 try: state = _wait_status_projection(value, name=name) except Exception: # noqa: BLE001 - malformed status is constant return _wait_error() run_id = state["run_id"] if pinned_run_id is None: pinned_run_id = run_id elif run_id != pinned_run_id: return _wait_error() last_state = state if state["state"] in _WAIT_TERMINAL_STATES: _wait_output(_WAIT_TERMINAL, state) if expect_state is None or state["state"] == expect_state: return 0 return _WAIT_EXPECTATION_MISMATCH remaining = deadline - clock.now() if remaining <= 0: _wait_output(_WAIT_TIMED_OUT, last_state) return 6 time.sleep(min(_WAIT_POLL_CADENCE_SECONDS, remaining)) except Exception: # noqa: BLE001 - clock and sleep failures are constant return _wait_error() class _RejectDuplicateIdentifier(argparse.Action): """Accept one identifier option and fail closed on a repeated spelling.""" def __call__( self, parser: argparse.ArgumentParser, namespace: argparse.Namespace, values: object, option_string: str | None = None, ) -> None: marker = f"_eggcracker_seen_{self.dest}" if getattr(namespace, marker, False): raise argparse.ArgumentError( self, f"{option_string or self.dest} may only be specified once", ) setattr(namespace, marker, True) setattr(namespace, self.dest, values) class _CheckStartParseFailure(Exception): """Internal marker for one closed check-start parse failure.""" class _CheckStartHelp(Exception): """Internal marker for successfully rendered check-start help.""" class _CheckStartParser(argparse.ArgumentParser): """Keep check-start parser failures fixed and nonidentifying.""" def error(self, message: str) -> None: raise _CheckStartParseFailure from None def exit(self, status: int = 0, message: str | None = None) -> None: if status == 0: raise _CheckStartHelp raise _CheckStartParseFailure from None def _add_check_start_arguments(parser: argparse.ArgumentParser) -> None: parser.add_argument("--name", required=True, action=_RejectDuplicateIdentifier) parser.add_argument("--max-pids", required=True, type=int) parser.add_argument("--max-memory-mib", default=2048, type=int) parser.add_argument("--cpu-quota-percent", default=400, type=int) parser.add_argument( "--exec-policy", action=_RejectDuplicateIdentifier, help="root-created sealed-exec policy identifier", ) parser.add_argument("argv", nargs=argparse.REMAINDER) def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="eggcracker") commands = parser.add_subparsers(dest="command", required=True) start = commands.add_parser("start", help="launch an explicitly selected protected workload") start.add_argument("--name", required=True, action=_RejectDuplicateIdentifier) start.add_argument("--max-pids", required=True, type=int) start.add_argument("--max-memory-mib", default=2048, type=int) start.add_argument("--cpu-quota-percent", default=400, type=int) start.add_argument( "--exec-policy", action=_RejectDuplicateIdentifier, help="root-created sealed-exec policy identifier", ) start.add_argument( "--require-approval", action="store_true", help="reject the launch unless an exact approval exists at admission time", ) start.add_argument("argv", nargs=argparse.REMAINDER) check_start = commands.add_parser( "check-start", help="check one exact approved launch without reserving or starting it", ) _add_check_start_arguments(check_start) kill = commands.add_parser("kill", help="terminate one protected workload") kill.add_argument("--name", required=True, action=_RejectDuplicateIdentifier) kill.add_argument("--receipt", required=True, type=Path) status = commands.add_parser("status", help="show one protected workload") status.add_argument("--name", required=True, action=_RejectDuplicateIdentifier) wait = commands.add_parser( "wait", help="wait for one protected workload to reach a terminal state", description=( "Wait for one protected workload to reach a terminal state. With " "--expect-state, a mismatch exits 7. COMPLETED_ALLOWED means an " "observed terminal/cgroup-empty completion; it does not mean " "application exit-code success or provide a safety guarantee." ), ) wait.add_argument("--name", required=True, action=_RejectDuplicateIdentifier) wait.add_argument( "--timeout-seconds", required=True, type=_wait_timeout, action=_RejectDuplicateIdentifier, help="maximum wait duration in seconds (1-300)", ) wait.add_argument( "--expect-state", choices=sorted(_WAIT_TERMINAL_STATES), action=_RejectDuplicateIdentifier, help=( "expected terminal state; mismatch exits 7 (COMPLETED_ALLOWED is " "observed cgroup-empty completion, not application success or a " "safety guarantee)" ), ) commands.add_parser("list", help="list protected workloads") history = commands.add_parser( "history", help="list retained terminal records in recorded monotonic order", ) history.add_argument( "--limit", default=20, type=_history_limit, action=_RejectDuplicateIdentifier, help="number of retained terminal records to return (1-20; default: 20)", ) history.add_argument( "--state", choices=sorted(_WAIT_TERMINAL_STATES), action=_RejectDuplicateIdentifier, help="show only records in the selected terminal state", ) approve = commands.add_parser("approve", help="approve one exact AI runtime invocation") approve.add_argument("--name", required=True, action=_RejectDuplicateIdentifier) approve.add_argument("--uid", required=True, type=int) approve.add_argument("--max-pids", default=64, type=int) approve.add_argument("--max-memory-mib", default=2048, type=int) approve.add_argument("--cpu-quota-percent", default=400, type=int) approve.add_argument( "--allow-interface-discovery", action="store_true", help="allow unprivileged interface lookup inside the offline namespace", ) approve.add_argument("argv", nargs=argparse.REMAINDER) revoke = commands.add_parser("revoke", help="revoke one exact AI runtime approval") revoke.add_argument("--name", required=True, action=_RejectDuplicateIdentifier) commands.add_parser("approvals", help="list exact AI runtime approvals") exec_policy = commands.add_parser("exec-policy", help="manage root-created executable policies") exec_commands = exec_policy.add_subparsers(dest="exec_action", required=True) create_policy = exec_commands.add_parser("create", help="create an immutable executable policy") create_policy.add_argument( "--name", required=True, action=_RejectDuplicateIdentifier ) create_policy.add_argument("paths", nargs=argparse.REMAINDER) revoke_policy = exec_commands.add_parser("revoke", help="revoke an executable policy") revoke_policy.add_argument( "--policy-id", required=True, action=_RejectDuplicateIdentifier ) exec_commands.add_parser("list", help="list executable policies") commands.add_parser("detections", help="list autonomous containment summaries") commands.add_parser("incidents", help="list bounded local lockdown incidents") incident = commands.add_parser("incident", help="inspect or clear one local lockdown incident") incident_commands = incident.add_subparsers(dest="incident_action", required=True) for action, help_text in ( ("show", "show one root-only incident detail"), ("acknowledge", "acknowledge one incident as root"), ("clear", "clear one local lockdown as root"), ): command = incident_commands.add_parser(action, help=help_text) command.add_argument("incident_id") doctor = commands.add_parser("doctor", help="check the installed protected supervisor") doctor_projection = doctor.add_mutually_exclusive_group() doctor_projection.add_argument( "--summary", action="store_true", help="emit a bounded readiness summary instead of the full doctor response", ) doctor_projection.add_argument( "--recovery-plan", action="store_true", help="emit a bounded manual recovery plan instead of the full doctor response", ) support = commands.add_parser( "support-bundle", help="write a local redacted health and receipt bundle" ) support.add_argument("--output", required=True, type=Path) validate_support = commands.add_parser( "validate-support-bundle", help="offline structural review of a support bundle", ) validate_support.add_argument("FILE", type=Path) validate_kill_receipt = commands.add_parser( "validate-kill-receipt", help="offline structural review of one manual kill receipt", ) validate_kill_receipt.add_argument("FILE", type=Path) handoff = commands.add_parser( "prepare-kill-receipt-handoff", help="prepare a local redacted manual kill receipt handoff", ) handoff.add_argument("RECEIPT", type=Path) handoff.add_argument( "--route", required=True, choices=("q-and-a", "bug-report", "private-security"), ) handoff.add_argument("--output", required=True, type=Path) commands.add_parser( "compare-detection-receipts", help="compare two redacted detection receipt exports", ) commands.add_parser("version", help="print the public version") return parser def _command(argv: list[str]) -> list[str]: return argv[1:] if argv[:1] == ["--"] else argv def _history_limit(value: str) -> int: if not isinstance(value, str) or not value.isascii() or not value.isdecimal(): raise argparse.ArgumentTypeError("history limit must be an integer from 1 to 20") result = int(value) if not 1 <= result <= 20: raise argparse.ArgumentTypeError("history limit must be an integer from 1 to 20") return result def _wait_timeout(value: str) -> int: if not isinstance(value, str) or not value.isascii() or not value.isdecimal(): raise argparse.ArgumentTypeError("wait timeout must be an integer from 1 to 300") result = int(value) if not 1 <= result <= 300: raise argparse.ArgumentTypeError("wait timeout must be an integer from 1 to 300") return result def _check_start_main(values: list[str]) -> int: parser = _CheckStartParser(prog="eggcracker check-start") _add_check_start_arguments(parser) try: args = parser.parse_args(values) check_args = { "name": args.name, "max_pids": args.max_pids, "max_memory_mib": args.max_memory_mib, "cpu_quota_percent": args.cpu_quota_percent, "argv": _command(args.argv), } if args.exec_policy is not None: check_args["exec_policy"] = args.exec_policy value = _check_start_projection(request("check_start", **check_args)) except _CheckStartHelp: return 0 except Exception: # noqa: BLE001 - parse/transport/shape failures are constant print(_CHECK_START_ERROR, file=sys.stderr) return 4 print(json.dumps(value, sort_keys=True)) return 0 if value["result"] == "READY_TO_ATTEMPT" else 6 def main(argv: list[str] | None = None) -> int: values = sys.argv[1:] if argv is None else argv if values[:1] == ["check-start"]: return _check_start_main(values[1:]) if values[:1] == ["compare-detection-receipts"]: return compare_detection_receipts_main( values[1:], prog="eggcracker compare-detection-receipts" ) if values[:1] == ["validate-support-bundle"]: if values[1:] == ["--help"]: _parser().parse_args(values) return 0 return validate_support_bundle_main(values[1:]) if values[:1] == ["validate-kill-receipt"]: if values[1:] == ["--help"]: _parser().parse_args(values) return 0 return validate_kill_receipt_main(values[1:]) if values[:1] == ["prepare-kill-receipt-handoff"]: if values[1:] == ["--help"]: _parser().parse_args(values) return 0 return prepare_handoff_main(values[1:]) if values[:1] == ["_supervisor"]: return supervisor_main(values[1:]) if values[:1] == ["_gate"]: return gate_main(values[1:]) if values[:1] == ["_watchdog"]: return watchdog_main(values[1:]) args = _parser().parse_args(values) if args.command == "version": print(__version__) return 0 if args.command == "support-bundle": return support_bundle_main(args.output) if args.command == "wait": return _wait_main( name=args.name, timeout_seconds=args.timeout_seconds, expect_state=args.expect_state, ) try: if args.command == "doctor": if args.summary: try: summary = _doctor_summary(request("doctor")) except Exception: # noqa: BLE001 - summary failures are intentionally constant print(_DOCTOR_SUMMARY_ERROR, file=sys.stderr) return 6 print(json.dumps(summary, sort_keys=True)) return 0 if summary["result"] == "READY" else 6 if args.recovery_plan: try: recovery_plan = _doctor_recovery_plan(request("doctor")) except Exception: # noqa: BLE001 - plan failures are intentionally constant print(_DOCTOR_RECOVERY_PLAN_ERROR, file=sys.stderr) return 6 print(json.dumps(recovery_plan, sort_keys=True)) return 0 if recovery_plan["result"] == "READY" else 6 value = request("doctor") print(json.dumps(value, sort_keys=True)) return 0 if value.get("result") == "PASS" else 6 if args.command == "start": start_args = { "action": "start", "name": args.name, "max_pids": args.max_pids, "max_memory_mib": args.max_memory_mib, "cpu_quota_percent": args.cpu_quota_percent, "argv": _command(args.argv), } if args.exec_policy is not None: start_args["exec_policy"] = args.exec_policy if args.require_approval: start_args["require_approval"] = True action = start_args.pop("action") value = request(action, **start_args) elif args.command == "status": value = request("status", name=args.name) elif args.command == "list": value = request("list") elif args.command == "history": history_args: dict[str, object] = {"limit": args.limit} if args.state is not None: history_args["state"] = args.state value = request("history", **history_args) elif args.command == "approve": if os.geteuid() != 0: raise JsonInputError("approve requires root administrative authority; use sudo") value = request( "approve", name=args.name, uid=args.uid, max_pids=args.max_pids, max_memory_mib=args.max_memory_mib, cpu_quota_percent=args.cpu_quota_percent, allow_interface_discovery=args.allow_interface_discovery, argv=_command(args.argv), ) elif args.command == "revoke": if os.geteuid() != 0: raise JsonInputError("revoke requires root administrative authority; use sudo") value = request("revoke", name=args.name) elif args.command == "approvals": value = request("approvals") elif args.command == "exec-policy": if args.exec_action == "list": value = request("exec_policies") elif args.exec_action == "create": if os.geteuid() != 0: raise JsonInputError("execution policy creation requires root administrative authority; use sudo") value = request("exec_policy_create", name=args.name, paths=_command(args.paths)) else: if os.geteuid() != 0: raise JsonInputError("execution policy revocation requires root administrative authority; use sudo") value = request("exec_policy_revoke", policy_id=args.policy_id) elif args.command == "detections": value = request("detections") elif args.command == "incidents": value = request("incidents") elif args.command == "incident": if os.geteuid() != 0: raise JsonInputError("incident administration requires root administrative authority; use sudo") action = { "show": "incident_show", "acknowledge": "incident_acknowledge", "clear": "incident_clear", }[args.incident_action] value = request(action, incident_id=args.incident_id) else: # This validation is deliberately read-only: no receipt reservation before kill. if args.receipt.exists() or args.receipt.is_symlink() or not args.receipt.parent.is_dir(): raise JsonInputError("receipt must be a new file under an existing directory") value = request("kill", name=args.name) write_new_json(args.receipt, value) print(json.dumps(value, sort_keys=True)) return 0 except (JsonInputError, OSError) as error: print(f"eggcracker: {error}", file=sys.stderr) return 4