318 lines
13 KiB
Python
318 lines
13 KiB
Python
import importlib.util
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
CHECKER = ROOT / "scripts" / "check-community-scope.py"
|
|
|
|
|
|
def load_checker_module():
|
|
spec = importlib.util.spec_from_file_location("community_scope_checker", CHECKER)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError("cannot load Community scope checker")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class CommunityScopeCheckTests(unittest.TestCase):
|
|
def run_checker(self, root: Path, files: list[str]) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
[sys.executable, str(CHECKER), "--root", str(root), "--files", *files],
|
|
check=False,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
|
|
def run_default_checker(self, root: Path) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
[sys.executable, str(CHECKER), "--root", str(root)],
|
|
check=False,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
|
|
def test_passes_for_clean_community_content(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "README.md").write_text(
|
|
"Crank publishes REST API endpoints as MCP tools.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = self.run_checker(root, ["README.md"])
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("Community scope check passed", result.stdout)
|
|
|
|
def test_fails_for_forbidden_product_marker(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "apps").mkdir()
|
|
(root / "apps" / "ui.md").write_text(
|
|
"Hidden GraphQL protocol card must not be here.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = self.run_checker(root, ["apps/ui.md"])
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("apps/ui.md:1", result.stderr)
|
|
self.assertIn("graphql", result.stderr.lower())
|
|
|
|
def test_inline_directive_allows_only_declared_marker(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "protocol.rs").write_text(
|
|
'assert_rejected_protocol("grpc"); '
|
|
"// community-scope: allow=grpc\n"
|
|
'assert_rejected_protocol("grpc");\n'
|
|
'assert_rejected_protocol("graphql"); '
|
|
"// community-scope: allow=grpc\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = self.run_checker(root, ["protocol.rs"])
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("protocol.rs:2: forbidden marker `grpc`", result.stderr)
|
|
self.assertIn("protocol.rs:3: forbidden marker `graphql`", result.stderr)
|
|
self.assertNotIn("protocol.rs:1:", result.stderr)
|
|
|
|
def test_rejects_marker_without_inline_directive(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "README.md").write_text(
|
|
"Enable grpc transport.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = self.run_checker(root, ["README.md"])
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("forbidden marker `grpc`", result.stderr)
|
|
|
|
def test_ignores_default_excluded_paths(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "LICENSE").write_text(
|
|
"The license text may mention commercial distribution terms.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = self.run_checker(root, ["LICENSE"])
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_excluded_explicit_paths_are_validated_before_skip(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "target").mkdir()
|
|
(root / "target" / "generated.txt").write_text(
|
|
"generated output\n", encoding="utf-8"
|
|
)
|
|
|
|
valid_excluded = self.run_checker(root, ["target/generated.txt"])
|
|
self.assertEqual(valid_excluded.returncode, 0, valid_excluded.stderr)
|
|
|
|
for raw_path in ["LICENSE", "target/missing.txt", "target/../missing.txt"]:
|
|
with self.subTest(raw_path=raw_path):
|
|
rejected = self.run_checker(root, [raw_path])
|
|
self.assertNotEqual(rejected.returncode, 0)
|
|
self.assertIn("invalid explicit file path", rejected.stderr)
|
|
|
|
def test_default_discovery_ignores_tracked_deletions(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
subprocess.run(["git", "init", "-q"], cwd=root, check=True)
|
|
deleted = root / "deleted.txt"
|
|
deleted.write_text("clean\n", encoding="utf-8")
|
|
subprocess.run(["git", "add", "deleted.txt"], cwd=root, check=True)
|
|
deleted.unlink()
|
|
|
|
result = self.run_default_checker(root)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_skips_binary_files(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "logo.png").write_bytes(b"\x89PNG\x00enterprise\x00")
|
|
|
|
result = self.run_checker(root, ["logo.png"])
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_allows_planned_community_capability_identifiers(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "inventory.txt").write_text(
|
|
"resources prompts tasks load_runs\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = self.run_checker(root, ["inventory.txt"])
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_rejects_forbidden_identifier_after_narrow_inline_allow(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "scope.txt").write_text(
|
|
"enterprise_rbac enforcement fixture "
|
|
"# community-scope: allow=enterprise\n"
|
|
"enterprise_rbac product capability\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = self.run_checker(root, ["scope.txt"])
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertNotIn("scope.txt:1:", result.stderr)
|
|
self.assertIn("scope.txt:2: forbidden marker `enterprise`", result.stderr)
|
|
|
|
def test_explicit_paths_fail_closed_without_disclosing_host_paths(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
outside = root.parent / "SECRET_CANARY_scope-outside.txt"
|
|
outside.write_text("clean\n", encoding="utf-8")
|
|
self.addCleanup(outside.unlink, missing_ok=True)
|
|
internal_target = root / "target.txt"
|
|
internal_target.write_text("clean\n", encoding="utf-8")
|
|
internal_link = root / "internal-link.txt"
|
|
internal_link.symlink_to(internal_target)
|
|
outside_link = root / "outside-link.txt"
|
|
outside_link.symlink_to(outside)
|
|
|
|
cases = [
|
|
"missing.txt",
|
|
f"../{outside.name}",
|
|
str(outside),
|
|
"internal-link.txt",
|
|
"outside-link.txt",
|
|
]
|
|
for raw_path in cases:
|
|
with self.subTest(raw_path=raw_path):
|
|
result = self.run_checker(root, [raw_path])
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("invalid explicit file path", result.stderr)
|
|
self.assertNotIn("SECRET_CANARY", result.stderr)
|
|
|
|
def test_invalid_path_report_is_bounded_and_deterministically_truncated(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
missing = [f"missing-{index:04d}.txt" for index in range(1_200, 0, -1)]
|
|
|
|
first = self.run_checker(root, missing)
|
|
second = self.run_checker(root, missing)
|
|
|
|
self.assertNotEqual(first.returncode, 0)
|
|
self.assertEqual(first.stderr, second.stderr)
|
|
self.assertLessEqual(len(first.stderr.encode("utf-8")), 64 * 1024)
|
|
self.assertIn("error: findings truncated", first.stderr)
|
|
self.assertNotIn("missing-", first.stderr)
|
|
|
|
def test_single_file_finding_limit_emits_truncation_marker(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "scope.txt").write_text(
|
|
"grpc\n" * 1_001,
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = self.run_checker(root, ["scope.txt"])
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("error: findings truncated", result.stderr)
|
|
self.assertLessEqual(len(result.stderr.encode("utf-8")), 64 * 1024)
|
|
|
|
def test_exact_finding_limit_is_not_reported_as_truncated(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "scope.txt").write_text("grpc\n" * 1_000, encoding="utf-8")
|
|
|
|
result = self.run_checker(root, ["scope.txt"])
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertNotIn("error: findings truncated", result.stderr)
|
|
|
|
def test_rejects_camel_and_pascal_case_forbidden_identifiers(self) -> None:
|
|
cases = {
|
|
"EnterpriseRbac": "enterprise",
|
|
"MultiWorkspace": "multi-workspace",
|
|
"NonRestUpstream": "non-rest-upstream",
|
|
"ArbitraryDistributedLoadTargets": "distributed-load-targets",
|
|
}
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
for index, (identifier, marker) in enumerate(cases.items()):
|
|
path = root / f"scope-{index}.txt"
|
|
path.write_text(identifier + "\n", encoding="utf-8")
|
|
with self.subTest(identifier=identifier):
|
|
result = self.run_checker(root, [path.name])
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn(f"forbidden marker `{marker}`", result.stderr)
|
|
|
|
def test_invalid_utf8_text_fails_closed(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "scope.txt").write_bytes(b"enterpr\xffise capability\n")
|
|
|
|
result = self.run_checker(root, ["scope.txt"])
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("invalid explicit file path", result.stderr)
|
|
self.assertNotIn("Traceback", result.stderr)
|
|
|
|
def test_control_characters_in_explicit_paths_cannot_forge_report_lines(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
forged_path = "forged\nerror: injected.txt"
|
|
(root / forged_path).write_text("grpc\n", encoding="utf-8")
|
|
|
|
result = self.run_checker(root, [forged_path])
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(result.stderr.count("\n"), 2)
|
|
self.assertNotIn("forged", result.stderr)
|
|
self.assertNotIn("injected", result.stderr)
|
|
|
|
def test_oversized_single_line_fails_closed_without_materializing_it(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "scope.txt").write_text("x" * 65_537, encoding="utf-8")
|
|
|
|
result = self.run_checker(root, ["scope.txt"])
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("invalid explicit file path", result.stderr)
|
|
self.assertNotIn("Traceback", result.stderr)
|
|
|
|
def test_open_failure_after_validation_returns_a_safe_failure(self) -> None:
|
|
checker = load_checker_module()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "scope.txt").write_text("clean\n", encoding="utf-8")
|
|
|
|
with mock.patch.object(Path, "open", side_effect=OSError("host path canary")):
|
|
findings, valid, truncated, scanned_bytes = checker.scan_file(
|
|
root, "scope.txt"
|
|
)
|
|
|
|
self.assertEqual(findings, [])
|
|
self.assertFalse(valid)
|
|
self.assertFalse(truncated)
|
|
self.assertEqual(scanned_bytes, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|