Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 102 Pull Requests 11 Actions Packages Projects Releases Wiki Activity

fix(runtime): reject unsupported repository-authority modes (#973 B10)

Browse Source 
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]>
This commit is contained in:
sysadmin
2026-07-29 17:52:32 -04:00
co-authored by Claude Opus 4.8
parent a6c7d1491e
commit 34968475c7
3 changed files with 839 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
canonical_repository_root.py
+89 -4
View File
@@ -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,
}
namespace_workspace_binding.py
+11 -1
View File
@@ -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)
tests/test_issue_973_b10_mode_contract.py
+739
View File
@@ -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", "[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()