175 lines
7.6 KiB
Python
175 lines
7.6 KiB
Python
import os
|
|
import sys
|
|
import tempfile
|
|
import subprocess
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "active-runtime" / "app"))
|
|
import aios_isolated_writer_policy as policy
|
|
|
|
|
|
class IsolatedWriterPolicyTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.saved = dict(os.environ)
|
|
os.environ["AIOS_GITOPS_ISOLATED_WRITER_ENFORCEMENT"] = "required"
|
|
os.environ["AIOS_GITOPS_ALLOWED_BRANCH"] = "patch/aios-agent-task-control-plane-20260724"
|
|
os.environ["AIOS_GITOPS_ALLOWED_BASE_OID"] = "a8e97868976b0dce17b7cb8487eadc1d9c687622"
|
|
os.environ["AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED"] = "2"
|
|
os.environ["AIOS_GITOPS_SERVICE_AUDIENCE"] = "test-isolated-writer.service"
|
|
os.environ["AIOS_GITOPS_TOKEN_REGISTRY_V2"] = "/tmp/test-writer-token-registry.sqlite3"
|
|
os.environ["AIOS_GITOPS_WRITER_OPERATION_LEDGER"] = "/tmp/test-writer-operations.sqlite3"
|
|
|
|
def tearDown(self):
|
|
os.environ.clear()
|
|
os.environ.update(self.saved)
|
|
|
|
def test_only_expected_branch_is_accepted(self):
|
|
policy.validate_branch_request({
|
|
"branch": os.environ["AIOS_GITOPS_ALLOWED_BRANCH"],
|
|
"base": "main",
|
|
})
|
|
for branch in ("patch/other", "main", ""):
|
|
with self.subTest(branch=branch), self.assertRaises(PermissionError):
|
|
policy.validate_branch_request({"branch": branch, "base": "main"})
|
|
|
|
def test_write_paths_are_scoped_and_canonical(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
workspace = Path(root)
|
|
self.assertEqual(
|
|
policy.validate_path("active-runtime/app/x.py", workspace),
|
|
"active-runtime/app/x.py",
|
|
)
|
|
self.assertEqual(policy.validate_path("docs/x.md", workspace), "docs/x.md")
|
|
self.assertEqual(
|
|
policy.validate_path("scripts/task-plane/run.sh", workspace),
|
|
"scripts/task-plane/run.sh",
|
|
)
|
|
self.assertEqual(
|
|
policy.validate_path("reports/task-plane.json", workspace),
|
|
"reports/task-plane.json",
|
|
)
|
|
rejected = (
|
|
"",
|
|
"../docs/x.md",
|
|
"/docs/x.md",
|
|
"docs/../x.md",
|
|
"%2e%2e/docs/x.md",
|
|
"%2Fdocs/x.md",
|
|
r"C:\temp\x",
|
|
r"\\server\share\x",
|
|
"./docs/x.md",
|
|
"docs//x.md",
|
|
"docs/./x.md",
|
|
"docs/x.md\x00tail",
|
|
"baseline/current.json",
|
|
)
|
|
for value in rejected:
|
|
with self.subTest(value=value), self.assertRaises(PermissionError):
|
|
policy.validate_path(value, workspace)
|
|
|
|
@unittest.skipIf(os.name == "nt", "symlink creation requires Windows privilege")
|
|
def test_external_symlink_is_rejected(self):
|
|
with tempfile.TemporaryDirectory() as root, tempfile.TemporaryDirectory() as outside:
|
|
workspace = Path(root)
|
|
(workspace / "docs").symlink_to(Path(outside), target_is_directory=True)
|
|
with self.assertRaises(PermissionError):
|
|
policy.validate_path("docs/escape.md", workspace)
|
|
|
|
def test_force_and_forbidden_endpoints_are_rejected(self):
|
|
with self.assertRaises(PermissionError):
|
|
policy.validate_branch_request({
|
|
"branch": os.environ["AIOS_GITOPS_ALLOWED_BRANCH"],
|
|
"force": True,
|
|
})
|
|
for endpoint in (
|
|
"/api/tag",
|
|
"/api/rollback",
|
|
"/api/model",
|
|
"/api/model-priority",
|
|
"/api/baseline",
|
|
"/api/registry",
|
|
"/api/deploy",
|
|
):
|
|
self.assertNotIn(endpoint, policy.WRITE_ENDPOINTS)
|
|
|
|
@unittest.skipIf(os.name == "nt", "authoritative writer runs on POSIX")
|
|
def test_approved_base_is_ancestor_even_when_main_drifts(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
workspace = Path(root) / "writer"
|
|
workspace.mkdir()
|
|
subprocess.run(["git", "init"], cwd=workspace, check=True, capture_output=True)
|
|
subprocess.run(
|
|
["git", "config", "user.email", "task-plane@example.invalid"],
|
|
cwd=workspace,
|
|
check=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.name", "Task Plane Test"],
|
|
cwd=workspace,
|
|
check=True,
|
|
)
|
|
(workspace / "README.md").write_text("base\n", encoding="utf-8")
|
|
subprocess.run(["git", "add", "README.md"], cwd=workspace, check=True)
|
|
subprocess.run(["git", "commit", "-m", "base"], cwd=workspace, check=True, capture_output=True)
|
|
base = subprocess.check_output(
|
|
["git", "rev-parse", "HEAD"], cwd=workspace, text=True
|
|
).strip()
|
|
branch = os.environ["AIOS_GITOPS_ALLOWED_BRANCH"]
|
|
subprocess.run(["git", "checkout", "-b", branch], cwd=workspace, check=True, capture_output=True)
|
|
(workspace / "README.md").write_text("branch\n", encoding="utf-8")
|
|
subprocess.run(["git", "commit", "-am", "branch"], cwd=workspace, check=True, capture_output=True)
|
|
os.environ.update(
|
|
{
|
|
"AIOS_GITOPS_ALLOWED_BASE_OID": base,
|
|
"AIOS_GITOPS_WRITES_ENABLED": "true",
|
|
"AIOS_GITOPS_RUNTIME_ROLE": "isolated-writer",
|
|
"AIOS_GITOPS_WORKSPACE_ROLE": "isolated-writer",
|
|
"AIOS_GITOPS_REPOSITORY_IDENTITY": "aios/aios",
|
|
"AIOS_GITOPS_AUTHORITATIVE_GIT_DIR": str(workspace / ".git"),
|
|
"AIOS_GITOPS_APPROVED_WRITER_ROOT": str(Path(root)),
|
|
}
|
|
)
|
|
for name in policy._DENY_FLAGS:
|
|
os.environ[name] = "false"
|
|
policy.require_service_invariants(workspace, workspace / ".git")
|
|
|
|
@unittest.skipIf(os.name == "nt", "POSIX audit mode and O_NOFOLLOW contract")
|
|
def test_audit_path_is_contained_durable_and_redacted(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
approved = Path(root) / "approved"
|
|
approved.mkdir(mode=0o750)
|
|
log = approved / "writer.jsonl"
|
|
os.environ["AIOS_GITOPS_APPROVED_AUDIT_ROOT"] = str(approved)
|
|
os.environ["AIOS_GITOPS_AUDIT_LOG"] = str(log)
|
|
policy.append_audit_record(
|
|
{
|
|
"action": "commit",
|
|
"Authorization": "Bearer must-not-appear",
|
|
"payload": "token=must-not-appear",
|
|
}
|
|
)
|
|
text = log.read_text(encoding="utf-8")
|
|
self.assertNotIn("must-not-appear", text)
|
|
self.assertEqual(0o600, log.stat().st_mode & 0o777)
|
|
os.environ["AIOS_GITOPS_AUDIT_LOG"] = str(Path(root) / "escape.jsonl")
|
|
with self.assertRaises(PermissionError):
|
|
policy.approved_audit_log_path()
|
|
|
|
|
|
class SecretScannerBoundaryTests(unittest.TestCase):
|
|
def test_task_control_plane_branch_is_not_an_api_key(self):
|
|
import aios_chatgpt_actions_api as api
|
|
|
|
payload = (
|
|
'{"branch":"patch/aios-agent-task-control-plane-20260724",'
|
|
'"created_by":"agent-task-plane-writer"}'
|
|
)
|
|
self.assertFalse(api.content_has_secret(payload))
|
|
|
|
def test_real_openai_style_key_is_still_rejected(self):
|
|
import aios_chatgpt_actions_api as api
|
|
|
|
payload = '{"value":"sk-abcdefghijklmnopqrstuvwxyz123456"}'
|
|
self.assertTrue(api.content_has_secret(payload))
|