From 34968475c7119426bcd22c4b36b6947d873562c3 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 29 Jul 2026 17:52:32 -0400 Subject: [PATCH] fix(runtime): reject unsupported repository-authority modes (#973 B10) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- canonical_repository_root.py | 93 ++- namespace_workspace_binding.py | 12 +- tests/test_issue_973_b10_mode_contract.py | 739 ++++++++++++++++++++++ 3 files changed, 839 insertions(+), 5 deletions(-) create mode 100644 tests/test_issue_973_b10_mode_contract.py diff --git a/canonical_repository_root.py b/canonical_repository_root.py index 20d7824..885821c 100644 --- a/canonical_repository_root.py +++ b/canonical_repository_root.py @@ -37,6 +37,48 @@ CANONICAL_ROOT_ENV = "GITEA_CANONICAL_REPOSITORY_ROOT" # Candidate git remote names probed when deriving repository identity. _IDENTITY_REMOTE_CANDIDATES = ("prgs", "origin", "dadeschools", "mdcps") +# Repository-authority modes (#973 B10). Exactly two values are supported. +# ``mode`` selects how repository authority is established, so an unrecognised +# value must never be normalised onto one of these: aliasing a trusted mode is +# precisely the defect. Omitting the argument keeps the documented safe default, +# ``validation``. +MODE_VALIDATION = "validation" +MODE_DERIVATION = "derivation" +SUPPORTED_MODES: tuple[str, ...] = (MODE_VALIDATION, MODE_DERIVATION) + +# Reason code emitted when an unsupported mode is refused. Mirrors the existing +# ``webui.sanctioned_restart.DENY_UNKNOWN_MODE`` convention so callers and tests +# can assert the refusal cause rather than string-matching prose. +DENY_UNKNOWN_MODE = "unknown_mode" + + +def unsupported_mode_reason(mode: object) -> str | None: + """Precise rejection reason for *mode*, or None when *mode* is supported. + + Only the two documented string values are accepted, compared exactly — no + stripping, no case folding — so misspellings and whitespace variants are + refused rather than coerced. An empty string, ``None``, and any non-string + are all *explicitly supplied* unsupported values and are refused on the same + footing; none of them is normalised to a supported mode. Omitting the + argument entirely never reaches here with an unsupported value because the + parameter default is ``"validation"``. + """ + if isinstance(mode, str) and mode in SUPPORTED_MODES: + return None + supported = ", ".join(repr(m) for m in SUPPORTED_MODES) + if not isinstance(mode, str): + return ( + f"unsupported repository-authority mode {mode!r} of type " + f"{type(mode).__name__}: only {supported} are supported; the mode " + "is refused before any repository assessment and no repository " + "identity was resolved through it (fail closed)" + ) + return ( + f"unsupported repository-authority mode {mode!r}: only {supported} are " + "supported; the mode is refused before any repository assessment and no " + "repository identity was resolved through it (fail closed)" + ) + def configured_canonical_root( profile: Mapping | None, @@ -141,14 +183,41 @@ def assess_canonical_repository_root( ``configured`` (whether a cross-repo binding was declared), ``resolved_slug`` and ``source``. - *mode*: + *mode* accepts exactly the two values in :data:`SUPPORTED_MODES`: - ``"validation"`` (default): an independently trusted expected repository slug is known (or required). Candidate root's observed identity must match. Unprovable or missing expected identity fails closed when *require_binding* is True. - ``"derivation"``: caller is deriving the canonical repository identity. No expected slug exists yet by design. Derivation succeeds if the configured root exists, is a git repository, and carries a resolvable git remote identity. + + Every other explicitly supplied value — unknown strings, misspellings, the + empty string, ``None``, and non-strings — is refused with ``proven`` False, + ``block`` True, and ``reason_code`` :data:`DENY_UNKNOWN_MODE` (#973 B10). + Omitting *mode* entirely keeps the documented ``"validation"`` default. """ + # #973 B10: refuse an unsupported repository-authority mode as the very first + # act, 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. An unsupported mode previously fell into the catch-all ``else`` + # on the configured-root path (silently receiving validation semantics) and + # skipped the identity comparison entirely on the single-repository default + # path (strictly weaker than validation), so matching identities could return + # ``proven`` True. No repository identity may be resolved through a mode the + # contract does not define. + mode_reason = unsupported_mode_reason(mode) + if mode_reason is not None: + return _assessment( + proven=False, + reasons=[mode_reason], + configured=bool((configured_value or "").strip()), + canonical_repo_root=None, + resolved_slug=None, + source=source, + reason_code=DENY_UNKNOWN_MODE, + ) + process_root = os.path.realpath(process_project_root) declared = (configured_value or "").strip() @@ -170,7 +239,10 @@ def assess_canonical_repository_root( derived = resolve_repo_toplevel(process_root) or process_root resolved_slug = repository_identity_slug(derived, remote=remote) reasons: list[str] = [] - if mode == "validation" and expected_slug: + # #973 B10: the allowlist above guarantees *mode* is one of the two + # supported values here, so an unsupported value can no longer skip this + # identity comparison and end up strictly weaker than validation. + if mode == MODE_VALIDATION and expected_slug: expected = expected_slug.strip() if resolved_slug and resolved_slug.lower() != expected.lower(): reasons.append( @@ -224,13 +296,16 @@ def assess_canonical_repository_root( resolved_slug = repository_identity_slug(toplevel, remote=remote) reasons: list[str] = [] - if mode == "derivation": + if mode == MODE_DERIVATION: if not resolved_slug: reasons.append( f"configured canonical repository root '{toplevel}' has no resolvable " "git remote identity (fail closed)" ) else: + # #973 B10: MODE_VALIDATION only. This arm is no longer a catch-all — the + # allowlist above admits no third value, so an unsupported mode can no + # longer silently receive validation semantics here. expected = (expected_slug or "").strip() or None if expected: if resolved_slug and resolved_slug.lower() != expected.lower(): @@ -282,10 +357,19 @@ def _assessment( proven: bool, reasons: list[str], configured: bool, - canonical_repo_root: str, + canonical_repo_root: str | None, resolved_slug: str | None, source: str | None, + reason_code: str | None = None, ) -> dict: + """Build the assessment payload. + + ``canonical_repo_root`` is None only when the assessment refused to resolve + one at all — today exactly the unsupported-mode refusal (#973 B10), which + must not derive a trusted repository identity through an undefined mode. + ``reason_code`` is a machine-checkable refusal cause; None for every + ordinary (non-coded) outcome. + """ return { "proven": proven, "block": not proven, @@ -294,4 +378,5 @@ def _assessment( "canonical_repo_root": canonical_repo_root, "resolved_slug": resolved_slug, "source": source, + "reason_code": reason_code, } diff --git a/namespace_workspace_binding.py b/namespace_workspace_binding.py index a51d4c7..ae65b7f 100644 --- a/namespace_workspace_binding.py +++ b/namespace_workspace_binding.py @@ -249,7 +249,16 @@ def resolve_namespace_mutation_context( remote=remote, require_binding=True, ) - canonical_root = crr_assessment["canonical_repo_root"] + # #973 B10: a refused repository-authority mode deliberately resolves no + # canonical root, so downstream guards keep evaluating the install + # checkout rather than an identity derived through an undefined mode. + # ``roots_aligned`` still follows ``proven`` and stays False, and the + # refusal (with its reason_code) rides along in + # ``canonical_root_assessment`` — this is a fail-closed fallback, never a + # normalisation of the mode. + canonical_root = crr_assessment["canonical_repo_root"] or amw.resolve_canonical_repo_root( + process_root, process_root + ) roots_aligned = crr_assessment["proven"] else: crr_assessment = { @@ -260,6 +269,7 @@ def resolve_namespace_mutation_context( "canonical_repo_root": amw.resolve_canonical_repo_root(process_root, process_root), "resolved_slug": None, "source": None, + "reason_code": None, } canonical_root = crr_assessment["canonical_repo_root"] roots_aligned = (canonical_root == process_root) diff --git a/tests/test_issue_973_b10_mode_contract.py b/tests/test_issue_973_b10_mode_contract.py new file mode 100644 index 0000000..ef9b352 --- /dev/null +++ b/tests/test_issue_973_b10_mode_contract.py @@ -0,0 +1,739 @@ +"""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", "test@example.com"], 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()