90 lines
3.6 KiB
Python
90 lines
3.6 KiB
Python
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
CHECKER = ROOT / "scripts" / "check-execution-boundaries.py"
|
|
|
|
|
|
class ExecutionBoundaryCheckTests(unittest.TestCase):
|
|
def run_checker(self, root: Path, *files: str) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
["python3", str(CHECKER), "--root", str(root), "--files", *files],
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
|
|
def assert_rejected(self, relative: str, source_text: str) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / relative
|
|
source.parent.mkdir(parents=True)
|
|
source.write_text(source_text, encoding="utf-8")
|
|
result = self.run_checker(root, relative)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("EXECUTION_BOUNDARY", result.stderr)
|
|
|
|
def test_rejects_adapter_alias(self) -> None:
|
|
self.assert_rejected(
|
|
"apps/admin-api/src/service/bypass.rs",
|
|
"use crank_adapter_rest as transport;\n",
|
|
)
|
|
|
|
def test_rejects_each_raw_runtime_entrypoint(self) -> None:
|
|
for method in ("execute_with_context", "execute_request", "prepare_request", "invoke_unary"):
|
|
with self.subTest(method=method):
|
|
self.assert_rejected(
|
|
"crates/crank-community-mcp/src/app/new_helper.rs",
|
|
f"runtime.{method}(request);\n",
|
|
)
|
|
|
|
def test_rejects_execution_reexport_and_direct_sqlx(self) -> None:
|
|
self.assert_rejected(
|
|
"apps/admin-api/src/routes/reexport.rs",
|
|
"pub use crank_runtime::RuntimeExecutor;\n",
|
|
)
|
|
self.assert_rejected(
|
|
"apps/admin-api/src/routes/persist.rs",
|
|
'let _ = sqlx::query("insert into invocation_logs values (...)");\n',
|
|
)
|
|
|
|
def test_rejects_direct_mapping_or_schema_execution(self) -> None:
|
|
self.assert_rejected(
|
|
"apps/admin-api/src/service/mapping_bypass.rs",
|
|
"operation.input_mapping.apply(&input);\n",
|
|
)
|
|
self.assert_rejected(
|
|
"crates/crank-community-mcp/src/app/schema_bypass.rs",
|
|
"operation.input_schema.validate_shape(&input);\n",
|
|
)
|
|
|
|
def test_allows_canonical_runtime_request(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "crates/crank-community-mcp/src/app.rs"
|
|
source.parent.mkdir(parents=True)
|
|
source.write_text("runtime.execute_outcome(request).await;\n", encoding="utf-8")
|
|
result = self.run_checker(root, "crates/crank-community-mcp/src/app.rs")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_explicit_missing_or_symlink_path_fails_closed(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
missing = self.run_checker(root, "apps/admin-api/src/service/missing.rs")
|
|
self.assertNotEqual(missing.returncode, 0)
|
|
target = root / "outside.rs"
|
|
target.write_text("safe", encoding="utf-8")
|
|
link = root / "apps/admin-api/src/service/link.rs"
|
|
link.parent.mkdir(parents=True)
|
|
link.symlink_to(target)
|
|
symlink = self.run_checker(root, "apps/admin-api/src/service/link.rs")
|
|
self.assertNotEqual(symlink.returncode, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|