sysadminandClaude Opus 4.8
34968475c7
assess_canonical_repository_root declared a public `mode` keyword defaulting to "validation" and documented exactly two supported values, but never checked the argument against an allowlist. Both dispatch points were permissive: * the configured-root path tested `mode == "derivation"` and routed every other value into a catch-all `else`, so an unsupported mode silently received validation semantics; and * the single-repository default path tested `mode == "validation"`, so an unsupported mode skipped the identity comparison entirely and was strictly weaker than validation, not an alias of it. Measured at the previous head: mode="invalid_mode" with require_binding=True and matching expected/observed identities returned proven=True, block=False with no reasons; on the unconfigured path an unsupported mode returned proven=True where mode="validation" returned proven=False for identical inputs. Empty string and None behaved like any other unsupported value, and no assessment ever emitted a mode-specific rejection. Add an explicit two-value allowlist (SUPPORTED_MODES) and refuse every other explicitly supplied value — unknown strings, misspellings, case and whitespace variants, the empty string, None, and non-strings — as the first act of the function, before any candidate-root existence check, path or symlink resolution, git top-level discovery, remote-URL or repository-identity discovery, and before any expected-versus-observed comparison or validation/derivation behaviour. The refusal reports proven=False, block=True, a mode-specific reason naming the offending value, and reason_code=DENY_UNKNOWN_MODE, following the existing webui.sanctioned_restart.DENY_UNKNOWN_MODE convention. No repository identity is resolved through a refused mode: resolved_slug and canonical_repo_root are both None. Omission continues to select validation, so the documented default is unchanged. Unsupported modes are refused rather than normalized onto a supported mode. No mode is exposed through MCP request parameters, environment variables, repository configuration, session input, or any public reviewer, merger, issue, PR or lease API; the only production call sites remain an omitted mode (validation) and the hardcoded "derivation" literal. resolve_namespace_mutation_context keeps the install checkout as the canonical root when an assessment resolves none, so a refused mode cannot bind a root derived through an undefined mode while roots_aligned and the carried assessment stay fail-closed. Regressions in tests/test_issue_973_b10_mode_contract.py cover invalid modes with missing, matching and conflicting identities, empty string, explicit None, representative non-strings, misspellings and whitespace variants, omission defaulting to validation, explicit validation and derivation behaviour, the single-repository default path differential, rejection ordering (both spied and mock-free), the absence of any request/environment/configuration injection surface, and fail-closed reviewer, merger, mutation-context and final mutation-authorization behaviour through the production paths. Closes #973 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
740 lines
32 KiB
Python
740 lines
32 KiB
Python
"""Regression tests for Issue #973 blocker B10: repository-authority mode contract.
|
|
|
|
Before this repair ``assess_canonical_repository_root`` never checked its
|
|
``mode`` argument against an allowlist. Both dispatch points were permissive:
|
|
|
|
* the configured-root path tested ``mode == "derivation"`` and sent every other
|
|
value into a catch-all ``else``, so an unsupported mode silently received
|
|
*validation* semantics, and
|
|
* the single-repository default path tested ``mode == "validation"``, so an
|
|
unsupported mode skipped the identity comparison entirely and was strictly
|
|
*weaker* than validation.
|
|
|
|
The measured consequence was that ``mode="invalid_mode"`` with matching expected
|
|
and observed identities returned ``proven: True`` / ``block: False`` with no
|
|
reasons, and that an unsupported mode passed on the default path where
|
|
``"validation"`` correctly blocked.
|
|
|
|
These tests exercise the production module directly with real git repositories —
|
|
no patched stand-in for the function under test — and drive the production
|
|
enforcement and mutation-context consumers rather than only the intermediate
|
|
assessment.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import inspect
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import canonical_repository_root as crr
|
|
import gitea_config
|
|
import gitea_mcp_server as mcp_server
|
|
import namespace_workspace_binding as nwb
|
|
import stable_control_runtime
|
|
|
|
|
|
INSTALL_SLUG = "Scaled-Tech-Consulting/Gitea-Tools"
|
|
TARGET_SLUG = "Scaled-Tech-Consulting/mcp-control-plane"
|
|
FOREIGN_SLUG = "Someone-Else/Evil-Repo"
|
|
|
|
# Explicitly supplied values that must all be refused. Omission is *not* in this
|
|
# list: omitting the argument keeps the documented ``"validation"`` default.
|
|
UNSUPPORTED_STRING_MODES = (
|
|
"invalid_mode",
|
|
"",
|
|
"validaton", # misspelling
|
|
"derivaton", # misspelling
|
|
"Validation", # case variant
|
|
"DERIVATION", # case variant
|
|
" validation", # leading whitespace
|
|
"validation ", # trailing whitespace
|
|
"derivation\n", # trailing newline
|
|
"validation,derivation",
|
|
)
|
|
|
|
UNSUPPORTED_NON_STRING_MODES = (
|
|
None,
|
|
0,
|
|
1,
|
|
True,
|
|
False,
|
|
3.14,
|
|
[],
|
|
["validation"],
|
|
{},
|
|
{"mode": "validation"},
|
|
("validation",),
|
|
object(),
|
|
)
|
|
|
|
|
|
def _init_repo(path: str, remote_url: str, *, user: str = "Test User") -> None:
|
|
os.makedirs(path, exist_ok=True)
|
|
subprocess.run(["git", "init", "-b", "master"], cwd=path, check=True,
|
|
capture_output=True)
|
|
subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=path,
|
|
check=True, capture_output=True)
|
|
subprocess.run(["git", "config", "user.name", user], cwd=path, check=True,
|
|
capture_output=True)
|
|
with open(os.path.join(path, "README.md"), "w") as handle:
|
|
handle.write(f"{os.path.basename(path)}\n")
|
|
subprocess.run(["git", "add", "README.md"], cwd=path, check=True,
|
|
capture_output=True)
|
|
subprocess.run(["git", "commit", "-m", "initial"], cwd=path, check=True,
|
|
capture_output=True)
|
|
subprocess.run(["git", "remote", "add", "prgs", remote_url], cwd=path,
|
|
check=True, capture_output=True)
|
|
|
|
|
|
class _CanonicalRootFixture(unittest.TestCase):
|
|
"""Real install / target / foreign git repositories, as in the #973 suite."""
|
|
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.tmp_dir = os.path.realpath(self._tmp.name)
|
|
|
|
self.install_root = os.path.join(self.tmp_dir, "Gitea-Tools")
|
|
_init_repo(
|
|
self.install_root,
|
|
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
|
|
)
|
|
|
|
self.target_root = os.path.join(self.tmp_dir, "mcp-control-plane")
|
|
_init_repo(
|
|
self.target_root,
|
|
"https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git",
|
|
)
|
|
|
|
self.evil_root = os.path.join(self.tmp_dir, "Evil-Repo")
|
|
_init_repo(
|
|
self.evil_root,
|
|
"https://gitea.prgs.cc/Someone-Else/Evil-Repo.git",
|
|
user="Evil User",
|
|
)
|
|
|
|
self.target_branches = os.path.join(self.target_root, "branches")
|
|
self.target_worktree = os.path.join(self.target_branches, "rev-pr-99")
|
|
subprocess.run(
|
|
["git", "worktree", "add", "-b", "rev-pr-99", self.target_worktree],
|
|
cwd=self.target_root, check=True, capture_output=True,
|
|
)
|
|
|
|
def tearDown(self):
|
|
self._tmp.cleanup()
|
|
|
|
def assertRefusedForMode(self, assessment: dict, mode) -> None:
|
|
"""Assert a fail-closed refusal attributable to *mode* and nothing else."""
|
|
self.assertFalse(assessment["proven"], assessment)
|
|
self.assertTrue(assessment["block"], assessment)
|
|
self.assertEqual(assessment["reason_code"], crr.DENY_UNKNOWN_MODE, assessment)
|
|
self.assertEqual(len(assessment["reasons"]), 1, assessment)
|
|
reason = assessment["reasons"][0]
|
|
self.assertIn("unsupported repository-authority mode", reason)
|
|
self.assertIn(repr(mode), reason)
|
|
# No trusted repository identity may be derived through an invalid mode.
|
|
self.assertIsNone(assessment["resolved_slug"], assessment)
|
|
self.assertIsNone(assessment["canonical_repo_root"], assessment)
|
|
|
|
|
|
class TestB10UnsupportedModeIsRejected(_CanonicalRootFixture):
|
|
"""Direct assessment tests for invalid-mode parsing."""
|
|
|
|
def test_invalid_string_with_missing_expected_identity(self):
|
|
"""G1: blocks for the mode, not incidentally for a missing identity."""
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root,
|
|
source="env",
|
|
expected_slug=None,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
mode="invalid_mode",
|
|
)
|
|
self.assertRefusedForMode(assessment, "invalid_mode")
|
|
self.assertFalse(
|
|
any("unprovable or missing" in r for r in assessment["reasons"]),
|
|
"must block because the mode is unsupported, not because the expected "
|
|
"identity happened to be missing",
|
|
)
|
|
|
|
def test_invalid_string_with_matching_identities(self):
|
|
"""G2: the contract violation — matching identities used to return proven."""
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root,
|
|
source="env",
|
|
expected_slug=TARGET_SLUG,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
mode="invalid_mode",
|
|
)
|
|
self.assertRefusedForMode(assessment, "invalid_mode")
|
|
|
|
def test_invalid_string_with_conflicting_identities(self):
|
|
"""G3: refused for the mode, not for the incidental identity mismatch."""
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.evil_root,
|
|
source="env",
|
|
expected_slug=INSTALL_SLUG,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
mode="invalid_mode",
|
|
)
|
|
self.assertRefusedForMode(assessment, "invalid_mode")
|
|
self.assertFalse(
|
|
any("identity mismatch" in r for r in assessment["reasons"]),
|
|
"the identity comparison must not have run at all",
|
|
)
|
|
|
|
def test_empty_string_mode(self):
|
|
"""G4a: an explicitly supplied empty string is an unsupported value."""
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root,
|
|
source="env",
|
|
expected_slug=TARGET_SLUG,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
mode="",
|
|
)
|
|
self.assertRefusedForMode(assessment, "")
|
|
|
|
def test_explicit_none_mode(self):
|
|
"""G4b: explicit None is refused; it is not treated as omission."""
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root,
|
|
source="env",
|
|
expected_slug=TARGET_SLUG,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
mode=None,
|
|
)
|
|
self.assertRefusedForMode(assessment, None)
|
|
self.assertIn("of type NoneType", assessment["reasons"][0])
|
|
|
|
def test_representative_non_string_modes(self):
|
|
for mode in UNSUPPORTED_NON_STRING_MODES:
|
|
with self.subTest(mode=repr(mode)):
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root,
|
|
source="env",
|
|
expected_slug=TARGET_SLUG,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
mode=mode,
|
|
)
|
|
self.assertRefusedForMode(assessment, mode)
|
|
self.assertIn(
|
|
f"of type {type(mode).__name__}", assessment["reasons"][0]
|
|
)
|
|
|
|
def test_unknown_strings_misspellings_and_whitespace_variants(self):
|
|
for mode in UNSUPPORTED_STRING_MODES:
|
|
with self.subTest(mode=repr(mode)):
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root,
|
|
source="env",
|
|
expected_slug=TARGET_SLUG,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
mode=mode,
|
|
)
|
|
self.assertRefusedForMode(assessment, mode)
|
|
|
|
def test_supported_modes_are_exactly_two(self):
|
|
self.assertEqual(
|
|
crr.SUPPORTED_MODES, ("validation", "derivation")
|
|
)
|
|
self.assertIsNone(crr.unsupported_mode_reason("validation"))
|
|
self.assertIsNone(crr.unsupported_mode_reason("derivation"))
|
|
self.assertIsNotNone(crr.unsupported_mode_reason("invalid_mode"))
|
|
|
|
|
|
class TestB10SupportedModesUnchanged(_CanonicalRootFixture):
|
|
"""The repair must not disturb the two documented modes."""
|
|
|
|
def test_omitted_mode_defaults_to_validation(self):
|
|
"""G4c/G4d: omission still selects validation, proven by both outcomes."""
|
|
matching = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root,
|
|
source="env",
|
|
expected_slug=TARGET_SLUG,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
)
|
|
self.assertTrue(matching["proven"], matching)
|
|
self.assertFalse(matching["block"], matching)
|
|
self.assertIsNone(matching["reason_code"], matching)
|
|
self.assertEqual(matching["resolved_slug"], TARGET_SLUG)
|
|
|
|
conflicting = crr.assess_canonical_repository_root(
|
|
configured_value=self.evil_root,
|
|
source="env",
|
|
expected_slug=INSTALL_SLUG,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
)
|
|
self.assertFalse(conflicting["proven"], conflicting)
|
|
self.assertTrue(conflicting["block"], conflicting)
|
|
self.assertIsNone(conflicting["reason_code"], conflicting)
|
|
self.assertTrue(
|
|
any("identity mismatch" in r for r in conflicting["reasons"]),
|
|
"omission must behave exactly like explicit validation",
|
|
)
|
|
|
|
def test_explicit_validation_retains_strict_behavior(self):
|
|
"""G5a/G5b/G5c."""
|
|
ok = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root, source="env",
|
|
expected_slug=TARGET_SLUG, process_project_root=self.install_root,
|
|
remote="prgs", require_binding=True, mode="validation",
|
|
)
|
|
self.assertTrue(ok["proven"], ok)
|
|
|
|
mismatch = crr.assess_canonical_repository_root(
|
|
configured_value=self.evil_root, source="env",
|
|
expected_slug=INSTALL_SLUG, process_project_root=self.install_root,
|
|
remote="prgs", require_binding=True, mode="validation",
|
|
)
|
|
self.assertTrue(mismatch["block"], mismatch)
|
|
self.assertTrue(any("identity mismatch" in r for r in mismatch["reasons"]))
|
|
|
|
unprovable = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root, source="env",
|
|
expected_slug=None, process_project_root=self.install_root,
|
|
remote="prgs", require_binding=True, mode="validation",
|
|
)
|
|
self.assertTrue(unprovable["block"], unprovable)
|
|
self.assertTrue(
|
|
any("unprovable or missing" in r for r in unprovable["reasons"])
|
|
)
|
|
|
|
def test_explicit_derivation_retains_trusted_derivation(self):
|
|
"""G5d: derivation still resolves identity with no expected slug."""
|
|
derived = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root, source="env",
|
|
expected_slug=None, process_project_root=self.install_root,
|
|
remote="prgs", require_binding=True, mode="derivation",
|
|
)
|
|
self.assertTrue(derived["proven"], derived)
|
|
self.assertFalse(derived["block"], derived)
|
|
self.assertIsNone(derived["reason_code"], derived)
|
|
self.assertEqual(derived["resolved_slug"], TARGET_SLUG)
|
|
|
|
def test_derivation_without_resolvable_remote_still_fails_closed(self):
|
|
no_remote = os.path.join(self.tmp_dir, "no-remote-target")
|
|
os.makedirs(no_remote)
|
|
subprocess.run(["git", "init", "-b", "master"], cwd=no_remote, check=True,
|
|
capture_output=True)
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=no_remote, source="env", expected_slug=None,
|
|
process_project_root=self.install_root, remote="prgs",
|
|
require_binding=True, mode="derivation",
|
|
)
|
|
self.assertTrue(assessment["block"], assessment)
|
|
self.assertTrue(
|
|
any("no resolvable" in r for r in assessment["reasons"]), assessment
|
|
)
|
|
|
|
|
|
class TestB10SingleRepositoryDefaultPath(_CanonicalRootFixture):
|
|
"""G6: on the unconfigured path an invalid mode used to be weaker than validation."""
|
|
|
|
def _assess(self, mode_kwargs: dict) -> dict:
|
|
return crr.assess_canonical_repository_root(
|
|
configured_value=None,
|
|
source=None,
|
|
expected_slug=FOREIGN_SLUG,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=False,
|
|
**mode_kwargs,
|
|
)
|
|
|
|
def test_validation_blocks_a_foreign_expected_identity(self):
|
|
"""G6b: the reference behaviour the invalid mode must not undercut."""
|
|
got = self._assess({"mode": "validation"})
|
|
self.assertFalse(got["proven"], got)
|
|
self.assertTrue(got["block"], got)
|
|
self.assertTrue(any("identity mismatch" in r for r in got["reasons"]))
|
|
|
|
def test_invalid_mode_no_longer_passes_where_validation_blocks(self):
|
|
"""G6a: identical inputs, only the mode differs — must not fail open."""
|
|
invalid = self._assess({"mode": "invalid_mode"})
|
|
self.assertRefusedForMode(invalid, "invalid_mode")
|
|
|
|
validation = self._assess({"mode": "validation"})
|
|
self.assertEqual(
|
|
invalid["proven"], validation["proven"],
|
|
"an unsupported mode must never be more permissive than validation",
|
|
)
|
|
self.assertTrue(invalid["block"] and validation["block"])
|
|
|
|
def test_empty_string_mode_on_default_path(self):
|
|
"""G6c."""
|
|
self.assertRefusedForMode(self._assess({"mode": ""}), "")
|
|
|
|
def test_omitted_mode_on_default_path_still_validates(self):
|
|
got = self._assess({})
|
|
self.assertFalse(got["proven"], got)
|
|
self.assertTrue(any("identity mismatch" in r for r in got["reasons"]), got)
|
|
|
|
|
|
class TestB10RejectionOrdering(_CanonicalRootFixture):
|
|
"""Refusal must precede every form of candidate-root or Git inspection."""
|
|
|
|
def test_no_git_or_identity_discovery_runs_for_an_unsupported_mode(self):
|
|
with patch.object(crr, "resolve_repo_toplevel") as toplevel, \
|
|
patch.object(crr, "repository_identity_slug") as identity:
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root,
|
|
source="env",
|
|
expected_slug=TARGET_SLUG,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
mode="invalid_mode",
|
|
)
|
|
self.assertRefusedForMode(assessment, "invalid_mode")
|
|
toplevel.assert_not_called()
|
|
identity.assert_not_called()
|
|
|
|
def test_the_same_spies_do_fire_for_a_supported_mode(self):
|
|
"""Control: proves the previous test's assertions are not vacuous."""
|
|
with patch.object(crr, "resolve_repo_toplevel",
|
|
wraps=crr.resolve_repo_toplevel) as toplevel, \
|
|
patch.object(crr, "repository_identity_slug",
|
|
wraps=crr.repository_identity_slug) as identity:
|
|
crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root,
|
|
source="env",
|
|
expected_slug=TARGET_SLUG,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
mode="validation",
|
|
)
|
|
toplevel.assert_called()
|
|
identity.assert_called()
|
|
|
|
def test_nonexistent_candidate_root_still_reports_the_mode_refusal(self):
|
|
"""No mocks: the existence check cannot have run before the refusal."""
|
|
nonexistent = os.path.join(self.tmp_dir, "no-such-repository")
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=nonexistent,
|
|
source="env",
|
|
expected_slug=TARGET_SLUG,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
mode="invalid_mode",
|
|
)
|
|
self.assertRefusedForMode(assessment, "invalid_mode")
|
|
self.assertFalse(
|
|
any("does not exist" in r for r in assessment["reasons"]), assessment
|
|
)
|
|
|
|
def test_symlinked_candidate_root_is_not_resolved_for_an_unsupported_mode(self):
|
|
link = os.path.join(self.tmp_dir, "target-alias")
|
|
os.symlink(self.target_root, link)
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=link,
|
|
source="env",
|
|
expected_slug=TARGET_SLUG,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
mode="invalid_mode",
|
|
)
|
|
self.assertRefusedForMode(assessment, "invalid_mode")
|
|
# The refusal payload must not leak a resolved path for the candidate.
|
|
self.assertIsNone(assessment["canonical_repo_root"], assessment)
|
|
|
|
|
|
class TestB10NoModeInjectionSurface(unittest.TestCase):
|
|
"""``mode`` must not be reachable from requests, environment, or config."""
|
|
|
|
PRODUCTION_MODULES = ("gitea_mcp_server.py", "namespace_workspace_binding.py")
|
|
|
|
def _repo_root(self) -> str:
|
|
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
def test_production_call_sites_only_pass_allowlisted_literal_modes(self):
|
|
"""Valid hardcoded modes reach the correct path; nothing else is passed."""
|
|
seen: list[tuple[str, str | None]] = []
|
|
for name in self.PRODUCTION_MODULES:
|
|
path = os.path.join(self._repo_root(), name)
|
|
with open(path) as handle:
|
|
tree = ast.parse(handle.read())
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, ast.Call):
|
|
continue
|
|
func = node.func
|
|
called = (
|
|
func.attr if isinstance(func, ast.Attribute)
|
|
else getattr(func, "id", None)
|
|
)
|
|
if called != "assess_canonical_repository_root":
|
|
continue
|
|
supplied = [k for k in node.keywords if k.arg == "mode"]
|
|
if not supplied:
|
|
seen.append((name, None))
|
|
continue
|
|
value = supplied[0].value
|
|
self.assertIsInstance(
|
|
value, ast.Constant,
|
|
f"{name}: mode must be a literal, never a variable or expression",
|
|
)
|
|
self.assertIn(
|
|
value.value, crr.SUPPORTED_MODES,
|
|
f"{name}: unsupported mode literal {value.value!r}",
|
|
)
|
|
seen.append((name, value.value))
|
|
self.assertTrue(seen, "expected production call sites to be found")
|
|
# Both documented modes are exercised by production, and omission is used.
|
|
self.assertIn(None, [mode for _, mode in seen])
|
|
self.assertIn("derivation", [mode for _, mode in seen])
|
|
|
|
def test_no_public_entry_point_exposes_a_mode_parameter(self):
|
|
for func in (
|
|
nwb.resolve_namespace_mutation_context,
|
|
nwb.assess_namespace_mutation_workspace,
|
|
mcp_server._resolve_namespace_mutation_context,
|
|
mcp_server._enforce_canonical_repository_root,
|
|
mcp_server._canonical_repository_slug,
|
|
mcp_server._trusted_session_repository,
|
|
mcp_server._resolve_expected_repository_slug,
|
|
):
|
|
with self.subTest(func=func.__name__):
|
|
self.assertNotIn("mode", inspect.signature(func).parameters)
|
|
|
|
def test_no_environment_key_selects_a_repository_authority_mode(self):
|
|
for key in gitea_config.RECOGNIZED_GITEA_ENV_KEYS:
|
|
self.assertNotIn(
|
|
"CANONICAL_REPOSITORY_MODE", key.upper(),
|
|
f"{key} would expose a repository-authority mode selector",
|
|
)
|
|
# An invented mode-ish variable is simply not consumed by crr.
|
|
env = {
|
|
"GITEA_CANONICAL_REPOSITORY_ROOT": "/some/path",
|
|
"GITEA_CANONICAL_REPOSITORY_MODE": "invalid_mode",
|
|
}
|
|
value, source = crr.configured_canonical_root(None, env)
|
|
self.assertEqual(value, "/some/path")
|
|
self.assertNotIn("mode", (source or "").lower())
|
|
self.assertIn(
|
|
"GITEA_CANONICAL_REPOSITORY_MODE",
|
|
gitea_config.get_unconsumed_gitea_env_overrides(env),
|
|
"an unknown GITEA_* key must still be rejected as unrecognised",
|
|
)
|
|
|
|
def test_repository_configuration_carries_no_mode_field(self):
|
|
profile = {
|
|
"canonical_repository_root": "/some/path",
|
|
"mode": "invalid_mode",
|
|
}
|
|
value, source = crr.configured_canonical_root(profile, {})
|
|
self.assertEqual(value, "/some/path")
|
|
self.assertEqual(source, "profile canonical_repository_root")
|
|
|
|
|
|
class TestB10ProductionEnforcementPaths(_CanonicalRootFixture):
|
|
"""Production enforcement, mutation-context, reviewer and merger consumers."""
|
|
|
|
def _force_invalid_mode(self):
|
|
"""Simulate a future call site threading an unsupported mode.
|
|
|
|
The real ``assess_canonical_repository_root`` still executes — only the
|
|
caller-side argument is substituted — so the refusal under test is
|
|
produced by production code, not by a stand-in. No caller-controlled
|
|
``mode`` parameter is added to any production signature to achieve this.
|
|
"""
|
|
real = crr.assess_canonical_repository_root
|
|
|
|
def _wrapper(**kwargs):
|
|
kwargs["mode"] = "invalid_mode"
|
|
return real(**kwargs)
|
|
|
|
return patch.object(crr, "assess_canonical_repository_root", _wrapper)
|
|
|
|
def test_mutation_context_fails_closed_under_a_refused_mode(self):
|
|
with self._force_invalid_mode():
|
|
ctx = nwb.resolve_namespace_mutation_context(
|
|
role_kind="reviewer",
|
|
worktree_path=self.target_worktree,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=self.target_root,
|
|
expected_slug=TARGET_SLUG,
|
|
remote="prgs",
|
|
)
|
|
assessment = ctx["canonical_root_assessment"]
|
|
self.assertFalse(ctx["roots_aligned"], ctx)
|
|
self.assertTrue(assessment["block"], assessment)
|
|
self.assertEqual(assessment["reason_code"], crr.DENY_UNKNOWN_MODE)
|
|
self.assertIsNone(assessment["resolved_slug"])
|
|
# The refused mode must not yield the candidate root as canonical.
|
|
self.assertNotEqual(ctx["canonical_repo_root"], self.target_root)
|
|
self.assertEqual(ctx["canonical_repo_root"], self.install_root)
|
|
|
|
def test_mutation_context_unchanged_for_the_supported_default(self):
|
|
ctx = nwb.resolve_namespace_mutation_context(
|
|
role_kind="reviewer",
|
|
worktree_path=self.target_worktree,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=self.target_root,
|
|
expected_slug=TARGET_SLUG,
|
|
remote="prgs",
|
|
)
|
|
self.assertTrue(ctx["roots_aligned"], ctx)
|
|
self.assertEqual(ctx["canonical_repo_root"], self.target_root)
|
|
self.assertIsNone(ctx["canonical_root_assessment"]["reason_code"])
|
|
|
|
def test_reviewer_and_merger_authorization_fails_closed_under_a_refused_mode(self):
|
|
for role in ("reviewer", "merger"):
|
|
with self.subTest(role=role), self._force_invalid_mode():
|
|
assessment = nwb.assess_namespace_mutation_workspace(
|
|
role_kind=role,
|
|
worktree_path=self.target_worktree,
|
|
worktree=None,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=self.target_root,
|
|
expected_slug=TARGET_SLUG,
|
|
remote="prgs",
|
|
)
|
|
self.assertTrue(assessment["block"], assessment)
|
|
self.assertTrue(
|
|
any("unsupported repository-authority mode" in r
|
|
for r in assessment["reasons"]),
|
|
assessment,
|
|
)
|
|
|
|
def test_reviewer_and_merger_authorization_unchanged_for_supported_modes(self):
|
|
for role in ("reviewer", "merger"):
|
|
with self.subTest(role=role):
|
|
assessment = nwb.assess_namespace_mutation_workspace(
|
|
role_kind=role,
|
|
worktree_path=self.target_worktree,
|
|
worktree=None,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=self.target_root,
|
|
expected_slug=TARGET_SLUG,
|
|
remote="prgs",
|
|
)
|
|
self.assertFalse(assessment["block"], assessment)
|
|
|
|
def test_final_mutation_gate_blocks_when_the_mode_refusal_unaligns_roots(self):
|
|
with self._force_invalid_mode():
|
|
ctx = nwb.resolve_namespace_mutation_context(
|
|
role_kind="reviewer",
|
|
worktree_path=self.target_worktree,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=self.target_root,
|
|
expected_slug=TARGET_SLUG,
|
|
remote="prgs",
|
|
)
|
|
report = stable_control_runtime.build_runtime_report(
|
|
process_root=self.install_root,
|
|
checkout_branch="master",
|
|
runtime_head="abcdef123456",
|
|
active_task_workspace=ctx["workspace_path"],
|
|
canonical_repository_root=ctx["canonical_repo_root"],
|
|
workspace_roots_aligned=ctx["roots_aligned"],
|
|
)
|
|
gate = stable_control_runtime.assess_runtime_mutation_gate(report)
|
|
self.assertTrue(gate["block"], gate)
|
|
|
|
def test_enforce_canonical_repository_root_uses_the_validation_default(self):
|
|
"""B8 boundary intact: a foreign configured root raises through production."""
|
|
with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
|
|
patch.object(mcp_server, "_configured_canonical_root",
|
|
return_value=(self.evil_root, "env")), \
|
|
patch.object(mcp_server.session_ctx, "get_session_context",
|
|
return_value=None):
|
|
with self.assertRaises(RuntimeError) as raised:
|
|
mcp_server._enforce_canonical_repository_root(remote="prgs")
|
|
self.assertIn("identity mismatch", str(raised.exception))
|
|
|
|
def test_enforce_canonical_repository_root_refuses_a_threaded_invalid_mode(self):
|
|
bound = {"org": "Scaled-Tech-Consulting",
|
|
"repository": "mcp-control-plane", "remote": "prgs"}
|
|
with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
|
|
patch.object(mcp_server, "_configured_canonical_root",
|
|
return_value=(self.target_root, "env")), \
|
|
patch.object(mcp_server.session_ctx, "get_session_context",
|
|
return_value=bound), \
|
|
self._force_invalid_mode():
|
|
with self.assertRaises(RuntimeError) as raised:
|
|
mcp_server._enforce_canonical_repository_root(remote="prgs")
|
|
self.assertIn("unsupported repository-authority mode", str(raised.exception))
|
|
|
|
def test_enforce_canonical_repository_root_passes_for_a_valid_binding(self):
|
|
bound = {"org": "Scaled-Tech-Consulting",
|
|
"repository": "mcp-control-plane", "remote": "prgs"}
|
|
with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
|
|
patch.object(mcp_server, "_configured_canonical_root",
|
|
return_value=(self.target_root, "env")), \
|
|
patch.object(mcp_server.session_ctx, "get_session_context",
|
|
return_value=bound), \
|
|
patch.object(mcp_server.session_ctx, "assess_session_context",
|
|
return_value={"block": False, "reasons": []}), \
|
|
patch.object(mcp_server, "get_profile",
|
|
return_value={"profile_name": "prgs-reviewer"}):
|
|
mcp_server._enforce_canonical_repository_root(remote="prgs")
|
|
|
|
def test_canonical_repository_slug_derivation_unbroken(self):
|
|
"""B9 boundary intact: legitimate cross-repository derivation still works."""
|
|
profile = {
|
|
"profile_name": "prgs-author",
|
|
"allowed_repositories": [TARGET_SLUG],
|
|
"canonical_repository_root": self.target_root,
|
|
}
|
|
with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
|
|
patch.object(mcp_server.session_ctx, "get_session_context",
|
|
return_value=None):
|
|
slug, reasons = mcp_server._canonical_repository_slug(profile, "prgs")
|
|
self.assertEqual(slug, TARGET_SLUG, reasons)
|
|
self.assertEqual(reasons, [])
|
|
|
|
def test_canonical_repository_slug_fails_closed_under_a_refused_mode(self):
|
|
profile = {
|
|
"profile_name": "prgs-author",
|
|
"allowed_repositories": [TARGET_SLUG],
|
|
"canonical_repository_root": self.target_root,
|
|
}
|
|
with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
|
|
patch.object(mcp_server.session_ctx, "get_session_context",
|
|
return_value=None), \
|
|
self._force_invalid_mode():
|
|
slug, reasons = mcp_server._canonical_repository_slug(profile, "prgs")
|
|
self.assertIsNone(slug)
|
|
self.assertTrue(
|
|
any("unsupported repository-authority mode" in r for r in reasons),
|
|
reasons,
|
|
)
|
|
result = mcp_server._trusted_session_repository(
|
|
profile, "prgs", for_mutation=True
|
|
)
|
|
self.assertIsNone(result["org"])
|
|
self.assertIsNone(result["repository"])
|
|
self.assertTrue(result["reasons"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|