106 lines
3.9 KiB
Python
106 lines
3.9 KiB
Python
import os
|
|
import sys
|
|
import tempfile
|
|
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"
|
|
|
|
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)
|
|
|
|
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)
|
|
|
|
|
|
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))
|