Spaces:
Running
Running
| from __future__ import annotations | |
| import contextlib | |
| import io | |
| import json | |
| import os | |
| import socket | |
| import stat | |
| import struct | |
| import sys | |
| import tempfile | |
| import unittest | |
| from pathlib import Path | |
| from types import SimpleNamespace | |
| from unittest import mock | |
| from lumi_eggcracker.brokered import BrokeredOperator, linux_ipc | |
| from lumi_eggcracker.brokered import operator as brokered_operator | |
| from scripts import brokered_operator_demo | |
| def canonical(value: object) -> bytes: | |
| return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode() | |
| class BrokeredLinuxProtocolTests(unittest.TestCase): | |
| def setUp(self) -> None: | |
| self.temporary = tempfile.TemporaryDirectory(dir=tempfile.gettempdir()) | |
| self.addCleanup(self.temporary.cleanup) | |
| state_directory = Path(self.temporary.name) / "state" | |
| self.operator, self.grant = BrokeredOperator._open_or_bootstrap_service(state_directory) | |
| self.workload_uid = 24680 | |
| if sys.platform == "linux" and self.workload_uid == os.geteuid(): | |
| self.workload_uid += 1 | |
| self.service_uid = 13579 | |
| if self.service_uid == self.workload_uid: | |
| self.service_uid += 1 | |
| self.service = linux_ipc.BrokeredLinuxService._for_trusted_test( | |
| self.operator, | |
| self.grant, | |
| workload_uid=self.workload_uid, | |
| workload_gid=24681, | |
| service_uid=self.service_uid, | |
| ) | |
| def request(self, operation: str, **fields: object) -> bytes: | |
| return canonical( | |
| {"op": operation, "schema_version": linux_ipc.IPC_SCHEMA, **fields} | |
| ) | |
| def response(self, raw: bytes, *, peer_uid: int | None = None) -> dict[str, object]: | |
| result = self.service.handle_packet( | |
| self.workload_uid if peer_uid is None else peer_uid, | |
| raw, | |
| ) | |
| return linux_ipc._parse_response(result) | |
| def receipt(self, response: dict[str, object]) -> dict[str, object]: | |
| receipt = response.get("receipt") | |
| self.assertIsInstance(receipt, dict) | |
| return receipt | |
| def test_strict_protocol_rejects_unknown_duplicate_bool_malformed_and_oversize(self) -> None: | |
| valid = self.request("submit", operation_id="strict-1") | |
| unknown = self.request("submit", operation_id="strict-2", target="/private/file") | |
| duplicate = valid.replace(b'"op":"submit",', b'"op":"submit","op":"submit",') | |
| bool_queue = self.request("dispatch", queue_id=True) | |
| cases = ( | |
| (unknown, "UNKNOWN_FIELDS"), | |
| (duplicate, "MALFORMED_INPUT"), | |
| (bool_queue, "MALFORMED_INPUT"), | |
| (b'{"broken":', "MALFORMED_INPUT"), | |
| (b" " * (linux_ipc.MAX_IPC_REQUEST_BYTES + 1), "MALFORMED_INPUT"), | |
| (valid.rstrip(b"\n"), "NON_CANONICAL_INPUT"), | |
| ( | |
| canonical( | |
| { | |
| "action": "increment", | |
| "capability": "caller-minted", | |
| "generation": 0, | |
| "op": "submit", | |
| "operation_id": "strict-3", | |
| "run_id": self.grant.run_id, | |
| "schema_version": linux_ipc.IPC_SCHEMA, | |
| } | |
| ), | |
| "UNKNOWN_FIELDS", | |
| ), | |
| ) | |
| for raw, code in cases: | |
| with self.subTest(code=code, size=len(raw)): | |
| result = self.response(raw) | |
| self.assertEqual("protocol", result["phase"]) | |
| self.assertEqual("DENIED", result["outcome"]) | |
| self.assertEqual(code, result["code"]) | |
| self.assertEqual(0, self.operator.world_snapshot().protected_effects) | |
| def test_peer_identity_denial_precedes_parse_and_effect(self) -> None: | |
| result = self.response(b"not-json", peer_uid=self.workload_uid + 1) | |
| self.assertEqual("PEER_IDENTITY", result["code"]) | |
| self.assertEqual(0, self.operator.world_snapshot().protected_effects) | |
| def test_service_only_construction_and_fixed_result_workflow(self) -> None: | |
| with self.assertRaises(PermissionError): | |
| linux_ipc.BrokeredLinuxService( | |
| self.operator, | |
| self.grant, | |
| Path("/tmp/unused.sock"), | |
| self.workload_uid, | |
| 24681, | |
| 0, | |
| ) | |
| self.assertFalse(hasattr(linux_ipc, "_SYNTHETIC_DATASET")) | |
| self.assertFalse(hasattr(linux_ipc, "_fixed_work_report")) | |
| admitted = self.receipt( | |
| self.response(self.request("submit", operation_id="research-batch-1")) | |
| ) | |
| self.assertEqual(("admission", "QUEUED", "ADMITTED"), (admitted["phase"], admitted["outcome"], admitted["code"])) | |
| queue_id = admitted["queue_id"] | |
| self.assertEqual("NOT_FOUND", self.response(self.request("get_result", queue_id=queue_id))["code"]) | |
| applied = self.receipt(self.response(self.request("dispatch", queue_id=queue_id))) | |
| self.assertEqual(("dispatch", "APPLIED", "EFFECT_APPLIED"), (applied["phase"], applied["outcome"], applied["code"])) | |
| self.assertTrue(applied["effect_applied"]) | |
| available = self.response(self.request("get_result", queue_id=queue_id)) | |
| self.assertEqual("AVAILABLE", available["outcome"]) | |
| self.assertTrue(linux_ipc._valid_report(available["report"])) | |
| self.assertEqual(available["report"], self.operator._result_for_run(queue_id, self.grant.run_id)) | |
| self.assertLessEqual( | |
| len(canonical(available["report"])), | |
| linux_ipc.MAX_RESULT_BYTES, | |
| ) | |
| self.assertEqual( | |
| "REPLAY", | |
| self.receipt( | |
| self.response(self.request("submit", operation_id="research-batch-1")) | |
| )["code"], | |
| ) | |
| self.assertEqual( | |
| "REPLAY", | |
| self.receipt(self.response(self.request("dispatch", queue_id=queue_id)))["code"], | |
| ) | |
| snapshot = self.operator.world_snapshot() | |
| self.assertEqual(1, snapshot.protected_effects) | |
| self.assertEqual(73, snapshot.unrelated_canary_allocation) | |
| _, reopened = BrokeredOperator.open(Path(self.temporary.name) / "state") | |
| resumed_grant = reopened._service_run_grant() | |
| reopened_service = linux_ipc.BrokeredLinuxService._for_trusted_test( | |
| reopened, | |
| resumed_grant, | |
| workload_uid=self.workload_uid, | |
| workload_gid=24681, | |
| service_uid=self.service_uid, | |
| ) | |
| after_reopen = linux_ipc._parse_response( | |
| reopened_service.handle_packet( | |
| self.workload_uid, | |
| self.request("get_result", queue_id=queue_id), | |
| ) | |
| ) | |
| self.assertEqual("AVAILABLE", after_reopen["outcome"]) | |
| self.assertEqual(1, reopened.world_snapshot().protected_effects) | |
| def test_stop_fence_stale_queue_and_single_run_survive_reopen(self) -> None: | |
| completed = self.receipt( | |
| self.response(self.request("submit", operation_id="applied-before-stop")) | |
| ) | |
| completed_queue_id = completed["queue_id"] | |
| self.receipt(self.response(self.request("dispatch", queue_id=completed_queue_id))) | |
| applied_report = self.response(self.request("get_result", queue_id=completed_queue_id))["report"] | |
| queued = self.receipt( | |
| self.response(self.request("submit", operation_id="queued-before-stop")) | |
| ) | |
| queue_id = queued["queue_id"] | |
| stopped = self.service.request_stop() | |
| self.assertEqual(("REVOKED", 1, "UNSUPPORTED"), (stopped.outcome, stopped.generation, stopped.process_stop)) | |
| stale = self.receipt(self.response(self.request("dispatch", queue_id=queue_id))) | |
| self.assertEqual("STALE_GENERATION", stale["code"]) | |
| self.assertEqual( | |
| "CAPABILITY_REVOKED", | |
| self.receipt( | |
| self.response(self.request("submit", operation_id="post-stop")) | |
| )["code"], | |
| ) | |
| self.assertEqual("NOT_FOUND", self.response(self.request("get_result", queue_id=queue_id))["code"]) | |
| _, reopened = BrokeredOperator.open(Path(self.temporary.name) / "state") | |
| resumed_grant = reopened._service_run_grant() | |
| self.assertEqual(self.grant.run_id, resumed_grant.run_id) | |
| reopened_service = linux_ipc.BrokeredLinuxService._for_trusted_test( | |
| reopened, | |
| resumed_grant, | |
| workload_uid=self.workload_uid, | |
| workload_gid=24681, | |
| service_uid=self.service_uid, | |
| ) | |
| self.assertEqual( | |
| "ALREADY_FINAL", | |
| self.receipt( | |
| linux_ipc._parse_response( | |
| reopened_service.handle_packet( | |
| self.workload_uid, | |
| self.request("dispatch", queue_id=queue_id), | |
| ) | |
| ) | |
| )["code"], | |
| ) | |
| self.assertEqual( | |
| "CAPABILITY_REVOKED", | |
| self.receipt( | |
| linux_ipc._parse_response( | |
| reopened_service.handle_packet( | |
| self.workload_uid, | |
| self.request("submit", operation_id="after-reopen"), | |
| ) | |
| ) | |
| )["code"], | |
| ) | |
| reopened_result = linux_ipc._parse_response( | |
| reopened_service.handle_packet( | |
| self.workload_uid, | |
| self.request("get_result", queue_id=completed_queue_id), | |
| ) | |
| ) | |
| self.assertEqual("AVAILABLE", reopened_result["outcome"]) | |
| self.assertEqual(applied_report, reopened_result["report"]) | |
| self.assertEqual(1, reopened.world_snapshot().protected_effects) | |
| self.assertEqual(73, reopened.world_snapshot().unrelated_canary_allocation) | |
| def test_direct_bypass_requests_are_denied_before_effect(self) -> None: | |
| for operation in ("get_dataset", "trusted_stop", "read_state"): | |
| with self.subTest(operation=operation): | |
| response = self.response(self.request(operation, path="/private/state")) | |
| self.assertEqual("DENIED", response["outcome"]) | |
| self.assertIn(response["code"], {"MALFORMED_INPUT", "UNKNOWN_FIELDS"}) | |
| self.assertEqual(0, self.operator.world_snapshot().protected_effects) | |
| class BrokeredLinuxOneShotCLITests(unittest.TestCase): | |
| queue_id = "a" * 32 | |
| other_queue_id = "b" * 32 | |
| def receipt( | |
| *, | |
| phase: str, | |
| outcome: str, | |
| code: str, | |
| effect_applied: bool, | |
| queue_id: str | None = None, | |
| ) -> dict[str, object]: | |
| value: dict[str, object] = { | |
| "code": code, | |
| "effect_applied": effect_applied, | |
| "outcome": outcome, | |
| "phase": phase, | |
| } | |
| if queue_id is not None: | |
| value["queue_id"] = queue_id | |
| return {"receipt": value, "schema_version": linux_ipc.IPC_SCHEMA} | |
| def result(*, queue_id: str | None = None) -> dict[str, object]: | |
| value: dict[str, object] = { | |
| "outcome": "AVAILABLE", | |
| "phase": "result", | |
| "report": { | |
| "accepted": 2, | |
| "dataset_id": "synthetic.research-batch.v1", | |
| "dataset_sha256": "c" * 64, | |
| "records_checked": 4, | |
| "review": 2, | |
| "units_total": 32, | |
| }, | |
| "schema_version": linux_ipc.IPC_SCHEMA, | |
| } | |
| if queue_id is not None: | |
| value["queue_id"] = queue_id | |
| return value | |
| def invoke( | |
| self, | |
| responses: dict[str, object], | |
| ) -> tuple[int, list[tuple[object, ...]], str, str, mock.Mock]: | |
| calls: list[tuple[object, ...]] = [] | |
| class FakeClient: | |
| def _call(self, stage: str, *arguments: object) -> object: | |
| calls.append((stage, *arguments)) | |
| response = responses[stage] | |
| if isinstance(response, Exception): | |
| raise response | |
| return response | |
| def submit(self, operation_id: str) -> object: | |
| return self._call("submit", operation_id) | |
| def dispatch(self, queue_id: str) -> object: | |
| return self._call("dispatch", queue_id) | |
| def get_result(self, queue_id: str) -> object: | |
| return self._call("get_result", queue_id) | |
| fake_client = FakeClient() | |
| standard_output = io.StringIO() | |
| standard_error = io.StringIO() | |
| arguments = [ | |
| "--linux-ipc", | |
| "run", | |
| "--socket", | |
| "/tmp/fake-broker.sock", | |
| "--service-uid", | |
| "1200", | |
| "--operation-id", | |
| "cli-run-1", | |
| ] | |
| with ( | |
| mock.patch.object(sys, "platform", "linux"), | |
| mock.patch.object(brokered_operator_demo.os, "geteuid", return_value=1201, create=True), | |
| mock.patch.object(linux_ipc, "BrokeredLinuxClient", return_value=fake_client) as constructor, | |
| contextlib.redirect_stdout(standard_output), | |
| contextlib.redirect_stderr(standard_error), | |
| ): | |
| status = brokered_operator_demo.main(arguments) | |
| return status, calls, standard_output.getvalue(), standard_error.getvalue(), constructor | |
| def successful_responses(self) -> dict[str, object]: | |
| return { | |
| "submit": self.receipt( | |
| phase="admission", | |
| outcome="QUEUED", | |
| code="ADMITTED", | |
| effect_applied=False, | |
| queue_id=self.queue_id, | |
| ), | |
| "dispatch": self.receipt( | |
| phase="dispatch", | |
| outcome="APPLIED", | |
| code="EFFECT_APPLIED", | |
| effect_applied=True, | |
| queue_id=self.queue_id, | |
| ), | |
| "get_result": self.result(queue_id=self.queue_id), | |
| } | |
| def test_run_calls_each_primitive_once_and_emits_one_combined_outcome(self) -> None: | |
| responses = self.successful_responses() | |
| status, calls, standard_output, standard_error, constructor = self.invoke(responses) | |
| self.assertEqual(0, status) | |
| self.assertEqual( | |
| [ | |
| ("submit", "cli-run-1"), | |
| ("dispatch", self.queue_id), | |
| ("get_result", self.queue_id), | |
| ], | |
| calls, | |
| ) | |
| constructor.assert_called_once_with("/tmp/fake-broker.sock", service_uid=1200) | |
| self.assertEqual("", standard_error) | |
| self.assertEqual(1, len(standard_output.splitlines())) | |
| self.assertEqual( | |
| {"run": { | |
| "admission": responses["submit"], | |
| "dispatch": responses["dispatch"], | |
| "result": responses["get_result"], | |
| }}, | |
| json.loads(standard_output), | |
| ) | |
| def test_run_stops_after_invalid_admission_or_dispatch(self) -> None: | |
| denied_admission = self.receipt( | |
| phase="admission", | |
| outcome="DENIED", | |
| code="CAPABILITY_REVOKED", | |
| effect_applied=False, | |
| queue_id=self.queue_id, | |
| ) | |
| malformed_admission = {"schema_version": linux_ipc.IPC_SCHEMA} | |
| missing_admission_id = self.receipt( | |
| phase="admission", | |
| outcome="QUEUED", | |
| code="ADMITTED", | |
| effect_applied=False, | |
| ) | |
| mismatched_dispatch = self.receipt( | |
| phase="dispatch", | |
| outcome="APPLIED", | |
| code="EFFECT_APPLIED", | |
| effect_applied=True, | |
| queue_id=self.other_queue_id, | |
| ) | |
| missing_dispatch_id = self.receipt( | |
| phase="dispatch", | |
| outcome="APPLIED", | |
| code="EFFECT_APPLIED", | |
| effect_applied=True, | |
| ) | |
| denied_dispatch = self.receipt( | |
| phase="dispatch", | |
| outcome="DENIED", | |
| code="STALE_GENERATION", | |
| effect_applied=False, | |
| queue_id=self.queue_id, | |
| ) | |
| for admission, dispatch, expected_calls in ( | |
| (denied_admission, self.successful_responses()["dispatch"], [("submit", "cli-run-1")]), | |
| (malformed_admission, self.successful_responses()["dispatch"], [("submit", "cli-run-1")]), | |
| (missing_admission_id, self.successful_responses()["dispatch"], [("submit", "cli-run-1")]), | |
| ( | |
| self.successful_responses()["submit"], | |
| mismatched_dispatch, | |
| [("submit", "cli-run-1"), ("dispatch", self.queue_id)], | |
| ), | |
| ( | |
| self.successful_responses()["submit"], | |
| missing_dispatch_id, | |
| [("submit", "cli-run-1"), ("dispatch", self.queue_id)], | |
| ), | |
| ( | |
| self.successful_responses()["submit"], | |
| denied_dispatch, | |
| [("submit", "cli-run-1"), ("dispatch", self.queue_id)], | |
| ), | |
| ): | |
| with self.subTest(admission=admission, dispatch=dispatch): | |
| status, calls, standard_output, standard_error, _ = self.invoke( | |
| { | |
| "submit": admission, | |
| "dispatch": dispatch, | |
| "get_result": self.result(queue_id=self.queue_id), | |
| } | |
| ) | |
| self.assertEqual(1, status) | |
| self.assertEqual(expected_calls, calls) | |
| self.assertEqual("", standard_output) | |
| self.assertIn("failed closed", standard_error) | |
| def test_run_rejects_bad_result_and_stops_on_exceptions(self) -> None: | |
| mismatched_result = self.result(queue_id=self.other_queue_id) | |
| missing_result_id = self.result() | |
| bad_result = self.result(queue_id=self.queue_id) | |
| bad_result["report"] = {"unbounded": "x" * (linux_ipc.MAX_RESULT_BYTES + 1)} | |
| denied_result = { | |
| "code": "NOT_FOUND", | |
| "outcome": "DENIED", | |
| "phase": "result", | |
| "queue_id": self.queue_id, | |
| "schema_version": linux_ipc.IPC_SCHEMA, | |
| } | |
| cases = ( | |
| ( | |
| {**self.successful_responses(), "get_result": mismatched_result}, | |
| [("submit", "cli-run-1"), ("dispatch", self.queue_id), ("get_result", self.queue_id)], | |
| ), | |
| ( | |
| {**self.successful_responses(), "get_result": missing_result_id}, | |
| [("submit", "cli-run-1"), ("dispatch", self.queue_id), ("get_result", self.queue_id)], | |
| ), | |
| ( | |
| {**self.successful_responses(), "get_result": denied_result}, | |
| [("submit", "cli-run-1"), ("dispatch", self.queue_id), ("get_result", self.queue_id)], | |
| ), | |
| ( | |
| {**self.successful_responses(), "get_result": bad_result}, | |
| [("submit", "cli-run-1"), ("dispatch", self.queue_id), ("get_result", self.queue_id)], | |
| ), | |
| ( | |
| {**self.successful_responses(), "submit": RuntimeError("submit failed")}, | |
| [("submit", "cli-run-1")], | |
| ), | |
| ( | |
| {**self.successful_responses(), "dispatch": RuntimeError("dispatch failed")}, | |
| [("submit", "cli-run-1"), ("dispatch", self.queue_id)], | |
| ), | |
| ( | |
| {**self.successful_responses(), "get_result": RuntimeError("result failed")}, | |
| [("submit", "cli-run-1"), ("dispatch", self.queue_id), ("get_result", self.queue_id)], | |
| ), | |
| ) | |
| for responses, expected_calls in cases: | |
| with self.subTest(expected_calls=expected_calls, responses=responses): | |
| status, calls, standard_output, standard_error, _ = self.invoke(responses) | |
| self.assertEqual(1, status) | |
| self.assertEqual(expected_calls, calls) | |
| self.assertEqual("", standard_output) | |
| self.assertIn("failed closed", standard_error) | |
| def test_run_missing_arguments_fail_before_client_construction(self) -> None: | |
| cases = ( | |
| ["--linux-ipc", "run", "--socket", "/tmp/fake.sock", "--service-uid", "1200"], | |
| ["--linux-ipc", "run", "--service-uid", "1200", "--operation-id", "cli-run-1"], | |
| ["--linux-ipc", "run", "--socket", "/tmp/fake.sock", "--operation-id", "cli-run-1"], | |
| ) | |
| with mock.patch.object(sys, "platform", "linux"), mock.patch.object( | |
| brokered_operator_demo.os, "geteuid", return_value=1201, create=True | |
| ): | |
| for arguments in cases: | |
| with self.subTest(arguments=arguments), mock.patch.object( | |
| linux_ipc, | |
| "BrokeredLinuxClient", | |
| side_effect=AssertionError("client must not be constructed"), | |
| ) as constructor: | |
| with self.assertRaises(SystemExit): | |
| brokered_operator_demo.main(arguments) | |
| constructor.assert_not_called() | |
| def test_run_client_construction_exception_returns_nonzero(self) -> None: | |
| arguments = [ | |
| "--linux-ipc", | |
| "run", | |
| "--socket", | |
| "/tmp/fake-broker.sock", | |
| "--service-uid", | |
| "1200", | |
| "--operation-id", | |
| "cli-run-1", | |
| ] | |
| standard_output = io.StringIO() | |
| standard_error = io.StringIO() | |
| with ( | |
| mock.patch.object(sys, "platform", "linux"), | |
| mock.patch.object(brokered_operator_demo.os, "geteuid", return_value=1201, create=True), | |
| mock.patch.object( | |
| linux_ipc, | |
| "BrokeredLinuxClient", | |
| side_effect=RuntimeError("client setup failed"), | |
| ) as constructor, | |
| contextlib.redirect_stdout(standard_output), | |
| contextlib.redirect_stderr(standard_error), | |
| ): | |
| status = brokered_operator_demo.main(arguments) | |
| self.assertEqual(1, status) | |
| constructor.assert_called_once_with("/tmp/fake-broker.sock", service_uid=1200) | |
| self.assertEqual("", standard_output.getvalue()) | |
| self.assertIn("failed closed", standard_error.getvalue()) | |
| def test_run_failure_before_valid_admission_never_exposes_a_queue_id(self) -> None: | |
| denied_admission = self.receipt( | |
| phase="admission", | |
| outcome="DENIED", | |
| code="CAPABILITY_REVOKED", | |
| effect_applied=False, | |
| queue_id=self.other_queue_id, | |
| ) | |
| malformed_admission = self.receipt( | |
| phase="admission", | |
| outcome="QUEUED", | |
| code="ADMITTED", | |
| effect_applied=False, | |
| queue_id="untrusted-queue-value", | |
| ) | |
| for stage, response, untrusted_id in ( | |
| ("submit", RuntimeError("untrusted submit detail"), None), | |
| ("admission_validation", denied_admission, self.other_queue_id), | |
| ("admission_validation", malformed_admission, "untrusted-queue-value"), | |
| ): | |
| with self.subTest(stage=stage): | |
| responses = self.successful_responses() | |
| responses["submit"] = response | |
| status, calls, standard_output, standard_error, _ = self.invoke(responses) | |
| failure = json.loads(standard_error)["run_failure"] | |
| self.assertEqual(1, status) | |
| self.assertEqual([("submit", "cli-run-1")], calls) | |
| self.assertEqual("", standard_output) | |
| self.assertEqual(stage, failure["failed_stage"]) | |
| self.assertEqual("UNKNOWN", failure["effect_status"]) | |
| self.assertNotIn("queue_id", failure) | |
| if untrusted_id is not None: | |
| self.assertNotIn(untrusted_id, standard_error) | |
| self.assertNotIn("untrusted submit detail", standard_error) | |
| self.assertLessEqual(len(standard_error.encode("utf-8")), 256) | |
| arguments = [ | |
| "--linux-ipc", | |
| "run", | |
| "--socket", | |
| "/tmp/fake-broker.sock", | |
| "--service-uid", | |
| "1200", | |
| "--operation-id", | |
| "cli-run-1", | |
| ] | |
| standard_output = io.StringIO() | |
| standard_error = io.StringIO() | |
| with ( | |
| mock.patch.object(sys, "platform", "linux"), | |
| mock.patch.object(brokered_operator_demo.os, "geteuid", return_value=1201, create=True), | |
| mock.patch.object( | |
| linux_ipc, | |
| "BrokeredLinuxClient", | |
| side_effect=RuntimeError("untrusted setup detail"), | |
| ), | |
| contextlib.redirect_stdout(standard_output), | |
| contextlib.redirect_stderr(standard_error), | |
| ): | |
| status = brokered_operator_demo.main(arguments) | |
| failure = json.loads(standard_error.getvalue())["run_failure"] | |
| self.assertEqual(1, status) | |
| self.assertEqual("client_setup", failure["failed_stage"]) | |
| self.assertEqual("UNKNOWN", failure["effect_status"]) | |
| self.assertNotIn("queue_id", failure) | |
| self.assertNotIn("untrusted setup detail", standard_error.getvalue()) | |
| self.assertLessEqual(len(standard_error.getvalue().encode("utf-8")), 256) | |
| self.assertEqual("", standard_output.getvalue()) | |
| def test_run_failure_before_valid_dispatch_reports_validated_id_and_unknown_effect(self) -> None: | |
| mismatched_dispatch = self.receipt( | |
| phase="dispatch", | |
| outcome="APPLIED", | |
| code="EFFECT_APPLIED", | |
| effect_applied=True, | |
| queue_id=self.other_queue_id, | |
| ) | |
| denied_dispatch = self.receipt( | |
| phase="dispatch", | |
| outcome="DENIED", | |
| code="STALE_GENERATION", | |
| effect_applied=False, | |
| queue_id=self.queue_id, | |
| ) | |
| cases = ( | |
| ("dispatch", RuntimeError("untrusted dispatch detail")), | |
| ("dispatch_validation", mismatched_dispatch), | |
| ("dispatch_validation", denied_dispatch), | |
| ) | |
| for expected_stage, dispatch_response in cases: | |
| with self.subTest(expected_stage=expected_stage): | |
| responses = self.successful_responses() | |
| responses["dispatch"] = dispatch_response | |
| status, calls, standard_output, standard_error, _ = self.invoke(responses) | |
| failure = json.loads(standard_error)["run_failure"] | |
| self.assertEqual(1, status) | |
| self.assertEqual( | |
| [("submit", "cli-run-1"), ("dispatch", self.queue_id)], | |
| calls, | |
| ) | |
| self.assertEqual("", standard_output) | |
| self.assertEqual(expected_stage, failure["failed_stage"]) | |
| self.assertEqual("admission", failure["last_validated_stage"]) | |
| self.assertEqual(self.queue_id, failure["queue_id"]) | |
| self.assertEqual("UNKNOWN", failure["effect_status"]) | |
| self.assertNotIn(self.other_queue_id, standard_error) | |
| self.assertNotIn("untrusted dispatch detail", standard_error) | |
| self.assertLessEqual(len(standard_error.encode("utf-8")), 256) | |
| def test_run_failure_after_validated_applied_dispatch_confirms_effect(self) -> None: | |
| responses = self.successful_responses() | |
| responses["get_result"] = RuntimeError("untrusted result detail") | |
| status, calls, standard_output, standard_error, _ = self.invoke(responses) | |
| failure = json.loads(standard_error)["run_failure"] | |
| self.assertEqual(1, status) | |
| self.assertEqual( | |
| [ | |
| ("submit", "cli-run-1"), | |
| ("dispatch", self.queue_id), | |
| ("get_result", self.queue_id), | |
| ], | |
| calls, | |
| ) | |
| self.assertEqual("", standard_output) | |
| self.assertEqual("result", failure["failed_stage"]) | |
| self.assertEqual("dispatch", failure["last_validated_stage"]) | |
| self.assertEqual(self.queue_id, failure["queue_id"]) | |
| self.assertEqual("CONFIRMED_APPLIED", failure["effect_status"]) | |
| self.assertNotIn("untrusted result detail", standard_error) | |
| self.assertLessEqual(len(standard_error.encode("utf-8")), 256) | |
| class LinuxIdentityAndPathGuardTests(unittest.TestCase): | |
| def test_service_and_workload_root_identity_configurations_are_refused(self) -> None: | |
| valid = {"service_gid": 13579, "service_groups": (13579,)} | |
| cases = ( | |
| (0, 24680, 24681, valid), | |
| (13579, 0, 24681, valid), | |
| (13579, 24680, 0, valid), | |
| (13579, 24680, 24681, {"service_gid": 0, "service_groups": ()}), | |
| (13579, 24680, 24681, {"service_gid": 13579, "service_groups": (0,)}), | |
| (13579, 13579, 24681, valid), | |
| ) | |
| for service_uid, workload_uid, workload_gid, keyword_args in cases: | |
| with ( | |
| self.subTest(service_uid=service_uid, workload_uid=workload_uid, workload_gid=workload_gid), | |
| self.assertRaises(linux_ipc.BrokeredIPCError), | |
| ): | |
| linux_ipc._validate_identity_config( | |
| service_uid, | |
| workload_uid, | |
| workload_gid, | |
| **keyword_args, | |
| ) | |
| def test_exact_private_state_modes_reject_workload_owner_and_special_bits(self) -> None: | |
| directory_mode = stat.S_IFDIR | 0o700 | |
| file_mode = stat.S_IFREG | 0o600 | |
| directory = SimpleNamespace(st_uid=13579, st_mode=directory_mode, st_nlink=2) | |
| file = SimpleNamespace(st_uid=13579, st_mode=file_mode, st_nlink=1) | |
| self.assertTrue(brokered_operator._private_state_metadata_valid(directory, 13579, directory=True)) | |
| self.assertTrue(brokered_operator._private_state_metadata_valid(file, 13579, directory=False)) | |
| self.assertFalse(brokered_operator._private_state_metadata_valid(directory, 24680, directory=True)) | |
| self.assertFalse(brokered_operator._private_state_metadata_valid(file, 24680, directory=False)) | |
| special_directory = SimpleNamespace(st_uid=13579, st_mode=stat.S_IFDIR | 0o4700, st_nlink=2) | |
| loose_file = SimpleNamespace(st_uid=13579, st_mode=stat.S_IFREG | 0o700, st_nlink=1) | |
| self.assertFalse(brokered_operator._private_state_metadata_valid(special_directory, 13579, directory=True)) | |
| self.assertFalse(brokered_operator._private_state_metadata_valid(loose_file, 13579, directory=False)) | |
| private_operator = object.__new__(BrokeredOperator) | |
| private_operator._directory = Path("/private") | |
| def private_metadata(path: Path) -> SimpleNamespace: | |
| return directory if path == private_operator._directory else file | |
| with mock.patch.object(Path, "lstat", autospec=True, side_effect=private_metadata): | |
| private_operator._assert_service_state_owner(13579) | |
| with self.assertRaises(brokered_operator.BrokeredStoreError): | |
| private_operator._assert_service_state_owner(24680) | |
| def test_pinned_directory_and_socket_identity_replacements_fail_closed(self) -> None: | |
| service = object.__new__(linux_ipc.BrokeredLinuxService) | |
| service._socket_path = Path("/test-only/broker.sock") | |
| service._service_uid = 13579 | |
| service._workload_gid = 24681 | |
| service._socket_directory_fd = 42 | |
| service._socket_directory_identity = (1, 2) | |
| service._socket_identity = (3, 4) | |
| parent = service._socket_path.parent | |
| directory_meta = SimpleNamespace( | |
| st_dev=1, | |
| st_ino=2, | |
| st_mode=stat.S_IFDIR | 0o710, | |
| st_uid=13579, | |
| st_gid=24681, | |
| ) | |
| replacement_directory = SimpleNamespace( | |
| st_dev=1, | |
| st_ino=9, | |
| st_mode=stat.S_IFDIR | 0o710, | |
| st_uid=13579, | |
| st_gid=24681, | |
| ) | |
| with ( | |
| mock.patch.object(linux_ipc.os, "fstat", return_value=replacement_directory), | |
| mock.patch.object(Path, "lstat", return_value=directory_meta), | |
| mock.patch.object(Path, "resolve", return_value=parent), | |
| self.assertRaises(linux_ipc.BrokeredIPCError), | |
| ): | |
| service._assert_pinned_socket_directory() | |
| socket_path_replacement = SimpleNamespace( | |
| st_dev=3, | |
| st_ino=5, | |
| st_mode=stat.S_IFSOCK | 0o660, | |
| st_uid=13579, | |
| st_gid=24681, | |
| ) | |
| def replaced_lstat(path: Path) -> SimpleNamespace: | |
| return directory_meta if path == parent else socket_path_replacement | |
| with ( | |
| mock.patch.object(linux_ipc.os, "fstat", return_value=directory_meta), | |
| mock.patch.object(Path, "lstat", autospec=True, side_effect=replaced_lstat), | |
| mock.patch.object(Path, "resolve", return_value=parent), | |
| mock.patch.object( | |
| linux_ipc.os, | |
| "stat", | |
| return_value=SimpleNamespace( | |
| st_dev=3, | |
| st_ino=4, | |
| st_mode=stat.S_IFSOCK | 0o660, | |
| st_uid=13579, | |
| st_gid=24681, | |
| ), | |
| ), | |
| self.assertRaises(linux_ipc.BrokeredIPCError), | |
| ): | |
| service._assert_bound_socket_identity() | |
| def test_client_identity_guard_rejects_root_and_root_group(self) -> None: | |
| for service_uid, workload_uid, workload_gid, groups in ( | |
| (0, 24680, 24681, {24681}), | |
| (13579, 0, 24681, {24681}), | |
| (13579, 24680, 0, {0}), | |
| (13579, 24680, 24681, {0, 24681}), | |
| (13579, 13579, 24681, {24681}), | |
| ): | |
| with ( | |
| self.subTest(service_uid=service_uid, workload_uid=workload_uid, groups=groups), | |
| self.assertRaises(linux_ipc.BrokeredIPCError), | |
| ): | |
| linux_ipc._validate_client_identity(service_uid, workload_uid, workload_gid, groups) | |
| class LinuxPeerCredentialTests(unittest.TestCase): | |
| def test_linux_transport_is_explicitly_separate_from_portable_core(self) -> None: | |
| self.assertTrue(hasattr(socket, "SO_PEERCRED")) | |
| self.assertTrue(hasattr(socket, "SOCK_SEQPACKET")) | |
| client, server = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) | |
| self.addCleanup(client.close) | |
| self.addCleanup(server.close) | |
| credentials = client.getsockopt( | |
| socket.SOL_SOCKET, | |
| socket.SO_PEERCRED, | |
| struct.calcsize("3i"), | |
| ) | |
| _, peer_uid, _ = struct.unpack("3i", credentials) | |
| self.assertEqual(os.geteuid(), peer_uid) | |
| if __name__ == "__main__": | |
| unittest.main() | |