diff --git a/canonical_repository_root.py b/canonical_repository_root.py index 6fde057..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, @@ -132,6 +174,7 @@ def assess_canonical_repository_root( process_project_root: str, remote: str | None = None, require_binding: bool = False, + mode: str = "validation", ) -> dict: """Validate the canonical repository root binding, failing closed on forgery. @@ -140,15 +183,41 @@ def assess_canonical_repository_root( ``configured`` (whether a cross-repo binding was declared), ``resolved_slug`` and ``source``. - Without a configured binding the single-repo default is preserved: the - canonical root is derived from *process_project_root* and never blocks - (unless *require_binding* explicitly demands one). + *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. - With a configured binding the path must exist, be a git repository, and — - when *expected_slug* is known — carry a matching repository identity. A - mismatched or (when *require_binding*) unprovable identity is a forged or - conflicting binding and fails closed. + 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() @@ -168,12 +237,31 @@ def assess_canonical_repository_root( ) # Single-repo default: canonical root follows the install checkout. derived = resolve_repo_toplevel(process_root) or process_root + resolved_slug = repository_identity_slug(derived, remote=remote) + reasons: list[str] = [] + # #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( + f"canonical repository root identity mismatch: '{derived}' resolves " + f"to repository '{resolved_slug}' but expected repository identity " + f"is '{expected}' (forged or conflicting binding, fail closed)" + ) + elif not resolved_slug and require_binding: + reasons.append( + f"canonical repository root '{derived}' has no resolvable git " + f"remote identity to confirm authorization for '{expected}' " + "(fail closed)" + ) return _assessment( - proven=True, - reasons=[], + proven=not reasons, + reasons=reasons, configured=False, canonical_repo_root=derived, - resolved_slug=None, + resolved_slug=resolved_slug, source=None, ) @@ -207,19 +295,36 @@ def assess_canonical_repository_root( resolved_slug = repository_identity_slug(toplevel, remote=remote) reasons: list[str] = [] - expected = (expected_slug or "").strip() or None - if expected: - if resolved_slug and resolved_slug.lower() != expected.lower(): + + if mode == MODE_DERIVATION: + if not resolved_slug: reasons.append( - f"canonical repository root identity mismatch: '{toplevel}' resolves " - f"to repository '{resolved_slug}' but the session is authorized for " - f"'{expected}' (forged or conflicting binding, fail closed)" + f"configured canonical repository root '{toplevel}' has no resolvable " + "git remote identity (fail closed)" ) - elif not resolved_slug and require_binding: + 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(): + reasons.append( + f"canonical repository root identity mismatch: '{toplevel}' resolves " + f"to repository '{resolved_slug}' but the session is authorized for " + f"'{expected}' (forged or conflicting binding, fail closed)" + ) + elif not resolved_slug and require_binding: + reasons.append( + f"canonical repository root '{toplevel}' has no resolvable git " + f"remote identity to confirm authorization for '{expected}' " + "(fail closed)" + ) + elif require_binding: reasons.append( - f"canonical repository root '{toplevel}' has no resolvable git " - f"remote identity to confirm authorization for '{expected}' " - "(fail closed)" + f"canonical repository root '{toplevel}' has configured value " + f"'{configured_value}' but authoritative expected repository identity " + "is unprovable or missing (fail closed)" ) return _assessment( @@ -252,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, @@ -264,4 +378,5 @@ def _assessment( "canonical_repo_root": canonical_repo_root, "resolved_slug": resolved_slug, "source": source, + "reason_code": reason_code, } diff --git a/gitea_config.py b/gitea_config.py index c69553c..edea975 100644 --- a/gitea_config.py +++ b/gitea_config.py @@ -1180,6 +1180,10 @@ RECOGNIZED_GITEA_ENV_KEYS = frozenset({ "GITEA_SERVER_PROVENANCE", "GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE", + "GITEA_REVIEWER_WORKTREE", + "GITEA_MERGER_WORKTREE", + "GITEA_CANONICAL_REPOSITORY_ROOT", + "GITEA_MCP_SESSION_STATE_TTL_HOURS", "GITEA_DISABLE_KEYCHAIN", "GITEA_CONTROL_PLANE_DB", "GITEA_DB_PATH", diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 4ffaff7..602ba2b 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -500,10 +500,66 @@ def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: return workspace -def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict: +def _process_root_git_remote_url(remote_name: str) -> str | None: + """Best-effort local ``git remote get-url`` strictly inside ``PROJECT_ROOT``. + + #973 (B8): Must derive expected repository identity from an authority + independent of the candidate configured canonical root. + """ + try: + proc = subprocess.run( + ["git", "remote", "get-url", remote_name], + capture_output=True, + text=True, + cwd=PROJECT_ROOT, + ) + if proc.returncode != 0: + return None + url = (proc.stdout or "").strip() + return url or None + except Exception: + return None + + +def _resolve_expected_repository_slug( + remote: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> str | None: + """Resolve expected repository slug from session context, parameters, or process root. + + Must derive expected repository identity ONLY from trusted sources independent + of the candidate configured canonical root (#973 B8). + + Bound session context takes precedence over request parameters so request-supplied + coordinates cannot replace a bound canonical root (#741 / #973 B9). + """ + bound = session_ctx.get_session_context() or {} + b_org = bound.get("org") + b_repo = bound.get("repository") + if b_org and b_repo: + return session_ctx.format_repository_slug(b_org, b_repo) + if org and repo: + return session_ctx.format_repository_slug(org, repo) + eff_remote = remote or bound.get("remote") or _effective_remote() + parsed = remote_repo_guard.parse_org_repo_from_remote_url( + _process_root_git_remote_url(eff_remote) + ) + if parsed: + return session_ctx.format_repository_slug(parsed[0], parsed[1]) + return None + + +def _resolve_namespace_mutation_context( + worktree_path: str | None = None, + remote: str | None = None, +) -> dict: """Canonical namespace workspace + repository root for guards (#460/#510/#706/#618).""" role = _effective_workspace_role() configured_root, _source = _configured_canonical_root() + bound = session_ctx.get_session_context() or {} + eff_remote = remote or bound.get("remote") or _effective_remote() + expected_slug = _resolve_expected_repository_slug(eff_remote) return nwb.resolve_namespace_mutation_context( role_kind=role, worktree_path=worktree_path, @@ -516,9 +572,12 @@ def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dic ), profile_name=get_profile().get("profile_name"), configured_canonical_root=configured_root, + expected_slug=expected_slug, + remote=eff_remote, ) + def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict: """Backward-compatible alias for namespace workspace context.""" return _resolve_namespace_mutation_context(worktree_path) @@ -620,6 +679,9 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st inspected_root = _get_git_root(workspace) process_root = ctx["process_project_root"] canonical_root = ctx["canonical_repo_root"] + crr_assessment = ctx.get("canonical_root_assessment") or {} + resolved_slug = crr_assessment.get("resolved_slug") + expected_slug = ctx.get("expected_slug") active_root = os.path.realpath(inspected_root or workspace) if active_root == canonical_root: dirty_scope = "control checkout" @@ -649,6 +711,10 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st "workspace_healthy": not bool( ctx.get("bound_worktree_missing") or ctx.get("author_worktree_block") ), + "canonical_root_assessment": crr_assessment, + "expected_repository_slug": expected_slug, + "observed_repository_identity": resolved_slug, + "worktree_registration_result": ctx.get("in_git_worktree_list"), } if not ctx["roots_aligned"]: details["workspace_root_mismatch"] = ( @@ -658,6 +724,7 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st return details + def _format_preflight_workspace_details(details: dict) -> str: parts = [ f"MCP server process root: {details.get('mcp_server_process_root')}", @@ -930,10 +997,7 @@ def _enforce_canonical_repository_root( if not configured_value: return - bound = session_ctx.get_session_context() or {} - expected_slug = session_ctx.format_repository_slug( - bound.get("org"), bound.get("repository") - ) + expected_slug = _resolve_expected_repository_slug(remote) assessment = crr.assess_canonical_repository_root( configured_value=configured_value, source=source, @@ -1866,6 +1930,9 @@ def _verify_role_mutation_workspace( # back to the install checkout and validated Gitea-Tools/branches/ instead # of the target repository the namespace is actually bound to. _configured_root, _configured_source = _configured_canonical_root() + bound = session_ctx.get_session_context() or {} + eff_remote = remote or bound.get("remote") or _effective_remote() + expected_slug = _resolve_expected_repository_slug(eff_remote, org=org, repo=repo) assessment = nwb.assess_namespace_mutation_workspace( role_kind=role, worktree_path=worktree_path, @@ -1880,6 +1947,8 @@ def _verify_role_mutation_workspace( profile_name=get_profile().get("profile_name"), current_branch=git_state.get("current_branch"), configured_canonical_root=_configured_root, + expected_slug=expected_slug, + remote=eff_remote, ) if assessment["block"]: raise RuntimeError( @@ -2184,6 +2253,7 @@ def _canonical_repository_slug( process_project_root=PROJECT_ROOT, remote=remote, require_binding=True, + mode="derivation", ) slug = assessment.get("resolved_slug") if assessment.get("block") or not slug: @@ -3414,7 +3484,7 @@ def cleanup_in_progress_for_pr( # ── Helpers ─────────────────────────────────────────────────────────────────── -def _effective_remote(remote: str) -> str: +def _effective_remote(remote: str = "dadeschools") -> str: """If remote is the default ('dadeschools') but the active profile base_url maps to a known remote, use that remote instead.""" try: profile = get_profile() @@ -15277,22 +15347,17 @@ def _current_runtime_mode_report(refresh: bool = False) -> dict: workspace_root = None aligned = None canonical_root = None + resolved_slug = None try: ctx = _resolve_namespace_mutation_context(None) workspace_root = ctx.get("workspace_path") canonical_root = ctx.get("canonical_repo_root") - # Alignment keeps its established repository-level meaning (#615 F1): - # does this namespace target the repository the process is installed in, - # i.e. canonical_repo_root == process_project_root. It is deliberately - # NOT path equality between the task workspace and the process root -- - # the global worktree rule requires task work to live in a branches/ - # worktree, so that comparison would classify every correctly bound - # author, reviewer, and merger session as unsafe. aligned = ctx.get("roots_aligned") + crr_assessment = ctx.get("canonical_root_assessment") or {} + resolved_slug = crr_assessment.get("resolved_slug") except Exception: - # An unresolvable binding is reported as unknown alignment, never as - # proof of alignment. aligned = None + resolved_slug = None try: profile_name = get_profile()["profile_name"] except Exception: @@ -15305,6 +15370,7 @@ def _current_runtime_mode_report(refresh: bool = False) -> dict: dirty_files=dirty_files, active_task_workspace=workspace_root, canonical_repository_root=canonical_root, + repository_slug=resolved_slug, workspace_roots_aligned=aligned, profile=profile_name, declared_mode=stable_control_runtime.declared_runtime_mode(), diff --git a/namespace_workspace_binding.py b/namespace_workspace_binding.py index be2ca82..ae65b7f 100644 --- a/namespace_workspace_binding.py +++ b/namespace_workspace_binding.py @@ -8,8 +8,10 @@ poison workspace purity checks in another namespace. from __future__ import annotations import os +import subprocess import author_mutation_worktree as amw +import canonical_repository_root as crr ACTIVE_WORKTREE_ENV = amw.ACTIVE_WORKTREE_ENV AUTHOR_WORKTREE_ENV = amw.AUTHOR_WORKTREE_ENV @@ -152,6 +154,60 @@ def resolve_namespace_workspace( return os.path.realpath(process_project_root), "MCP server process root (default)" +def verify_git_common_directory_membership( + workspace_path: str, + canonical_repo_root: str, +) -> tuple[bool, str | None]: + """Verify that workspace_path belongs to canonical_repo_root via git common-dir or branches containment.""" + ws = (workspace_path or "").strip() + root = (canonical_repo_root or "").strip() + if not ws or not root: + return False, "empty workspace or canonical root path" + + try: + real_ws = os.path.realpath(os.path.abspath(ws)) + real_root = os.path.realpath(os.path.abspath(root)) + except Exception as exc: + return False, f"invalid workspace or root path: {exc}" + + if real_ws == real_root: + return True, None + + if not os.path.isdir(real_ws): + return False, f"workspace directory '{real_ws}' does not exist" + + try: + res = subprocess.run( + ["git", "-C", real_ws, "rev-parse", "--git-common-dir"], + capture_output=True, + text=True, + check=False, + ) + if res.returncode == 0: + common_raw = (res.stdout or "").strip() + common_dir = amw._realpath_git_common_dir(real_ws, common_raw) + real_common = os.path.realpath(common_dir) + canonical_git = os.path.realpath(os.path.join(real_root, ".git")) + + if real_common in (canonical_git, real_root): + return True, None + return ( + False, + f"workspace '{real_ws}' git common directory '{real_common}' does not match " + f"canonical repository root '{real_root}' (.git at '{canonical_git}')" + ) + else: + return ( + False, + f"workspace '{real_ws}' is not a valid git repository or git rev-parse failed" + ) + except Exception as exc: + return ( + False, + f"failed to inspect git common directory for workspace '{real_ws}': {exc}" + ) + + def resolve_namespace_mutation_context( *, role_kind: str, @@ -163,6 +219,8 @@ def resolve_namespace_mutation_context( worktree: str | None = None, profile_name: str | None = None, configured_canonical_root: str | None = None, + expected_slug: str | None = None, + remote: str | None = None, ) -> dict: """Shared workspace resolution for runtime_context and mutation guards. @@ -180,11 +238,41 @@ def resolve_namespace_mutation_context( env_map = env if env is not None else os.environ process_root = os.path.realpath(process_project_root) role = normalize_role_kind(role_kind, profile_name=profile_name) - configured = (configured_canonical_root or "").strip() - if configured: - canonical_root = os.path.realpath(configured) + + configured_val = (configured_canonical_root or "").strip() + if configured_val: + crr_assessment = crr.assess_canonical_repository_root( + configured_value=configured_val, + source="configured_canonical_root", + expected_slug=expected_slug, + process_project_root=process_root, + remote=remote, + require_binding=True, + ) + # #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: - canonical_root = amw.resolve_canonical_repo_root(process_root, process_root) + crr_assessment = { + "proven": True, + "block": False, + "reasons": [], + "configured": False, + "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) durable: dict | None = None if role == "author": @@ -234,7 +322,10 @@ def resolve_namespace_mutation_context( "ignored_bindings": demotions + (pollution.get("ignored_bindings") or []), "process_project_root": process_root, "canonical_repo_root": canonical_root, - "roots_aligned": canonical_root == process_root, + "roots_aligned": roots_aligned, + "canonical_root_assessment": crr_assessment, + "expected_slug": expected_slug, + "remote": remote, } if durable is not None: result["author_worktree_resolution"] = durable @@ -246,6 +337,15 @@ def resolve_namespace_mutation_context( result["author_worktree_reasons"] = list(durable.get("reasons") or []) result["author_worktree_blocker_kind"] = durable.get("blocker_kind") result["operator_recovery"] = durable.get("operator_recovery") + else: + path_exists = os.path.exists(workspace) + result["path_exists"] = path_exists + result["in_git_worktree_list"] = ( + amw.path_in_git_worktree_list(workspace, canonical_root) + if path_exists + else False + ) + result["bound_worktree_missing"] = not path_exists return result @@ -378,8 +478,8 @@ def format_namespace_workspace_binding_error( def assess_namespace_mutation_workspace( *, role_kind: str, - worktree_path: str | None, - worktree: str | None, + worktree_path: str | None = None, + worktree: str | None = None, process_project_root: str, env: dict[str, str] | os._Environ | None = None, session_lease_worktree: str | None = None, @@ -387,6 +487,8 @@ def assess_namespace_mutation_workspace( profile_name: str | None = None, current_branch: str | None = None, configured_canonical_root: str | None = None, + expected_slug: str | None = None, + remote: str | None = None, ) -> dict: """Evaluate namespace workspace binding before preflight/mutation.""" ctx = resolve_namespace_mutation_context( @@ -399,6 +501,8 @@ def assess_namespace_mutation_workspace( session_lock_worktree=session_lock_worktree, profile_name=profile_name, configured_canonical_root=configured_canonical_root, + expected_slug=expected_slug, + remote=remote, ) mutation_workspace = ctx["workspace_path"] binding_source = ctx["workspace_binding_source"] @@ -422,6 +526,27 @@ def assess_namespace_mutation_workspace( reasons = list(metadata.get("reasons") or []) operator_recovery = ctx.get("operator_recovery") + + crr_reasons = list(ctx.get("canonical_root_assessment", {}).get("reasons") or []) + if crr_reasons: + reasons.extend(crr_reasons) + + path_exists = ctx.get("path_exists") + if path_exists is None: + path_exists = os.path.exists(mutation_workspace) + + if not path_exists: + if role != "author": + reasons.append( + f"{role} mutation blocked: configured workspace directory '{mutation_workspace}' does not exist (nonexistent worktree)" + ) + else: + valid_common, common_err = verify_git_common_directory_membership( + mutation_workspace, ctx["canonical_repo_root"] + ) + if not valid_common and common_err: + reasons.append(common_err) + if role == "author": # #618 durable resolution already validated existence, membership, # branches/, lock ownership, and traversal safety when present. @@ -438,26 +563,27 @@ def assess_namespace_mutation_workspace( ) if branches["block"]: reasons.extend(branches["reasons"]) - elif ( - role == "reviewer" - and mutation_workspace == process_root - and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]) - ): - reasons.append( - f"{role} mutation blocked: workspace is the stable control checkout; " - f"create or reconnect to a session-owned worktree under branches/ " - f"or set {ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} / " - f"{ACTIVE_WORKTREE_ENV}" - ) - elif ( - role in {"reviewer", "merger"} - and mutation_workspace != process_root - and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]) - ): - reasons.append( - f"{role} mutation blocked: workspace '{mutation_workspace}' is not under " - f"'{ctx['canonical_repo_root']}/branches/'" - ) + elif role in {"reviewer", "merger"}: + if mutation_workspace == process_root and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]): + reasons.append( + f"{role} mutation blocked: workspace is the stable control checkout; " + f"create or reconnect to a session-owned worktree under branches/ " + f"or set {ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} / " + f"{ACTIVE_WORKTREE_ENV}" + ) + elif mutation_workspace != process_root and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]): + reasons.append( + f"{role} mutation blocked: workspace '{mutation_workspace}' is not under " + f"'{ctx['canonical_repo_root']}/branches/'" + ) + + if path_exists and amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]): + in_list = ctx.get("in_git_worktree_list") + if in_list is False: + reasons.append( + f"{role} mutation blocked: workspace '{mutation_workspace}' is under branches/ " + f"but is not registered in git worktree list for '{ctx['canonical_repo_root']}'" + ) block = bool(reasons) return { diff --git a/tests/test_cross_repo_canonical_consumption.py b/tests/test_cross_repo_canonical_consumption.py index 2eaee02..c397da0 100644 --- a/tests/test_cross_repo_canonical_consumption.py +++ b/tests/test_cross_repo_canonical_consumption.py @@ -313,6 +313,7 @@ class TestCanonicalRootGuardBinding(_ServerHarness): process_project_root=self.install_root, remote="prgs", require_binding=True, + mode="derivation", ) self.assertFalse(got.get("block"), got.get("reasons")) self.assertEqual(got["resolved_slug"], TARGET_SLUG) diff --git a/tests/test_issue_706_canonical_repository_root.py b/tests/test_issue_706_canonical_repository_root.py index a51004f..2a5c4cb 100644 --- a/tests/test_issue_706_canonical_repository_root.py +++ b/tests/test_issue_706_canonical_repository_root.py @@ -241,9 +241,10 @@ class TestNamespaceContextUsesConfiguredRoot(unittest.TestCase): process_project_root=self.install, env={}, configured_canonical_root=self.target, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", ) self.assertEqual(ctx["canonical_repo_root"], self.target) - self.assertFalse(ctx["roots_aligned"]) + self.assertTrue(ctx["roots_aligned"]) def test_target_worktree_is_member_of_target_root(self): got = nwb.amw.assess_workspace_repo_membership( @@ -268,6 +269,7 @@ class TestNamespaceContextUsesConfiguredRoot(unittest.TestCase): env={}, current_branch="feat/issue-1", configured_canonical_root=self.target, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", ) self.assertFalse(assessment["block"], assessment.get("reasons")) self.assertEqual(assessment["canonical_repo_root"], self.target) 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() diff --git a/tests/test_issue_973_cross_repo_canonical_roots.py b/tests/test_issue_973_cross_repo_canonical_roots.py new file mode 100644 index 0000000..a854619 --- /dev/null +++ b/tests/test_issue_973_cross_repo_canonical_roots.py @@ -0,0 +1,546 @@ +"""Regression tests for Issue #973: validated cross-repository canonical roots.""" + +from __future__ import annotations + +import os +import shutil +import tempfile +import unittest +from unittest.mock import patch, MagicMock +from pathlib import Path +import subprocess + +import gitea_config +import namespace_workspace_binding as nwb +import canonical_repository_root as crr +import stable_control_runtime +import gitea_mcp_server as mcp_server + + +class TestIssue973RecognizedEnvKeys(unittest.TestCase): + """Test recognized environment variable keys under #973.""" + + def test_recognized_gitea_env_keys(self): + for key in ( + "GITEA_CANONICAL_REPOSITORY_ROOT", + "GITEA_REVIEWER_WORKTREE", + "GITEA_MERGER_WORKTREE", + "GITEA_MCP_SESSION_STATE_TTL_HOURS", + ): + self.assertIn(key, gitea_config.RECOGNIZED_GITEA_ENV_KEYS) + + def test_get_unconsumed_gitea_env_overrides_ignores_recognized(self): + env = { + "GITEA_CANONICAL_REPOSITORY_ROOT": "/some/path", + "GITEA_REVIEWER_WORKTREE": "/some/reviewer/path", + "GITEA_MERGER_WORKTREE": "/some/merger/path", + "GITEA_MCP_SESSION_STATE_TTL_HOURS": "24", + "GITEA_UNRECOGNIZED_FOO_VAR": "bar", + } + unconsumed = gitea_config.get_unconsumed_gitea_env_overrides(env) + self.assertNotIn("GITEA_CANONICAL_REPOSITORY_ROOT", unconsumed) + self.assertNotIn("GITEA_REVIEWER_WORKTREE", unconsumed) + self.assertNotIn("GITEA_MERGER_WORKTREE", unconsumed) + self.assertNotIn("GITEA_MCP_SESSION_STATE_TTL_HOURS", unconsumed) + self.assertIn("GITEA_UNRECOGNIZED_FOO_VAR", unconsumed) + + +class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase): + """Test workspace binding and canonical root validation for cross-repo namespaces.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.tmp_dir = os.path.realpath(self._tmp.name) + + # Create simulated installation root + self.install_root = os.path.join(self.tmp_dir, "Gitea-Tools") + os.makedirs(self.install_root) + subprocess.run(["git", "init", "-b", "master"], cwd=self.install_root, check=True) + subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=self.install_root, check=True) + subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.install_root, check=True) + with open(os.path.join(self.install_root, "README.md"), "w") as f: + f.write("install\n") + subprocess.run(["git", "add", "README.md"], cwd=self.install_root, check=True) + subprocess.run(["git", "commit", "-m", "initial"], cwd=self.install_root, check=True) + + # Create simulated target repository root + self.target_root = os.path.join(self.tmp_dir, "mcp-control-plane") + os.makedirs(self.target_root) + subprocess.run(["git", "init", "-b", "master"], cwd=self.target_root, check=True) + subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=self.target_root, check=True) + subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.target_root, check=True) + with open(os.path.join(self.target_root, "README.md"), "w") as f: + f.write("target\n") + subprocess.run(["git", "add", "README.md"], cwd=self.target_root, check=True) + subprocess.run(["git", "commit", "-m", "initial"], cwd=self.target_root, check=True) + + # Add remotes to simulate real git repositories with identities + subprocess.run(["git", "remote", "add", "prgs", "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"], cwd=self.install_root, check=True) + subprocess.run(["git", "remote", "add", "prgs", "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git"], cwd=self.target_root, check=True) + + # Create simulated foreign repository root + self.evil_root = os.path.join(self.tmp_dir, "Evil-Repo") + os.makedirs(self.evil_root) + subprocess.run(["git", "init", "-b", "master"], cwd=self.evil_root, check=True) + subprocess.run(["git", "config", "user.email", "evil@example.com"], cwd=self.evil_root, check=True) + subprocess.run(["git", "config", "user.name", "Evil User"], cwd=self.evil_root, check=True) + with open(os.path.join(self.evil_root, "README.md"), "w") as f: + f.write("evil\n") + subprocess.run(["git", "add", "README.md"], cwd=self.evil_root, check=True) + subprocess.run(["git", "commit", "-m", "initial"], cwd=self.evil_root, check=True) + subprocess.run(["git", "remote", "add", "prgs", "https://gitea.prgs.cc/Someone-Else/Evil-Repo.git"], cwd=self.evil_root, check=True) + + # Create branches/ directory and a valid registered worktree in target repository + 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) + + def tearDown(self): + self._tmp.cleanup() + + def test_valid_same_repository_configuration(self): + ctx = nwb.resolve_namespace_mutation_context( + role_kind="reviewer", + worktree_path=None, + process_project_root=self.install_root, + env={}, + configured_canonical_root=None, + ) + self.assertEqual(ctx["canonical_repo_root"], self.install_root) + self.assertTrue(ctx["roots_aligned"]) + self.assertTrue(ctx["canonical_root_assessment"]["proven"]) + + def test_valid_cross_repo_canonical_root(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="Scaled-Tech-Consulting/mcp-control-plane", + ) + self.assertEqual(ctx["canonical_repo_root"], self.target_root) + self.assertTrue(ctx["roots_aligned"]) + self.assertTrue(ctx["canonical_root_assessment"]["proven"]) + + def test_expected_repository_identity_match(self): + assessment = crr.assess_canonical_repository_root( + configured_value=self.target_root, + source="test", + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", + process_project_root=self.install_root, + remote="prgs", + ) + self.assertTrue(assessment["proven"]) + self.assertFalse(assessment["block"]) + + def test_foreign_repository_identity_mismatch(self): + assessment = crr.assess_canonical_repository_root( + configured_value=self.evil_root, + source="test", + expected_slug="Scaled-Tech-Consulting/Gitea-Tools", + process_project_root=self.install_root, + remote="prgs", + ) + self.assertFalse(assessment["proven"]) + self.assertTrue(assessment["block"]) + self.assertTrue(any("identity mismatch" in r for r in assessment["reasons"])) + + def test_native_repository_binding_mismatch(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.evil_root, + expected_slug="Scaled-Tech-Consulting/Gitea-Tools", + remote="prgs", + ) + self.assertFalse(ctx["roots_aligned"]) + self.assertFalse(ctx["canonical_root_assessment"]["proven"]) + self.assertTrue(any("identity mismatch" in r for r in ctx["canonical_root_assessment"]["reasons"])) + + def test_unpatched_foreign_configured_root_derives_expected_from_process_root_and_blocks(self): + """B8: Production path test where foreign configured root cannot self-authorize.""" + with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \ + patch.object(mcp_server, "_configured_canonical_root", return_value=(self.evil_root, "env")): + expected_slug = mcp_server._resolve_expected_repository_slug("prgs") + self.assertEqual(expected_slug, "Scaled-Tech-Consulting/Gitea-Tools") + assessment = crr.assess_canonical_repository_root( + configured_value=self.evil_root, + source="env", + expected_slug=expected_slug, + process_project_root=self.install_root, + remote="prgs", + require_binding=True, + ) + self.assertFalse(assessment["proven"]) + self.assertTrue(assessment["block"]) + self.assertEqual(assessment["resolved_slug"], "Someone-Else/Evil-Repo") + self.assertTrue(any("identity mismatch" in r for r in assessment["reasons"])) + + def test_unpatched_valid_cross_repo_matching_session_context(self): + """B8: Valid cross-repo namespace matches when session context is bound to target repo.""" + bound_ctx = {"org": "Scaled-Tech-Consulting", "repository": "mcp-control-plane", "remote": "prgs"} + with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \ + patch.object(mcp_server.session_ctx, "get_session_context", return_value=bound_ctx): + expected_slug = mcp_server._resolve_expected_repository_slug("prgs") + self.assertEqual(expected_slug, "Scaled-Tech-Consulting/mcp-control-plane") + assessment = crr.assess_canonical_repository_root( + configured_value=self.target_root, + source="env", + expected_slug=expected_slug, + process_project_root=self.install_root, + remote="prgs", + require_binding=True, + ) + self.assertTrue(assessment["proven"]) + self.assertFalse(assessment["block"]) + self.assertEqual(assessment["resolved_slug"], "Scaled-Tech-Consulting/mcp-control-plane") + + def test_unprovable_expected_identity_fails_closed(self): + """B8: If expected repository identity is unprovable for a configured root, fail closed.""" + no_remote_root = os.path.join(self.tmp_dir, "no-remote-process-root") + os.makedirs(no_remote_root) + subprocess.run(["git", "init", "-b", "master"], cwd=no_remote_root, check=True) + with patch.object(mcp_server, "PROJECT_ROOT", no_remote_root), \ + patch.object(mcp_server.session_ctx, "get_session_context", return_value=None): + expected_slug = mcp_server._resolve_expected_repository_slug("prgs") + self.assertIsNone(expected_slug) + assessment = crr.assess_canonical_repository_root( + configured_value=self.target_root, + source="env", + expected_slug=expected_slug, + process_project_root=no_remote_root, + remote="prgs", + require_binding=True, + ) + self.assertFalse(assessment["proven"]) + self.assertTrue(assessment["block"]) + self.assertTrue(any("unprovable or missing" in r for r in assessment["reasons"])) + + def test_missing_canonical_root(self): + ctx = nwb.resolve_namespace_mutation_context( + role_kind="author", + worktree_path=None, + process_project_root=self.install_root, + env={}, + configured_canonical_root="", + ) + self.assertEqual(ctx["canonical_repo_root"], self.install_root) + self.assertTrue(ctx["roots_aligned"]) + + def test_nonexistent_configured_canonical_root(self): + nonexistent = os.path.join(self.tmp_dir, "nonexistent-repo") + ctx = nwb.resolve_namespace_mutation_context( + role_kind="reviewer", + worktree_path=self.target_worktree, + process_project_root=self.install_root, + env={}, + configured_canonical_root=nonexistent, + ) + self.assertFalse(ctx["roots_aligned"]) + self.assertFalse(ctx["canonical_root_assessment"]["proven"]) + self.assertTrue(any("does not exist" in r for r in ctx["canonical_root_assessment"]["reasons"])) + + def test_non_git_configured_canonical_root(self): + non_git = os.path.join(self.tmp_dir, "non-git-dir") + os.makedirs(non_git) + ctx = nwb.resolve_namespace_mutation_context( + role_kind="reviewer", + worktree_path=self.target_worktree, + process_project_root=self.install_root, + env={}, + configured_canonical_root=non_git, + ) + self.assertFalse(ctx["roots_aligned"]) + self.assertFalse(ctx["canonical_root_assessment"]["proven"]) + self.assertTrue(any("not a git repository" in r for r in ctx["canonical_root_assessment"]["reasons"])) + + def test_git_common_directory_membership_matching(self): + valid, err = nwb.verify_git_common_directory_membership( + self.target_worktree, self.target_root + ) + self.assertTrue(valid, err) + self.assertIsNone(err) + + def test_foreign_git_common_directory(self): + # Foreign worktree created under install_root + install_branches = os.path.join(self.install_root, "branches") + foreign_wt = os.path.join(install_branches, "foreign-wt") + subprocess.run(["git", "worktree", "add", "-b", "foreign-wt", foreign_wt], cwd=self.install_root, check=True) + + valid, err = nwb.verify_git_common_directory_membership( + foreign_wt, self.target_root + ) + self.assertFalse(valid) + self.assertIn("does not match", err) + + def test_normalized_path_aliases(self): + alias_path = self.target_worktree + "/../rev-pr-99/./" + valid, err = nwb.verify_git_common_directory_membership( + alias_path, self.target_root + ) + self.assertTrue(valid, err) + + def test_safe_symlink_identity(self): + link_path = os.path.join(self.target_branches, "symlink-rev-99") + try: + os.symlink(self.target_worktree, link_path) + valid, err = nwb.verify_git_common_directory_membership( + link_path, self.target_root + ) + self.assertTrue(valid, err) + finally: + if os.path.exists(link_path): + os.unlink(link_path) + + def test_symlink_escape_or_foreign_alias(self): + outside_dir = os.path.join(self.tmp_dir, "outside-target") + os.makedirs(outside_dir) + link_escape = os.path.join(self.target_branches, "escape-link") + try: + os.symlink(outside_dir, link_escape) + assessment = nwb.assess_namespace_mutation_workspace( + role_kind="reviewer", + worktree_path=link_escape, + worktree=None, + process_project_root=self.install_root, + env={}, + configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", + ) + self.assertTrue(assessment["block"]) + finally: + if os.path.exists(link_escape): + os.unlink(link_escape) + + def test_reviewer_worktree_registered_and_valid(self): + assessment = nwb.assess_namespace_mutation_workspace( + role_kind="reviewer", + worktree_path=self.target_worktree, + worktree=None, + process_project_root=self.install_root, + env={}, + configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", + ) + self.assertFalse(assessment["block"]) + + def test_reviewer_worktree_unregistered_blocks(self): + unreg_wt = os.path.join(self.target_branches, "unregistered-reviewer") + os.makedirs(unreg_wt) + assessment = nwb.assess_namespace_mutation_workspace( + role_kind="reviewer", + worktree_path=unreg_wt, + worktree=None, + process_project_root=self.install_root, + env={}, + configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", + ) + self.assertTrue(assessment["block"]) + self.assertTrue(any("is not registered in git worktree list" in r for r in assessment["reasons"])) + + def test_merger_worktree_registered_and_valid(self): + assessment = nwb.assess_namespace_mutation_workspace( + role_kind="merger", + worktree_path=self.target_worktree, + worktree=None, + process_project_root=self.install_root, + env={}, + configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", + ) + self.assertFalse(assessment["block"]) + + def test_merger_worktree_unregistered_blocks(self): + unreg_wt = os.path.join(self.target_branches, "unregistered-merger") + os.makedirs(unreg_wt) + assessment = nwb.assess_namespace_mutation_workspace( + role_kind="merger", + worktree_path=unreg_wt, + worktree=None, + process_project_root=self.install_root, + env={}, + configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", + ) + self.assertTrue(assessment["block"]) + self.assertTrue(any("is not registered in git worktree list" in r for r in assessment["reasons"])) + + def test_reviewer_or_merger_worktree_outside_branches_blocks(self): + outside_wt = os.path.join(self.target_root, "outside_branches_wt") + os.makedirs(outside_wt) + for r_kind in ("reviewer", "merger"): + assessment = nwb.assess_namespace_mutation_workspace( + role_kind=r_kind, + worktree_path=outside_wt, + worktree=None, + process_project_root=self.install_root, + env={}, + configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", + ) + self.assertTrue(assessment["block"]) + self.assertTrue(any("is not under" in r for r in assessment["reasons"])) + + def test_nonexistent_reviewer_or_merger_worktree_blocks(self): + nonexistent_wt = os.path.join(self.target_branches, "nonexistent-wt") + for r_kind in ("reviewer", "merger"): + assessment = nwb.assess_namespace_mutation_workspace( + role_kind=r_kind, + worktree_path=nonexistent_wt, + worktree=None, + process_project_root=self.install_root, + env={}, + configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", + ) + self.assertTrue(assessment["block"]) + self.assertTrue(any("does not exist" in r for r in assessment["reasons"])) + + def test_safe_and_unsafe_mutation_alignment_outcomes(self): + # Safe alignment (same repo) + report_safe = stable_control_runtime.build_runtime_report( + process_root=self.install_root, + checkout_branch="master", + runtime_head="abcdef123456", + active_task_workspace=self.install_root, + canonical_repository_root=self.install_root, + workspace_roots_aligned=True, + ) + gate_safe = stable_control_runtime.assess_runtime_mutation_gate(report_safe) + self.assertFalse(gate_safe["block"]) + + # Unsafe alignment + report_unsafe = stable_control_runtime.build_runtime_report( + process_root=self.install_root, + checkout_branch="master", + runtime_head="abcdef123456", + active_task_workspace=self.target_worktree, + canonical_repository_root=self.target_root, + workspace_roots_aligned=False, + ) + gate_unsafe = stable_control_runtime.assess_runtime_mutation_gate(report_unsafe) + self.assertTrue(gate_unsafe["block"]) + + def test_reviewer_lease_lifecycle_production_path(self): + """B5: Automated regression for reviewer lease acquire and release through production path.""" + mock_whoami = { + "authenticated": True, + "username": "sysadmin", + "remote": "prgs", + "profile": { + "profile_name": "prgs-reviewer", + "role": "reviewer", + "role_kind": "reviewer", + "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.approve", "gitea.pr.request_changes"], + "forbidden_operations": [], + }, + } + + def mock_api_request(method, url, auth=None, json_data=None): + if method == "GET": + return {"number": 99, "head": {"sha": "abc1234"}, "state": "open", "merged": False, "merged_at": None} + elif method == "POST": + return {"id": 9999, "body": (json_data or {}).get("body", "")} + return {} + + with patch.object(mcp_server, "gitea_whoami", return_value=mock_whoami), \ + patch.object(mcp_server, "get_profile", return_value=mock_whoami["profile"]), \ + patch.object(mcp_server, "_effective_workspace_role", return_value="reviewer"), \ + patch.object(mcp_server, "_configured_canonical_root", return_value=(self.target_root, "env")), \ + patch.object(mcp_server, "_reviewer_session_worktree", return_value=self.target_worktree), \ + patch.object(mcp_server, "_auth", return_value="token mock-token"), \ + patch.object(mcp_server, "_fetch_pr_comments", return_value=[]), \ + patch.object(mcp_server, "api_request", side_effect=mock_api_request), \ + patch("reviewer_pr_lease.assess_acquire_lease", return_value={"acquire_allowed": True, "reasons": [], "lease_body": ""}), \ + patch("reviewer_pr_lease.find_active_reviewer_lease", return_value={"session_id": "sid-123", "reviewer": "sysadmin"}), \ + patch("reviewer_pr_lease.get_session_lease", return_value={"session_id": "sid-123", "reviewer": "sysadmin"}), \ + patch("reviewer_pr_lease.clear_session_lease") as mock_clear: + + acq_res = mcp_server.gitea_acquire_reviewer_pr_lease( + pr_number=99, + remote="prgs", + worktree=self.target_worktree, + org="Scaled-Tech-Consulting", + repo="mcp-control-plane", + ) + self.assertTrue(acq_res.get("success"), acq_res) + + rel_res = mcp_server.gitea_release_reviewer_pr_lease( + pr_number=99, + worktree=self.target_worktree, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="mcp-control-plane", + ) + self.assertTrue(rel_res.get("success"), rel_res) + mock_clear.assert_called_once() + + def test_unbound_session_derives_and_retains_configured_target_repository(self): + """B9: Unbound session with a legitimate configured target root derives and retains target repository.""" + prof = { + "profile_name": "prgs-author", + "allowed_repositories": ["Scaled-Tech-Consulting/mcp-control-plane"], + "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(prof, "prgs") + self.assertEqual(slug, "Scaled-Tech-Consulting/mcp-control-plane", f"reasons: {reasons}") + self.assertEqual(reasons, []) + res = mcp_server._trusted_session_repository(prof, "prgs") + self.assertEqual(res["org"], "Scaled-Tech-Consulting") + self.assertEqual(res["repository"], "mcp-control-plane") + + def test_process_root_a_configured_target_b_retains_b(self): + """B9: Process root A (Gitea-Tools) and configured target B (mcp-control-plane) keep B as expected identity when bound.""" + bound_ctx = {"org": "Scaled-Tech-Consulting", "repository": "mcp-control-plane", "remote": "prgs"} + with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \ + patch.object(mcp_server.session_ctx, "get_session_context", return_value=bound_ctx): + expected = mcp_server._resolve_expected_repository_slug("prgs") + self.assertEqual(expected, "Scaled-Tech-Consulting/mcp-control-plane") + + def test_request_supplied_repository_c_cannot_replace_derived_or_bound_repository_b(self): + """B9: Request-supplied repository C (Timesheet) cannot replace bound repository B (mcp-control-plane).""" + bound_ctx = {"org": "Scaled-Tech-Consulting", "repository": "mcp-control-plane", "remote": "prgs"} + with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \ + patch.object(mcp_server.session_ctx, "get_session_context", return_value=bound_ctx): + expected = mcp_server._resolve_expected_repository_slug("prgs", org="Scaled-Tech-Consulting", repo="Timesheet") + self.assertEqual(expected, "Scaled-Tech-Consulting/mcp-control-plane") + + def test_known_expected_repo_a_plus_candidate_root_b_fails_closed(self): + """B9: Known expected repository A plus candidate root B fails closed in validation mode.""" + assessment = crr.assess_canonical_repository_root( + configured_value=self.target_root, + source="env", + expected_slug="Scaled-Tech-Consulting/Gitea-Tools", + process_project_root=self.install_root, + remote="prgs", + require_binding=True, + mode="validation", + ) + self.assertFalse(assessment["proven"]) + self.assertTrue(assessment["block"]) + self.assertTrue(any("identity mismatch" in r for r in assessment["reasons"])) + + def test_validation_mode_with_unprovable_expected_identity_fails_closed(self): + """B9: Validation mode with expected_slug=None and require_binding=True fails closed.""" + 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="validation", + ) + self.assertFalse(assessment["proven"]) + self.assertTrue(assessment["block"]) + self.assertTrue(any("unprovable or missing" in r for r in assessment["reasons"])) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_namespace_workspace_binding.py b/tests/test_namespace_workspace_binding.py index 46e0b19..fedfe4e 100644 --- a/tests/test_namespace_workspace_binding.py +++ b/tests/test_namespace_workspace_binding.py @@ -245,10 +245,15 @@ class TestNamespaceWorkspaceIntegration(unittest.TestCase): def test_pr487_style_merge_binds_clean_merger_workspace( self, _exists, _isdir, mock_run ): - mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") + def mock_git(cmd, *args, **kwargs): + if "rev-parse" in cmd: + return MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") + return MagicMock(returncode=0, stdout=f"worktree {CONTROL_ROOT}\nworktree {MERGER_CLEAN}\n") + mock_run.side_effect = mock_git os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY + os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN srv._preflight_resolved_role = "reviewer" with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()): resolved = srv._verify_role_mutation_workspace("prgs") - self.assertEqual(resolved, os.path.realpath(MCP_PROCESS_ROOT)) \ No newline at end of file + self.assertEqual(resolved, os.path.realpath(MERGER_CLEAN)) \ No newline at end of file