188 lines
7.5 KiB
Python
188 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import os
|
|
from pathlib import Path
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
|
|
MODULE = Path(__file__).resolve().parents[1] / "automation" / "health_probe_common.py"
|
|
SPEC = importlib.util.spec_from_file_location("health_probe_common", MODULE)
|
|
assert SPEC and SPEC.loader
|
|
common = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(common)
|
|
sys.modules["health_probe_common"] = common
|
|
|
|
CENTRAL_MODULE = Path(__file__).resolve().parents[1] / "automation" / "central_gpt_health_probe.py"
|
|
CENTRAL_SPEC = importlib.util.spec_from_file_location("central_gpt_health_probe", CENTRAL_MODULE)
|
|
assert CENTRAL_SPEC and CENTRAL_SPEC.loader
|
|
central = importlib.util.module_from_spec(CENTRAL_SPEC)
|
|
CENTRAL_SPEC.loader.exec_module(central)
|
|
|
|
CONTRACT = "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc"
|
|
|
|
|
|
class HealthProbeSecurityTests(unittest.TestCase):
|
|
def test_existing_pending_event_is_rechecked_without_duplicate_post(self):
|
|
config = {
|
|
"gateway_event_url": "http://127.0.0.1:18920/v1/orchestrator/events",
|
|
"gateway_token_file": "/run/aios-orchestrator-gateway/token",
|
|
"project_key": "v2r1-personal-weixin",
|
|
"contract_commit": CONTRACT,
|
|
"central_timeout_seconds": 1,
|
|
"poll_interval_seconds": 0.01,
|
|
"central_receipt_path": "/run/aios-health/central-gpt.json",
|
|
}
|
|
event_id = "health-central-" + "a" * 32
|
|
with (
|
|
mock.patch.object(central, "load_config", return_value=config),
|
|
mock.patch.object(central, "read_token", return_value="not-a-real-token"),
|
|
mock.patch.object(central, "_load_pending", return_value=(event_id, 3)),
|
|
mock.patch.object(
|
|
central,
|
|
"request_json",
|
|
return_value=(200, {"status": "TRIGGER_FAILED"}),
|
|
) as request,
|
|
):
|
|
with self.assertRaisesRegex(common.ProbeError, "central_decision_not_proven"):
|
|
central.run(Path("/not/read"))
|
|
self.assertTrue(request.call_args_list)
|
|
self.assertTrue(all(call.kwargs.get("method", "GET") == "GET" for call in request.call_args_list))
|
|
|
|
def test_central_pending_marker_is_metadata_only_and_reusable(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
path = Path(temp) / "central-gpt.pending.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"contract_commit": CONTRACT,
|
|
"event_id": "health-central-" + "a" * 32,
|
|
"state_version": 7,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
os.chmod(path, 0o640)
|
|
self.assertEqual(
|
|
central._load_pending(path, CONTRACT, owners={os.geteuid()}),
|
|
("health-central-" + "a" * 32, 7),
|
|
)
|
|
|
|
def test_central_pending_marker_wrong_contract_fails_closed(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
path = Path(temp) / "central-gpt.pending.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"contract_commit": "0" * 40,
|
|
"event_id": "health-central-" + "a" * 32,
|
|
"state_version": 0,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
os.chmod(path, 0o640)
|
|
with self.assertRaises(common.ProbeError):
|
|
central._load_pending(path, CONTRACT, owners={os.geteuid()})
|
|
|
|
def test_central_pending_marker_symlink_fails_closed(self):
|
|
if not hasattr(os, "symlink"):
|
|
self.skipTest("symlink unavailable")
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
target = Path(temp) / "target"
|
|
target.write_text("{}", encoding="utf-8")
|
|
link = Path(temp) / "central-gpt.pending.json"
|
|
try:
|
|
link.symlink_to(target)
|
|
except OSError:
|
|
self.skipTest("symlink privilege unavailable")
|
|
with self.assertRaises(common.ProbeError):
|
|
central._load_pending(link, CONTRACT, owners={os.geteuid()})
|
|
|
|
def test_probe_error_reason_is_metadata_only(self):
|
|
self.assertEqual(central._safe_reason(common.ProbeError("gateway_unavailable")), "gateway_unavailable")
|
|
self.assertEqual(central._safe_reason(ValueError("secret value leaked!")), "probe_error")
|
|
|
|
def test_availability_receipts_are_readable_only_by_quorum_group(self):
|
|
source = MODULE.read_text(encoding="utf-8")
|
|
self.assertIn("os.fchown(fd, 0, info.st_gid)", source)
|
|
self.assertIn("os.fchmod(fd, 0o640)", source)
|
|
self.assertNotIn("os.fchmod(fd, 0o644)", source)
|
|
|
|
def test_codex_probe_commands_are_group_readable_by_watcher(self):
|
|
unit = (
|
|
Path(__file__).resolve().parents[1]
|
|
/ "systemd"
|
|
/ "aios-codex-executor-health-probe.service"
|
|
).read_text(encoding="utf-8")
|
|
self.assertIn("User=root\n", unit)
|
|
self.assertIn("Group=aios-orchestrator\n", unit)
|
|
self.assertIn("UMask=0027\n", unit)
|
|
self.assertIn(
|
|
"ReadOnlyPaths=/etc/aios-automation-quorum "
|
|
"/etc/aios-orchestrator-gateway /var/log/aios-codex-watcher\n",
|
|
unit,
|
|
)
|
|
self.assertIn(
|
|
"ReadWritePaths=/run/aios-health "
|
|
"/data/aios-gitops/workspace/agents/commands/active "
|
|
"/var/lib/aios-codex-watcher\n",
|
|
unit,
|
|
)
|
|
self.assertNotIn("Group=root\n", unit)
|
|
|
|
def test_gateway_endpoint_is_exact_loopback(self):
|
|
self.assertEqual(
|
|
common.gateway_base("http://127.0.0.1:18920/v1/orchestrator/events"),
|
|
"http://127.0.0.1:18920/v1/orchestrator",
|
|
)
|
|
for url in (
|
|
"http://127.0.0.1:8765/v1/orchestrator/events",
|
|
"http://localhost:18920/v1/orchestrator/events",
|
|
"https://127.0.0.1:18920/v1/orchestrator/events",
|
|
"http://127.0.0.1:18920/v1/orchestrator/events?token=x",
|
|
):
|
|
with self.assertRaises(common.ProbeError):
|
|
common.gateway_base(url)
|
|
|
|
def test_token_newline_is_rejected(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
path = Path(temp) / "token"
|
|
path.write_bytes(b"not-a-secret-value\n")
|
|
os.chmod(path, 0o600)
|
|
with self.assertRaises(common.ProbeError):
|
|
common.read_token(path)
|
|
|
|
def test_token_symlink_is_rejected(self):
|
|
if not hasattr(os, "symlink"):
|
|
self.skipTest("symlink unavailable")
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
target = Path(temp) / "target"
|
|
target.write_bytes(b"not-a-secret-value")
|
|
os.chmod(target, 0o600)
|
|
link = Path(temp) / "link"
|
|
try:
|
|
link.symlink_to(target)
|
|
except OSError:
|
|
self.skipTest("symlink privilege unavailable")
|
|
with self.assertRaises(common.ProbeError):
|
|
common.read_token(link)
|
|
|
|
def test_token_wide_mode_is_rejected(self):
|
|
if os.name != "posix":
|
|
self.skipTest("POSIX mode semantics required")
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
path = Path(temp) / "token"
|
|
path.write_bytes(b"not-a-secret-value")
|
|
os.chmod(path, 0o644)
|
|
with self.assertRaises(common.ProbeError):
|
|
common.read_token(path)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|