Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
15a8a76e99 | ||
|
|
a8d2087b4a | ||
|
|
61c0d73cd1 | ||
|
|
6a0d7bbef4 | ||
|
|
29d96c8946 | ||
|
|
8d8d2d8f81 | ||
|
|
0d8a2c2b1d | ||
|
|
11d1d2e99f |
@@ -0,0 +1,267 @@
|
||||
"""Immutable canonical repository root for cross-repository namespaces (#706).
|
||||
|
||||
The Gitea-Tools MCP server historically derived the ``canonical_repo_root`` from
|
||||
the *install checkout* the server script lives in
|
||||
(``PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))``). A namespace
|
||||
that runs the same server script against an *external* repository (e.g.
|
||||
``eagenda-author`` targeting ``eAgenda``) then failed every mutation: the
|
||||
branches-only / worktree-membership guards (#274) compared the task workspace
|
||||
against the Gitea-Tools ``.git`` directory, which it can never belong to.
|
||||
|
||||
This module separates two distinct concepts:
|
||||
|
||||
* the immutable code/install root (``PROJECT_ROOT``) — where the server lives, and
|
||||
* the namespace-scoped **canonical repository root** — the working root of the
|
||||
repository whose issues/PRs the namespace mutates.
|
||||
|
||||
The canonical repository root is configured per namespace (profile field or an
|
||||
environment variable, typically set alongside the namespace ``cwd`` in the MCP
|
||||
config). It is validated (existence, git identity, git common-directory
|
||||
membership) and pinned immutably into the session context so a later call cannot
|
||||
forge or swap it. When *no* binding is configured the single-repo default is
|
||||
preserved unchanged: the canonical root is derived from the process checkout.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
from typing import Mapping
|
||||
|
||||
import remote_repo_guard
|
||||
|
||||
# Namespace-scoped override, typically exported next to the server ``cwd`` in the
|
||||
# MCP config for a cross-repository namespace.
|
||||
CANONICAL_ROOT_ENV = "GITEA_CANONICAL_REPOSITORY_ROOT"
|
||||
|
||||
# Candidate git remote names probed when deriving repository identity.
|
||||
_IDENTITY_REMOTE_CANDIDATES = ("prgs", "origin", "dadeschools", "mdcps")
|
||||
|
||||
|
||||
def configured_canonical_root(
|
||||
profile: Mapping | None,
|
||||
env: Mapping | None,
|
||||
) -> tuple[str | None, str | None]:
|
||||
"""Return ``(value, source)`` for the declared canonical repository root.
|
||||
|
||||
Precedence: the ``GITEA_CANONICAL_REPOSITORY_ROOT`` environment variable
|
||||
(namespace-scoped) overrides the profile ``canonical_repository_root``
|
||||
field. Blank values are treated as unset. Returns ``(None, None)`` when no
|
||||
binding is declared (the single-repo default).
|
||||
"""
|
||||
env_map = env if env is not None else os.environ
|
||||
env_val = (env_map.get(CANONICAL_ROOT_ENV) or "").strip()
|
||||
if env_val:
|
||||
return env_val, f"{CANONICAL_ROOT_ENV} environment variable"
|
||||
if profile:
|
||||
prof_val = (profile.get("canonical_repository_root") or "").strip()
|
||||
if prof_val:
|
||||
return prof_val, "profile canonical_repository_root"
|
||||
return None, None
|
||||
|
||||
|
||||
def resolve_repo_toplevel(path: str) -> str | None:
|
||||
"""Realpath of the git working-tree top level for *path*, or None."""
|
||||
text = (path or "").strip()
|
||||
if not text:
|
||||
return None
|
||||
try:
|
||||
res = subprocess.run(
|
||||
["git", "-C", text, "rev-parse", "--show-toplevel"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
except Exception:
|
||||
return None
|
||||
top = (res.stdout or "").strip()
|
||||
return os.path.realpath(top) if top else None
|
||||
|
||||
|
||||
def repository_identity_slug(path: str, *, remote: str | None = None) -> str | None:
|
||||
"""``owner/repository`` derived from a git remote configured at *path*.
|
||||
|
||||
Tries the caller-named remote first, then a small set of known remote names,
|
||||
then whatever remote the repository actually has. Returns None when no remote
|
||||
URL is parseable (identity cannot be proven).
|
||||
"""
|
||||
text = (path or "").strip()
|
||||
if not text:
|
||||
return None
|
||||
|
||||
ordered: list[str] = []
|
||||
for name in (remote, *_IDENTITY_REMOTE_CANDIDATES):
|
||||
clean = (name or "").strip()
|
||||
if clean and clean not in ordered:
|
||||
ordered.append(clean)
|
||||
|
||||
try:
|
||||
listed = subprocess.run(
|
||||
["git", "-C", text, "remote"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
).stdout.split()
|
||||
except Exception:
|
||||
listed = []
|
||||
for name in listed:
|
||||
if name and name not in ordered:
|
||||
ordered.append(name)
|
||||
|
||||
for name in ordered:
|
||||
try:
|
||||
url = subprocess.run(
|
||||
["git", "-C", text, "remote", "get-url", name],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
).stdout.strip()
|
||||
except Exception:
|
||||
continue
|
||||
parsed = remote_repo_guard.parse_org_repo_from_remote_url(url)
|
||||
if parsed:
|
||||
return f"{parsed[0]}/{parsed[1]}"
|
||||
return None
|
||||
|
||||
|
||||
def assess_canonical_repository_root(
|
||||
*,
|
||||
configured_value: str | None,
|
||||
source: str | None,
|
||||
expected_slug: str | None,
|
||||
process_project_root: str,
|
||||
remote: str | None = None,
|
||||
require_binding: bool = False,
|
||||
) -> dict:
|
||||
"""Validate the canonical repository root binding, failing closed on forgery.
|
||||
|
||||
Returns a dict with ``proven`` / ``block`` / ``reasons`` plus the resolved
|
||||
``canonical_repo_root`` (the value downstream guards must use),
|
||||
``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).
|
||||
|
||||
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.
|
||||
"""
|
||||
process_root = os.path.realpath(process_project_root)
|
||||
declared = (configured_value or "").strip()
|
||||
|
||||
if not declared:
|
||||
if require_binding:
|
||||
return _assessment(
|
||||
proven=False,
|
||||
reasons=[
|
||||
"no canonical_repository_root configured for a cross-repository "
|
||||
f"namespace; set {CANONICAL_ROOT_ENV} or the profile "
|
||||
"canonical_repository_root field (fail closed)"
|
||||
],
|
||||
configured=False,
|
||||
canonical_repo_root=process_root,
|
||||
resolved_slug=None,
|
||||
source=None,
|
||||
)
|
||||
# Single-repo default: canonical root follows the install checkout.
|
||||
derived = resolve_repo_toplevel(process_root) or process_root
|
||||
return _assessment(
|
||||
proven=True,
|
||||
reasons=[],
|
||||
configured=False,
|
||||
canonical_repo_root=derived,
|
||||
resolved_slug=None,
|
||||
source=None,
|
||||
)
|
||||
|
||||
real = os.path.realpath(os.path.abspath(declared))
|
||||
if not os.path.isdir(real):
|
||||
return _assessment(
|
||||
proven=False,
|
||||
reasons=[
|
||||
f"configured canonical repository root '{real}' does not exist "
|
||||
"or is not a directory (fail closed)"
|
||||
],
|
||||
configured=True,
|
||||
canonical_repo_root=real,
|
||||
resolved_slug=None,
|
||||
source=source,
|
||||
)
|
||||
|
||||
toplevel = resolve_repo_toplevel(real)
|
||||
if not toplevel:
|
||||
return _assessment(
|
||||
proven=False,
|
||||
reasons=[
|
||||
f"configured canonical repository root '{real}' is not a git "
|
||||
"repository (fail closed)"
|
||||
],
|
||||
configured=True,
|
||||
canonical_repo_root=real,
|
||||
resolved_slug=None,
|
||||
source=source,
|
||||
)
|
||||
|
||||
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():
|
||||
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)"
|
||||
)
|
||||
|
||||
return _assessment(
|
||||
proven=not reasons,
|
||||
reasons=reasons,
|
||||
configured=True,
|
||||
canonical_repo_root=toplevel,
|
||||
resolved_slug=resolved_slug,
|
||||
source=source,
|
||||
)
|
||||
|
||||
|
||||
def format_canonical_repository_root_error(assessment: Mapping) -> str:
|
||||
"""Single RuntimeError message for MCP preflight gates."""
|
||||
root = assessment.get("canonical_repo_root") or "(unknown)"
|
||||
source = assessment.get("source") or "(unconfigured)"
|
||||
reasons = "; ".join(
|
||||
assessment.get("reasons") or ["unknown canonical repository root violation"]
|
||||
)
|
||||
return (
|
||||
f"Canonical repository root guard (#706): {reasons}. "
|
||||
f"binding source: {source}; canonical repository root: {root}. "
|
||||
"Configure a valid canonical_repository_root for the target repository "
|
||||
"and relaunch; do not point it at the Gitea-Tools install checkout."
|
||||
)
|
||||
|
||||
|
||||
def _assessment(
|
||||
*,
|
||||
proven: bool,
|
||||
reasons: list[str],
|
||||
configured: bool,
|
||||
canonical_repo_root: str,
|
||||
resolved_slug: str | None,
|
||||
source: str | None,
|
||||
) -> dict:
|
||||
return {
|
||||
"proven": proven,
|
||||
"block": not proven,
|
||||
"reasons": list(reasons),
|
||||
"configured": configured,
|
||||
"canonical_repo_root": canonical_repo_root,
|
||||
"resolved_slug": resolved_slug,
|
||||
"source": source,
|
||||
}
|
||||
@@ -802,6 +802,10 @@ def get_profile():
|
||||
# environment variable must never widen or forge the set of
|
||||
# repositories a session may bind to.
|
||||
"allowed_repositories": _json_list("allowed_repositories"),
|
||||
# #706 cross-repository canonical root binding. Config-sourced here (the
|
||||
# namespace-scoped GITEA_CANONICAL_REPOSITORY_ROOT env override is applied
|
||||
# by canonical_repository_root.configured_canonical_root, not widened here).
|
||||
"canonical_repository_root": jp.get("canonical_repository_root") or None,
|
||||
"audit_label": audit_label,
|
||||
"token_source_name": token_source,
|
||||
"auth_source_type": auth_type,
|
||||
|
||||
@@ -502,6 +502,30 @@ def _validate_allowed_repositories(name, raw):
|
||||
)
|
||||
|
||||
|
||||
def _validate_canonical_repository_root(name, raw):
|
||||
"""Validate the optional per-profile canonical repository root (#706).
|
||||
|
||||
``canonical_repository_root`` binds a cross-repository namespace to the
|
||||
working root of its target repository (separate from the immutable
|
||||
Gitea-Tools install checkout). It is an absolute filesystem path; existence
|
||||
and git identity are validated at bind time by the runtime guard, not here
|
||||
(config validation stays filesystem-independent). Absent means the
|
||||
single-repo default and is allowed.
|
||||
"""
|
||||
if raw is None:
|
||||
return
|
||||
if not isinstance(raw, str) or not raw.strip():
|
||||
raise ConfigError(
|
||||
f"profile '{name}' canonical_repository_root must be a non-empty "
|
||||
"absolute path string to the target repository working root"
|
||||
)
|
||||
if not os.path.isabs(raw.strip()):
|
||||
raise ConfigError(
|
||||
f"profile '{name}' canonical_repository_root {raw!r} must be an "
|
||||
"absolute path"
|
||||
)
|
||||
|
||||
|
||||
def _reject_inline_secrets(kind, name, obj):
|
||||
for key in _INLINE_SECRET_KEYS:
|
||||
if key in obj:
|
||||
@@ -577,6 +601,9 @@ def _load_v2_contexts(data, path):
|
||||
if not isinstance(allowed, list) or not isinstance(forbidden, list):
|
||||
raise ConfigError(f"profile '{name}' operation fields must be lists")
|
||||
_validate_allowed_repositories(name, raw.get("allowed_repositories"))
|
||||
_validate_canonical_repository_root(
|
||||
name, raw.get("canonical_repository_root")
|
||||
)
|
||||
allowed_n = {_normalize_op("gitea", op, name) for op in allowed}
|
||||
forbidden_n = {_normalize_op("gitea", op, name) for op in forbidden}
|
||||
# Reviewer-identity deadlock rule (#100/#103) applies here unchanged.
|
||||
|
||||
+315
-31
@@ -194,6 +194,7 @@ MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
|
||||
RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
|
||||
|
||||
import namespace_workspace_binding as nwb # noqa: E402
|
||||
import canonical_repository_root as crr # noqa: E402 # #706 cross-repo canonical root
|
||||
import mcp_namespace_health # noqa: E402
|
||||
import stale_binding_recovery # noqa: E402
|
||||
|
||||
@@ -376,6 +377,22 @@ def _assess_stale_active_binding(auto_recover: bool = False) -> dict:
|
||||
return report
|
||||
|
||||
|
||||
def _configured_canonical_root() -> tuple[str | None, str | None]:
|
||||
"""Declared cross-repository canonical root for the active namespace (#706).
|
||||
|
||||
Returns ``(value, source)`` from the profile/env binding, or ``(None, None)``
|
||||
for the single-repo default. The raw declared value is threaded into
|
||||
workspace resolution so the branches-only / membership guards evaluate
|
||||
against the configured target repository; the strict identity/existence
|
||||
checks run separately in :func:`_enforce_canonical_repository_root`.
|
||||
"""
|
||||
try:
|
||||
profile = get_profile()
|
||||
except Exception:
|
||||
profile = {}
|
||||
return crr.configured_canonical_root(profile, os.environ)
|
||||
|
||||
|
||||
def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
|
||||
"""Resolve the namespace-scoped workspace root inspected by pre-flight guards.
|
||||
|
||||
@@ -399,8 +416,9 @@ def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
|
||||
|
||||
|
||||
def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict:
|
||||
"""Canonical namespace workspace + repository root for guards (#460/#510)."""
|
||||
"""Canonical namespace workspace + repository root for guards (#460/#510/#706)."""
|
||||
role = _effective_workspace_role()
|
||||
configured_root, _source = _configured_canonical_root()
|
||||
return nwb.resolve_namespace_mutation_context(
|
||||
role_kind=role,
|
||||
worktree_path=worktree_path,
|
||||
@@ -409,6 +427,7 @@ def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dic
|
||||
_reviewer_session_worktree() if role in {"reviewer", "merger"} else None
|
||||
),
|
||||
profile_name=get_profile().get("profile_name"),
|
||||
configured_canonical_root=configured_root,
|
||||
)
|
||||
|
||||
|
||||
@@ -721,6 +740,54 @@ def record_preflight_check(
|
||||
_preflight_resolved_task = resolved_task
|
||||
|
||||
|
||||
def _enforce_canonical_repository_root(
|
||||
worktree_path: str | None = None,
|
||||
*,
|
||||
remote: str | None = None,
|
||||
) -> None:
|
||||
"""#706: validate the immutable cross-repository canonical root binding.
|
||||
|
||||
A no-op for the single-repo default (no configured binding). When a
|
||||
namespace declares a ``canonical_repository_root`` (profile/env), the
|
||||
configured target repository must exist, be a git repository, and carry a
|
||||
repository identity matching the session's authorized repository. Missing,
|
||||
conflicting, or forged bindings fail closed, and the validated root is
|
||||
checked against the immutable session pin so it cannot be swapped mid
|
||||
session.
|
||||
"""
|
||||
configured_value, source = _configured_canonical_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")
|
||||
)
|
||||
assessment = crr.assess_canonical_repository_root(
|
||||
configured_value=configured_value,
|
||||
source=source,
|
||||
expected_slug=expected_slug,
|
||||
process_project_root=PROJECT_ROOT,
|
||||
remote=remote,
|
||||
require_binding=True,
|
||||
)
|
||||
if assessment["block"]:
|
||||
raise RuntimeError(
|
||||
crr.format_canonical_repository_root_error(assessment)
|
||||
)
|
||||
|
||||
drift = session_ctx.assess_session_context(
|
||||
profile_name=get_profile().get("profile_name"),
|
||||
remote=remote,
|
||||
canonical_repository_root=assessment["canonical_repo_root"],
|
||||
)
|
||||
if drift["block"]:
|
||||
raise RuntimeError(
|
||||
"Canonical repository root guard (#706): "
|
||||
+ "; ".join(drift["reasons"])
|
||||
)
|
||||
|
||||
|
||||
def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
|
||||
"""#274: author file/branch mutations must run from a branches/ worktree.
|
||||
|
||||
@@ -767,6 +834,15 @@ def _anti_stomp_in_test_mode() -> bool:
|
||||
return not bool(os.environ.get("GITEA_TEST_FORCE_ANTI_STOMP"))
|
||||
|
||||
|
||||
# Mutation tasks that carry an explicit cross-repository *target* contract and
|
||||
# must NOT auto-resolve the workspace repository for omitted coordinates. Only
|
||||
# ``delete_branch`` (#733) qualifies: a destructive raw-branch deletion names
|
||||
# its org/repo explicitly and fails closed otherwise. Every other mutation is
|
||||
# session-bound to the single workspace-aligned repository and resolves it from
|
||||
# the trusted git remote (mirroring :func:`_resolve`).
|
||||
_EXPLICIT_TARGET_MUTATION_TASKS = frozenset({"delete_branch"})
|
||||
|
||||
|
||||
def _run_anti_stomp_preflight(
|
||||
task: str | None,
|
||||
*,
|
||||
@@ -832,21 +908,56 @@ def _run_anti_stomp_preflight(
|
||||
local_remote_url = None
|
||||
org_explicit = org is not None
|
||||
repo_explicit = repo is not None
|
||||
filled_org = False
|
||||
filled_repo = False
|
||||
if remote:
|
||||
try:
|
||||
local_remote_url = _local_git_remote_url(remote)
|
||||
except Exception:
|
||||
local_remote_url = None
|
||||
if resolved_org is None or resolved_repo is None:
|
||||
try:
|
||||
rem = _effective_remote(remote)
|
||||
if rem in REMOTES:
|
||||
if resolved_org is None:
|
||||
resolved_org = REMOTES[rem]["org"]
|
||||
if resolved_repo is None:
|
||||
resolved_repo = REMOTES[rem]["repo"]
|
||||
except Exception:
|
||||
pass
|
||||
# Session-bound mutations act on the single workspace-aligned
|
||||
# repository, so prefer the trusted git-remote-derived identity for
|
||||
# omitted coordinates — exactly as :func:`_resolve` does — instead
|
||||
# of the remote-wide ``REMOTES`` default *target*. A bare ``prgs``
|
||||
# default resolves to ``Timesheet`` and false-positived the #530
|
||||
# repo guard against a ``Gitea-Tools`` checkout, blocking native
|
||||
# issue/PR/review/merge/lock/cleanup mutations at preflight while
|
||||
# the actual resolve stage (``_resolve``) would have targeted the
|
||||
# workspace repo correctly. Workspace-filled sides are intentional
|
||||
# alignment and are marked explicit so the guard accepts them.
|
||||
# ``delete_branch`` keeps its explicit cross-repo target contract
|
||||
# (#733) and is excluded, so it still fails closed on omitted
|
||||
# coordinates. ``REMOTES`` stays a best-effort last resort only when
|
||||
# no workspace identity can be derived (preserves the dadeschools
|
||||
# default-target behavior and the existing unit suite).
|
||||
workspace_parts = None
|
||||
if task not in _EXPLICIT_TARGET_MUTATION_TASKS:
|
||||
try:
|
||||
workspace_parts = (
|
||||
remote_repo_guard.parse_org_repo_from_remote_url(
|
||||
local_remote_url
|
||||
)
|
||||
)
|
||||
except Exception:
|
||||
workspace_parts = None
|
||||
if workspace_parts:
|
||||
if resolved_org is None:
|
||||
resolved_org = workspace_parts[0]
|
||||
filled_org = True
|
||||
if resolved_repo is None:
|
||||
resolved_repo = workspace_parts[1]
|
||||
filled_repo = True
|
||||
if resolved_org is None or resolved_repo is None:
|
||||
try:
|
||||
rem = _effective_remote(remote)
|
||||
if rem in REMOTES:
|
||||
if resolved_org is None:
|
||||
resolved_org = REMOTES[rem]["org"]
|
||||
if resolved_repo is None:
|
||||
resolved_repo = REMOTES[rem]["repo"]
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
parity = _current_master_parity()
|
||||
startup_head = parity.get("startup_head")
|
||||
@@ -900,8 +1011,10 @@ def _run_anti_stomp_preflight(
|
||||
resolved_org=resolved_org,
|
||||
resolved_repo=resolved_repo,
|
||||
local_remote_url=local_remote_url,
|
||||
org_explicit=org_explicit,
|
||||
repo_explicit=repo_explicit,
|
||||
# Workspace-filled sides are intentional alignment for #530 (same
|
||||
# contract as :func:`_resolve`), so treat them as explicit.
|
||||
org_explicit=org_explicit or filled_org,
|
||||
repo_explicit=repo_explicit or filled_repo,
|
||||
check_repo=check_repo,
|
||||
profile_name=profile_name,
|
||||
profile_role=profile_role,
|
||||
@@ -1069,6 +1182,7 @@ def verify_preflight_purity(
|
||||
)
|
||||
|
||||
# Historical path: root + branches after purity-order when dirty paths live.
|
||||
_enforce_canonical_repository_root(worktree_path, remote=remote)
|
||||
_enforce_root_checkout_guard(worktree_path)
|
||||
_enforce_branches_only_author_mutation(worktree_path)
|
||||
_enforce_issue_scope_guard(
|
||||
@@ -1102,6 +1216,7 @@ def verify_preflight_purity(
|
||||
# #683: under pytest unit isolation, FORCE_PRODUCTION_GUARDS still runs
|
||||
# production root + branches + issue scope (no silent no-op of guards).
|
||||
if production_active:
|
||||
_enforce_canonical_repository_root(worktree_path, remote=remote)
|
||||
_enforce_root_checkout_guard(worktree_path)
|
||||
_enforce_branches_only_author_mutation(worktree_path)
|
||||
_enforce_issue_scope_guard(
|
||||
@@ -1533,6 +1648,45 @@ def _workspace_repository_slug(remote: str | None) -> str | None:
|
||||
return session_ctx.format_repository_slug(parsed[0], parsed[1])
|
||||
|
||||
|
||||
def _canonical_repository_slug(
|
||||
profile: dict,
|
||||
remote: str | None,
|
||||
) -> tuple[str | None, list[str]]:
|
||||
"""Repository identity of the *configured* canonical root, if any (#739 F2).
|
||||
|
||||
Returns ``(slug, reasons)``. A namespace with no configured
|
||||
``canonical_repository_root`` yields ``(None, [])`` so callers keep the
|
||||
install-derived single-repository default. A configured root that cannot be
|
||||
resolved to a repository identity yields ``(None, reasons)`` — a fail-closed
|
||||
signal, never a silent fallback to the installation checkout, because that
|
||||
fallback is exactly what lets a cross-repository namespace validate against
|
||||
the wrong repository.
|
||||
|
||||
Env-over-profile precedence and the existence / git-toplevel / resolvable
|
||||
remote-identity validation all live in ``crr``.
|
||||
"""
|
||||
configured_value, configured_source = crr.configured_canonical_root(
|
||||
profile, os.environ
|
||||
)
|
||||
if not configured_value:
|
||||
return None, []
|
||||
assessment = crr.assess_canonical_repository_root(
|
||||
configured_value=configured_value,
|
||||
source=configured_source,
|
||||
expected_slug=None,
|
||||
process_project_root=PROJECT_ROOT,
|
||||
remote=remote,
|
||||
require_binding=True,
|
||||
)
|
||||
slug = assessment.get("resolved_slug")
|
||||
if assessment.get("block") or not slug:
|
||||
return None, list(assessment.get("reasons") or [
|
||||
"configured canonical repository root has no resolvable "
|
||||
"repository identity (fail closed)"
|
||||
])
|
||||
return slug, []
|
||||
|
||||
|
||||
def _trusted_session_repository(
|
||||
profile: dict,
|
||||
remote: str | None,
|
||||
@@ -1560,7 +1714,21 @@ def _trusted_session_repository(
|
||||
"repository": None,
|
||||
"reasons": [str(exc)],
|
||||
}
|
||||
slug = _workspace_repository_slug(remote)
|
||||
# #706 F1: when a cross-repository canonical root is configured, the session
|
||||
# repository identity must be derived from that validated *target* repository,
|
||||
# not from the install-checkout git remote. ``_workspace_repository_slug``
|
||||
# reads ``_local_git_remote_url`` in ``PROJECT_ROOT`` (always Gitea-Tools), so
|
||||
# without this a cross-repo namespace pinned Gitea-Tools while #274 filesystem
|
||||
# membership bound the external root, and ``_enforce_canonical_repository_root``
|
||||
# then failed closed on a self-inflicted identity mismatch. The derived
|
||||
# identity is still authorized by the profile allowlist below (never
|
||||
# self-authorizing). Env-over-profile precedence and fail-closed validation
|
||||
# (existence, git toplevel, resolvable remote identity) are handled by
|
||||
# ``crr``; unconfigured single-repo namespaces keep the install-derived slug.
|
||||
canonical_slug, canonical_reasons = _canonical_repository_slug(profile, remote)
|
||||
if canonical_reasons:
|
||||
return {"org": None, "repository": None, "reasons": canonical_reasons}
|
||||
slug = canonical_slug or _workspace_repository_slug(remote)
|
||||
scope = session_ctx.assess_repository_scope(
|
||||
workspace_slug=slug,
|
||||
allowed=allowed,
|
||||
@@ -1598,6 +1766,18 @@ def _seed_session_context(
|
||||
"""
|
||||
expected = (profile.get("username") or "").strip() or None
|
||||
trusted = _trusted_session_repository(profile, remote)
|
||||
# #706: pin the configured cross-repository canonical root immutably so a
|
||||
# later call cannot forge/swap it. Store the resolved toplevel so drift
|
||||
# checks compare against the same value the mutation gate validates.
|
||||
configured_value, _crr_source = crr.configured_canonical_root(
|
||||
profile, os.environ
|
||||
)
|
||||
canonical_root_pin = None
|
||||
if configured_value:
|
||||
canonical_root_pin = (
|
||||
crr.resolve_repo_toplevel(configured_value)
|
||||
or os.path.realpath(configured_value)
|
||||
)
|
||||
return session_ctx.seed_session_context_if_unbound(
|
||||
profile_name=profile.get("profile_name") or "",
|
||||
remote=remote,
|
||||
@@ -1608,6 +1788,7 @@ def _seed_session_context(
|
||||
role_kind=_profile_role_kind(profile),
|
||||
expected_username=expected,
|
||||
source=source,
|
||||
canonical_repository_root=canonical_root_pin,
|
||||
)
|
||||
import issue_work_duplicate_gate # noqa: E402
|
||||
import issue_workflow_labels # noqa: E402
|
||||
@@ -2696,8 +2877,13 @@ def gitea_create_issue(
|
||||
return blocked
|
||||
try:
|
||||
with _mutation_stage("preflight_purity"):
|
||||
# #735: forward explicit org/repo into shared anti-stomp preflight.
|
||||
verify_preflight_purity(
|
||||
remote, worktree_path=worktree_path, task="create_issue"
|
||||
remote,
|
||||
worktree_path=worktree_path,
|
||||
task="create_issue",
|
||||
org=org,
|
||||
repo=repo,
|
||||
)
|
||||
except Exception as exc:
|
||||
typed = _production_guard_block_from_exc(exc, number=None)
|
||||
@@ -2919,7 +3105,14 @@ def gitea_lock_issue(
|
||||
)
|
||||
else:
|
||||
git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
|
||||
verify_preflight_purity(remote, worktree_path=resolved_worktree, task="lock_issue")
|
||||
# #735: forward explicit org/repo into shared anti-stomp preflight.
|
||||
verify_preflight_purity(
|
||||
remote,
|
||||
worktree_path=resolved_worktree,
|
||||
task="lock_issue",
|
||||
org=org,
|
||||
repo=repo,
|
||||
)
|
||||
lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
|
||||
worktree_path=resolved_worktree,
|
||||
current_branch=git_state.get("current_branch"),
|
||||
@@ -3150,7 +3343,14 @@ def gitea_create_pr(
|
||||
if blocked:
|
||||
return blocked
|
||||
with _mutation_stage("preflight_purity"):
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr")
|
||||
# #735: forward explicit org/repo into shared anti-stomp preflight.
|
||||
verify_preflight_purity(
|
||||
remote,
|
||||
worktree_path=worktree_path,
|
||||
task="create_pr",
|
||||
org=org,
|
||||
repo=repo,
|
||||
)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
|
||||
# ── Issue Lock Validation (Issue #194 / #196 / #443) ──
|
||||
@@ -7446,9 +7646,9 @@ def gitea_edit_pr(
|
||||
# Closing uses close_pr; non-closing title/body/base edits use edit_pr so
|
||||
# the shared #604 anti-stomp inventory stays consistent with wiring.
|
||||
if closing:
|
||||
verify_preflight_purity(remote, task="close_pr")
|
||||
verify_preflight_purity(remote, task="close_pr", org=org, repo=repo)
|
||||
else:
|
||||
verify_preflight_purity(remote, task="edit_pr")
|
||||
verify_preflight_purity(remote, task="edit_pr", org=org, repo=repo)
|
||||
|
||||
# PR closure is a first-class capability, distinct from retitling or
|
||||
# rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
|
||||
@@ -7699,7 +7899,8 @@ def gitea_commit_files(
|
||||
branch="",
|
||||
)
|
||||
|
||||
verify_preflight_purity(remote, task="commit_files")
|
||||
# #735: forward explicit org/repo into shared anti-stomp preflight.
|
||||
verify_preflight_purity(remote, task="commit_files", org=org, repo=repo)
|
||||
processed_files, source_proofs = _prepare_commit_payload_files(files)
|
||||
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
@@ -8584,8 +8785,28 @@ def _delete_branch_repository_binding_block(
|
||||
(unproven target).
|
||||
|
||||
Returns a structured block dict, or ``None`` when the target is authorized.
|
||||
|
||||
#739 F2: the trusted identity is the session's configured
|
||||
``canonical_repository_root`` when the namespace declares one. Deriving it
|
||||
from ``_workspace_repository_slug`` alone reads the *installation* checkout
|
||||
(``_local_git_remote_url`` runs in ``PROJECT_ROOT``, always Gitea-Tools),
|
||||
which inverts the guard for a cross-repository namespace: the genuinely
|
||||
bound target is rejected and Gitea-Tools is accepted. A configured root that
|
||||
cannot be resolved fails closed rather than falling back to the install
|
||||
identity; unconfigured namespaces keep the install-derived default.
|
||||
"""
|
||||
workspace_slug = _workspace_repository_slug(remote)
|
||||
canonical_slug, canonical_reasons = _canonical_repository_slug(
|
||||
get_profile(), remote
|
||||
)
|
||||
if canonical_reasons:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"required_permission": "gitea.branch.delete",
|
||||
"reasons": canonical_reasons,
|
||||
"blocker_kind": "repository_binding",
|
||||
}
|
||||
workspace_slug = canonical_slug or _workspace_repository_slug(remote)
|
||||
workspace_parts = (
|
||||
session_ctx.parse_repository_slug(workspace_slug) if workspace_slug else None
|
||||
)
|
||||
@@ -8814,6 +9035,8 @@ def gitea_cleanup_merged_pr_branch(
|
||||
remote,
|
||||
worktree_path=worktree_path,
|
||||
task="cleanup_merged_pr_branch",
|
||||
org=org,
|
||||
repo=repo,
|
||||
)
|
||||
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
@@ -9541,7 +9764,9 @@ def gitea_reconcile_merged_cleanups(
|
||||
report["executed"] = False
|
||||
return {"success": True, "performed": False, **report}
|
||||
|
||||
verify_preflight_purity(remote, task="reconcile_merged_cleanups")
|
||||
verify_preflight_purity(
|
||||
remote, task="reconcile_merged_cleanups", org=org, repo=repo
|
||||
)
|
||||
actions: list[dict] = []
|
||||
for entry in report.get("entries") or []:
|
||||
head_branch = entry.get("head_branch") or ""
|
||||
@@ -10070,7 +10295,9 @@ def gitea_reconcile_already_landed_pr(
|
||||
)
|
||||
return result
|
||||
|
||||
verify_preflight_purity(remote, task="reconcile_already_landed_pr")
|
||||
verify_preflight_purity(
|
||||
remote, task="reconcile_already_landed_pr", org=org, repo=repo
|
||||
)
|
||||
|
||||
if post_comment and comment_body.strip():
|
||||
comment_block = _profile_operation_gate("gitea.pr.comment")
|
||||
@@ -10341,7 +10568,9 @@ def gitea_reconcile_superseded_by_merged_pr(
|
||||
result["safe_next_action"] = "rerun with explicit mutation flags"
|
||||
return result
|
||||
|
||||
verify_preflight_purity(remote, task="reconcile_close_superseded_pr")
|
||||
verify_preflight_purity(
|
||||
remote, task="reconcile_close_superseded_pr", org=org, repo=repo
|
||||
)
|
||||
|
||||
if post_comment:
|
||||
comment_url = f"{base}/issues/{target_pr_number}/comments"
|
||||
@@ -10444,7 +10673,7 @@ def gitea_close_issue(
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, task="close_issue")
|
||||
verify_preflight_purity(remote, task="close_issue", org=org, repo=repo)
|
||||
|
||||
# #529: block closure that buries an unproven non-zero suite exit as an
|
||||
# "expected pre-existing failure". Skipped when no closure report is given.
|
||||
@@ -12411,10 +12640,13 @@ def gitea_create_issue_comment(
|
||||
# allowed while an author holds a different implementation lock.
|
||||
# Scope ownership for source edits is enforced via branch/worktree
|
||||
# binding + root diagnostic checks (#683).
|
||||
# #735: forward explicit org/repo into shared anti-stomp preflight.
|
||||
verify_preflight_purity(
|
||||
remote,
|
||||
worktree_path=worktree_path,
|
||||
task="comment_issue",
|
||||
org=org,
|
||||
repo=repo,
|
||||
)
|
||||
except Exception as exc:
|
||||
typed = _production_guard_block_from_exc(
|
||||
@@ -13841,6 +14073,16 @@ def gitea_get_runtime_context(
|
||||
remote: Known instance — 'dadeschools' or 'prgs'.
|
||||
host: Override the Gitea host.
|
||||
"""
|
||||
# #739 F1: normalize the remote *before* any host lookup, identity
|
||||
# resolution, or session seeding. ``remote`` defaults to 'dadeschools'
|
||||
# across the whole tool surface, so a caller that omits it on a prgs-hosted
|
||||
# namespace would otherwise pin the argument default: the identity lookup
|
||||
# below would authenticate against the wrong host, and because first-bind is
|
||||
# first-write-wins a later, correct ``gitea_whoami(remote="prgs")`` could not
|
||||
# repair the binding. ``gitea_whoami`` has always normalized here; this is
|
||||
# the same call, not a new policy. Explicit non-default remotes pass through
|
||||
# untouched, and a dadeschools-hosted profile still resolves to dadeschools.
|
||||
remote = _effective_remote(remote)
|
||||
profile = get_profile()
|
||||
config = gitea_config.load_config()
|
||||
reveal = _reveal_endpoints()
|
||||
@@ -14018,13 +14260,23 @@ def gitea_assess_master_parity(
|
||||
Never mutates and makes no network calls. Read-only operations are never
|
||||
blocked by staleness; only mutating operations fail closed while stale.
|
||||
|
||||
The top-level ``startup_head``/``current_head``/``in_parity`` fields keep
|
||||
their original meaning: the *Gitea-Tools server implementation* only. #739
|
||||
F3 adds two separately labelled dimensions so a cross-repository namespace
|
||||
can tell them apart — ``server_implementation`` (this installation, restated
|
||||
under an explicit name) and ``target_repository`` (the configured canonical
|
||||
target checkout and its last-known remote master). Only the server dimension
|
||||
gates mutations.
|
||||
|
||||
Returns:
|
||||
dict with 'in_parity', 'stale', 'restart_required', 'startup_head',
|
||||
'current_head', 'mutation_gate_enforced', 'summary', 'reasons', and a
|
||||
'report' recovery payload when stale.
|
||||
'current_head', 'mutation_gate_enforced', 'summary', 'reasons',
|
||||
'server_implementation', 'target_repository', and a 'report' recovery
|
||||
payload when stale.
|
||||
"""
|
||||
parity = _current_master_parity()
|
||||
enforced = not master_parity_gate.gate_disabled()
|
||||
canonical_root, canonical_source = _configured_canonical_root()
|
||||
out = {
|
||||
"in_parity": parity["in_parity"],
|
||||
"stale": parity["stale"],
|
||||
@@ -14036,6 +14288,17 @@ def gitea_assess_master_parity(
|
||||
"summary": master_parity_gate.format_parity(parity),
|
||||
"reasons": parity["reasons"],
|
||||
"process_root": PROJECT_ROOT,
|
||||
"server_implementation": {
|
||||
"installation_root": PROJECT_ROOT,
|
||||
"startup_head": parity["startup_head"],
|
||||
"current_head": parity["current_head"],
|
||||
"stale": parity["stale"],
|
||||
"determinable": parity["determinable"],
|
||||
},
|
||||
"target_repository": master_parity_gate.assess_target_repository_parity(
|
||||
canonical_root=canonical_root,
|
||||
source=canonical_source,
|
||||
),
|
||||
}
|
||||
if parity["stale"] and enforced:
|
||||
out["report"] = master_parity_gate.parity_report(parity)
|
||||
@@ -14550,7 +14813,14 @@ def gitea_mark_issue(
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path, task="mark_issue")
|
||||
# #735: forward explicit org/repo into shared anti-stomp preflight.
|
||||
verify_preflight_purity(
|
||||
remote,
|
||||
worktree_path=worktree_path,
|
||||
task="mark_issue",
|
||||
org=org,
|
||||
repo=repo,
|
||||
)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
base = repo_api_url(h, o, r)
|
||||
@@ -14640,7 +14910,7 @@ def gitea_post_heartbeat(
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, task="post_heartbeat")
|
||||
verify_preflight_purity(remote, task="post_heartbeat", org=org, repo=repo)
|
||||
active_profile = profile or get_profile().get("profile_name")
|
||||
body = issue_claim_heartbeat.format_heartbeat_body(
|
||||
kind="progress",
|
||||
@@ -15719,7 +15989,9 @@ def gitea_cleanup_stale_claims(
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, task="cleanup_stale_claims")
|
||||
verify_preflight_purity(
|
||||
remote, task="cleanup_stale_claims", org=org, repo=repo
|
||||
)
|
||||
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
@@ -15839,7 +16111,13 @@ def gitea_create_label(
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path, task="create_label")
|
||||
verify_preflight_purity(
|
||||
remote,
|
||||
worktree_path=worktree_path,
|
||||
task="create_label",
|
||||
org=org,
|
||||
repo=repo,
|
||||
)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
base = repo_api_url(h, o, r)
|
||||
@@ -15893,7 +16171,13 @@ def gitea_set_issue_labels(
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path, task="set_issue_labels")
|
||||
verify_preflight_purity(
|
||||
remote,
|
||||
worktree_path=worktree_path,
|
||||
task="set_issue_labels",
|
||||
org=org,
|
||||
repo=repo,
|
||||
)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
base = repo_api_url(h, o, r)
|
||||
|
||||
@@ -166,3 +166,128 @@ def format_parity(assessment: dict) -> str:
|
||||
if not assessment.get("determinable"):
|
||||
return "parity indeterminate (baseline or current HEAD unknown)"
|
||||
return f"in parity at {_short(assessment.get('current_head'))}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Target-repository parity (#739 F3)
|
||||
#
|
||||
# Everything above measures ONE dimension: the Gitea-Tools server's own
|
||||
# implementation commit, comparing the SHA this process was loaded from against
|
||||
# the SHA now on disk at PROJECT_ROOT. That is deliberate and is left untouched
|
||||
# — it is what proves the in-memory capability gates are current.
|
||||
#
|
||||
# It is not, however, a statement about the repository a cross-repository
|
||||
# namespace actually mutates. The assessment below is a separate, separately
|
||||
# labelled dimension for the configured canonical target repository. It never
|
||||
# feeds the mutation gate and never changes startup_head/current_head.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
DEFAULT_TARGET_TRACKING_REF = "refs/remotes/origin/master"
|
||||
|
||||
|
||||
def _git_capture(root: str, *args: str) -> str | None:
|
||||
"""Run a read-only git command in *root*; ``None`` on any failure.
|
||||
|
||||
Deliberately does not honour ``GITEA_TEST_CURRENT_HEAD``: that override
|
||||
exists to pin the *server's* HEAD, and applying it here would make a target
|
||||
repository silently report the server's forced SHA.
|
||||
"""
|
||||
if not root:
|
||||
return None
|
||||
try:
|
||||
res = subprocess.run(
|
||||
["git", "-C", root, *args],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
except Exception:
|
||||
return None
|
||||
if res.returncode != 0:
|
||||
return None
|
||||
return (res.stdout or "").strip() or None
|
||||
|
||||
|
||||
def assess_target_repository_parity(
|
||||
*,
|
||||
canonical_root: str | None,
|
||||
source: str | None,
|
||||
tracking_ref: str = DEFAULT_TARGET_TRACKING_REF,
|
||||
) -> dict:
|
||||
"""Assess the configured cross-repository target checkout.
|
||||
|
||||
Reports the target's canonical root, repository identity, checked-out
|
||||
commit, and last-known remote master commit, plus whether the checkout is
|
||||
behind that ref. No network call is made: the remote side is read from the
|
||||
existing remote-tracking ref, so a target that has never been fetched is
|
||||
reported as indeterminate rather than guessed at.
|
||||
|
||||
An unconfigured namespace is ``configured=False`` and never ``stale`` — the
|
||||
single-repository default has no second dimension to be stale about. A
|
||||
configured root that cannot be read is ``determinable=False`` with reasons.
|
||||
"""
|
||||
result = {
|
||||
"configured": bool(canonical_root),
|
||||
"canonical_repository_root": None,
|
||||
"source": source,
|
||||
"repository_slug": None,
|
||||
"checkout_head": None,
|
||||
"tracking_ref": tracking_ref,
|
||||
"remote_tracking_head": None,
|
||||
"determinable": False,
|
||||
"stale": False,
|
||||
"reasons": [],
|
||||
}
|
||||
if not canonical_root:
|
||||
result["determinable"] = True
|
||||
return result
|
||||
|
||||
result["canonical_repository_root"] = canonical_root
|
||||
if not os.path.isdir(canonical_root):
|
||||
result["reasons"].append(
|
||||
f"configured canonical repository root '{canonical_root}' "
|
||||
f"does not exist or is not a directory"
|
||||
)
|
||||
return result
|
||||
|
||||
toplevel = _git_capture(canonical_root, "rev-parse", "--show-toplevel")
|
||||
if not toplevel:
|
||||
result["reasons"].append(
|
||||
f"configured canonical repository root '{canonical_root}' "
|
||||
f"is not a git checkout"
|
||||
)
|
||||
return result
|
||||
|
||||
head = _git_capture(canonical_root, "rev-parse", "HEAD")
|
||||
if not head:
|
||||
result["reasons"].append(
|
||||
f"target repository HEAD could not be read at '{canonical_root}'"
|
||||
)
|
||||
return result
|
||||
result["checkout_head"] = head
|
||||
result["determinable"] = True
|
||||
|
||||
remote_url = _git_capture(canonical_root, "remote", "get-url", "origin")
|
||||
if remote_url:
|
||||
# Local import keeps this module dependency-light for its startup role.
|
||||
import remote_repo_guard
|
||||
|
||||
parsed = remote_repo_guard.parse_org_repo_from_remote_url(remote_url)
|
||||
if parsed:
|
||||
result["repository_slug"] = f"{parsed[0]}/{parsed[1]}"
|
||||
if not result["repository_slug"]:
|
||||
result["reasons"].append(
|
||||
"target repository identity could not be derived from its git remote"
|
||||
)
|
||||
|
||||
tracking_head = _git_capture(canonical_root, "rev-parse", tracking_ref)
|
||||
if not tracking_head:
|
||||
result["reasons"].append(
|
||||
f"remote-tracking ref '{tracking_ref}' is unknown in the target "
|
||||
f"checkout; target staleness is indeterminate (no fetch is "
|
||||
f"performed by this assessment)"
|
||||
)
|
||||
return result
|
||||
result["remote_tracking_head"] = tracking_head
|
||||
result["stale"] = tracking_head != head
|
||||
return result
|
||||
|
||||
@@ -109,8 +109,17 @@ def resolve_namespace_mutation_context(
|
||||
session_lease_worktree: str | None = None,
|
||||
worktree: str | None = None,
|
||||
profile_name: str | None = None,
|
||||
configured_canonical_root: str | None = None,
|
||||
) -> dict:
|
||||
"""Shared workspace resolution for runtime_context and mutation guards."""
|
||||
"""Shared workspace resolution for runtime_context and mutation guards.
|
||||
|
||||
When *configured_canonical_root* is supplied (a cross-repository namespace
|
||||
bound to an external target repository, #706), the canonical repository root
|
||||
is that configured target rather than the MCP install checkout. This keeps
|
||||
the branches-only / worktree-membership guards (#274) evaluating against the
|
||||
repository the namespace actually mutates. Without it the single-repo
|
||||
default is preserved: the canonical root follows the process checkout.
|
||||
"""
|
||||
demotions: list[str] = []
|
||||
workspace, binding_source = resolve_namespace_workspace(
|
||||
role_kind=role_kind,
|
||||
@@ -132,7 +141,11 @@ def resolve_namespace_mutation_context(
|
||||
env=env,
|
||||
profile_name=profile_name,
|
||||
)
|
||||
canonical_root = amw.resolve_canonical_repo_root(process_root, process_root)
|
||||
configured = (configured_canonical_root or "").strip()
|
||||
if configured:
|
||||
canonical_root = os.path.realpath(configured)
|
||||
else:
|
||||
canonical_root = amw.resolve_canonical_repo_root(process_root, process_root)
|
||||
return {
|
||||
"workspace_path": workspace,
|
||||
"workspace_binding_source": binding_source,
|
||||
@@ -258,6 +271,7 @@ def assess_namespace_mutation_workspace(
|
||||
session_lease_worktree: str | None = None,
|
||||
profile_name: str | None = None,
|
||||
current_branch: str | None = None,
|
||||
configured_canonical_root: str | None = None,
|
||||
) -> dict:
|
||||
"""Evaluate namespace workspace binding before preflight/mutation."""
|
||||
ctx = resolve_namespace_mutation_context(
|
||||
@@ -268,6 +282,7 @@ def assess_namespace_mutation_workspace(
|
||||
env=env,
|
||||
session_lease_worktree=session_lease_worktree,
|
||||
profile_name=profile_name,
|
||||
configured_canonical_root=configured_canonical_root,
|
||||
)
|
||||
mutation_workspace = ctx["workspace_path"]
|
||||
binding_source = ctx["workspace_binding_source"]
|
||||
|
||||
@@ -32,6 +32,7 @@ class _SessionContext:
|
||||
expected_username: str | None
|
||||
source: str
|
||||
pid: int
|
||||
canonical_repository_root: str | None = None
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
@@ -45,6 +46,7 @@ class _SessionContext:
|
||||
"expected_username": self.expected_username,
|
||||
"source": self.source,
|
||||
"pid": self.pid,
|
||||
"canonical_repository_root": self.canonical_repository_root,
|
||||
}
|
||||
|
||||
|
||||
@@ -143,6 +145,7 @@ def bind_session_context(
|
||||
role_kind: str | None = None,
|
||||
expected_username: str | None = None,
|
||||
source: str = "bind",
|
||||
canonical_repository_root: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Atomically bind/re-bind context (the explicit activation path)."""
|
||||
with _SESSION_CONTEXT_LOCK:
|
||||
@@ -156,6 +159,7 @@ def bind_session_context(
|
||||
role_kind=role_kind,
|
||||
expected_username=expected_username,
|
||||
source=source,
|
||||
canonical_repository_root=canonical_repository_root,
|
||||
)
|
||||
|
||||
|
||||
@@ -170,6 +174,7 @@ def _bind_session_context_unlocked(
|
||||
role_kind: str | None,
|
||||
expected_username: str | None,
|
||||
source: str,
|
||||
canonical_repository_root: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Store a complete immutable context while the caller holds the lock."""
|
||||
global _SESSION_CONTEXT
|
||||
@@ -184,6 +189,7 @@ def _bind_session_context_unlocked(
|
||||
expected_username=(expected_username or "").strip() or None,
|
||||
source=source,
|
||||
pid=os.getpid(),
|
||||
canonical_repository_root=(canonical_repository_root or "").strip() or None,
|
||||
)
|
||||
return _SESSION_CONTEXT.as_dict()
|
||||
|
||||
@@ -199,6 +205,7 @@ def seed_session_context_if_unbound(
|
||||
role_kind: str | None = None,
|
||||
expected_username: str | None = None,
|
||||
source: str = "seed",
|
||||
canonical_repository_root: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Atomically bind only when this process has no current context.
|
||||
|
||||
@@ -219,6 +226,7 @@ def seed_session_context_if_unbound(
|
||||
role_kind=role_kind,
|
||||
expected_username=expected_username,
|
||||
source=source,
|
||||
canonical_repository_root=canonical_repository_root,
|
||||
)
|
||||
return _SESSION_CONTEXT.as_dict()
|
||||
|
||||
@@ -232,6 +240,7 @@ def assess_session_context(
|
||||
repository: str | None = None,
|
||||
org: str | None = None,
|
||||
expected_username: str | None = None,
|
||||
canonical_repository_root: str | None = None,
|
||||
require_bound: bool = False,
|
||||
require_complete: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
@@ -272,6 +281,7 @@ def assess_session_context(
|
||||
live_identity = (identity or "").strip() or None
|
||||
live_repo = (repository or "").strip() or None
|
||||
live_org = (org or "").strip() or None
|
||||
live_canonical = (canonical_repository_root or "").strip() or None
|
||||
|
||||
if ctx.get("profile_name") and live_profile and live_profile != ctx.get("profile_name"):
|
||||
reasons.append(
|
||||
@@ -307,6 +317,13 @@ def assess_session_context(
|
||||
f"org drift: live '{live_org}' != bound "
|
||||
f"'{ctx.get('org')}' (fail closed)"
|
||||
)
|
||||
bound_canonical = ctx.get("canonical_repository_root")
|
||||
if bound_canonical and live_canonical and live_canonical != bound_canonical:
|
||||
reasons.append(
|
||||
f"canonical repository root drift: live '{live_canonical}' != bound "
|
||||
f"'{bound_canonical}' (forged or conflicting cross-repository "
|
||||
"binding, fail closed)"
|
||||
)
|
||||
|
||||
expected = expected_username or ctx.get("expected_username")
|
||||
if expected and live_identity and live_identity != expected:
|
||||
@@ -580,6 +597,7 @@ def mutation_context_audit_fields(
|
||||
"session_org": data.get("org"),
|
||||
"session_role_kind": data.get("role_kind"),
|
||||
"session_context_source": data.get("source"),
|
||||
"session_canonical_repository_root": data.get("canonical_repository_root"),
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,704 @@
|
||||
"""Complete canonical-root consumption for cross-repository MCP namespaces.
|
||||
|
||||
#706 introduced an immutable, configured ``canonical_repository_root`` and
|
||||
routed the #274 filesystem guards and the session repository *slug* through it.
|
||||
Three consumption paths were left behind, and this module drives each one
|
||||
through its production entry point:
|
||||
|
||||
* **A — remote initialization.** ``gitea_get_runtime_context`` never normalized
|
||||
its ``remote`` argument, while ``gitea_whoami`` did. A fresh process whose
|
||||
first native call is the runtime-context path therefore pinned the
|
||||
``dadeschools`` argument default instead of the profile's configured remote,
|
||||
and — because first-bind is first-write-wins — no later correct call could
|
||||
repair it.
|
||||
|
||||
* **C — reconciler branch deletion.** The delete-branch repository-binding
|
||||
guard derived its expected slug from ``_workspace_repository_slug``, which
|
||||
reads the git remote of the *installation* checkout (always Gitea-Tools),
|
||||
rather than from the session's configured canonical root.
|
||||
|
||||
* **D — parity evidence.** ``gitea_assess_master_parity`` proves Gitea-Tools
|
||||
*server implementation* parity only, which is intentional. It carried no
|
||||
target-repository dimension at all, so a cross-repository namespace had no
|
||||
evidence that its target checkout was current. The existing
|
||||
``startup_head``/``current_head`` semantics are preserved unchanged and the
|
||||
target-repository assessment is reported under separately labelled fields.
|
||||
|
||||
Groups B and E assert *intentional* behaviour (the #274 guards already consume
|
||||
the canonical root; the configuration surface already validates a
|
||||
repository-specific namespace) so that a regression in either is caught.
|
||||
|
||||
Real git repositories are used throughout; no network calls are made.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import canonical_repository_root as crr # noqa: E402
|
||||
import gitea_config # noqa: E402
|
||||
import gitea_mcp_server as srv # noqa: E402
|
||||
import master_parity_gate # noqa: E402
|
||||
import mcp_server # noqa: E402
|
||||
import namespace_workspace_binding as nwb # noqa: E402
|
||||
import session_context_binding as session_ctx # noqa: E402
|
||||
|
||||
INSTALL_ORG = "Scaled-Tech-Consulting"
|
||||
INSTALL_REPO = "Gitea-Tools"
|
||||
INSTALL_SLUG = f"{INSTALL_ORG}/{INSTALL_REPO}"
|
||||
INSTALL_URL = f"https://gitea.prgs.cc/{INSTALL_SLUG}.git"
|
||||
|
||||
TARGET_ORG = "Scaled-Tech-Consulting"
|
||||
TARGET_REPO = "mcp-control-plane"
|
||||
TARGET_SLUG = f"{TARGET_ORG}/{TARGET_REPO}"
|
||||
TARGET_URL = f"https://gitea.prgs.cc/{TARGET_SLUG}.git"
|
||||
|
||||
|
||||
def _git(cwd: str, *args: str) -> str:
|
||||
res = subprocess.run(
|
||||
["git", "-C", cwd, *args], capture_output=True, text=True, check=True
|
||||
)
|
||||
return res.stdout.strip()
|
||||
|
||||
|
||||
def _init_repo(path: Path, remote_url: str, *, remote_name: str = "origin") -> str:
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
_git(str(path), "init", "-q")
|
||||
_git(str(path), "config", "user.email", "[email protected]")
|
||||
_git(str(path), "config", "user.name", "Test")
|
||||
_git(str(path), "remote", "add", remote_name, remote_url)
|
||||
(path / "README.md").write_text("seed\n")
|
||||
_git(str(path), "add", "README.md")
|
||||
_git(str(path), "commit", "-q", "-m", "seed")
|
||||
return os.path.realpath(str(path))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared profile/config construction
|
||||
# ---------------------------------------------------------------------------
|
||||
_AUTHOR_OPS = [
|
||||
"gitea.read",
|
||||
"gitea.issue.create",
|
||||
"gitea.issue.comment",
|
||||
"gitea.pr.create",
|
||||
"gitea.pr.comment",
|
||||
"gitea.branch.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.repo.commit",
|
||||
]
|
||||
_AUTHOR_FORBIDDEN = [
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.merge",
|
||||
"gitea.pr.request_changes",
|
||||
]
|
||||
# A profile that may approve or merge must forbid authoring (gitea_config's
|
||||
# reviewer-identity deadlock rule).
|
||||
_REVIEWER_OPS = [
|
||||
"gitea.read",
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.request_changes",
|
||||
"gitea.pr.comment",
|
||||
"gitea.issue.comment",
|
||||
]
|
||||
_REVIEWER_FORBIDDEN = ["gitea.pr.create", "gitea.branch.push", "gitea.pr.merge"]
|
||||
_MERGER_OPS = ["gitea.read", "gitea.pr.merge", "gitea.pr.comment"]
|
||||
_MERGER_FORBIDDEN = [
|
||||
"gitea.pr.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.request_changes",
|
||||
]
|
||||
_RECONCILER_OPS = [
|
||||
"gitea.read",
|
||||
"gitea.pr.close",
|
||||
"gitea.pr.comment",
|
||||
"gitea.branch.delete",
|
||||
]
|
||||
_RECONCILER_FORBIDDEN = [
|
||||
"gitea.pr.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.merge",
|
||||
"gitea.pr.request_changes",
|
||||
]
|
||||
|
||||
_ROLE_MATRIX = {
|
||||
"author": (_AUTHOR_OPS, _AUTHOR_FORBIDDEN),
|
||||
"reviewer": (_REVIEWER_OPS, _REVIEWER_FORBIDDEN),
|
||||
"merger": (_MERGER_OPS, _MERGER_FORBIDDEN),
|
||||
"reconciler": (_RECONCILER_OPS, _RECONCILER_FORBIDDEN),
|
||||
}
|
||||
|
||||
|
||||
def _profile(
|
||||
role: str,
|
||||
*,
|
||||
canonical_root: str | None = None,
|
||||
allowed_repositories: list[str] | None = None,
|
||||
username: str = "jcwalker3",
|
||||
) -> dict:
|
||||
allowed, forbidden = _ROLE_MATRIX[role]
|
||||
profile = {
|
||||
"enabled": True,
|
||||
"context": "prgs",
|
||||
"role": role,
|
||||
"username": username,
|
||||
"base_url": "https://gitea.prgs.cc",
|
||||
"auth": {"type": "env", "name": f"GITEA_TOKEN_PRGS_{role.upper()}"},
|
||||
"allowed_operations": list(allowed),
|
||||
"forbidden_operations": list(forbidden),
|
||||
"execution_profile": f"prgs-{role}",
|
||||
}
|
||||
if canonical_root is not None:
|
||||
profile["canonical_repository_root"] = canonical_root
|
||||
if allowed_repositories is not None:
|
||||
profile["allowed_repositories"] = list(allowed_repositories)
|
||||
return profile
|
||||
|
||||
|
||||
def _config(profiles: dict) -> dict:
|
||||
return {
|
||||
"version": 2,
|
||||
"rules": {"allow_runtime_switching": True},
|
||||
"contexts": {
|
||||
"prgs": {
|
||||
"enabled": True,
|
||||
"gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"},
|
||||
},
|
||||
"mdcps": {
|
||||
"enabled": True,
|
||||
"gitea": {
|
||||
"enabled": True,
|
||||
"base_url": "https://gitea.dadeschools.net",
|
||||
},
|
||||
},
|
||||
},
|
||||
"profiles": profiles,
|
||||
}
|
||||
|
||||
|
||||
class _ServerHarness(unittest.TestCase):
|
||||
"""Temp profiles.json + pinned install remote + no network."""
|
||||
|
||||
def setUp(self):
|
||||
self._dir = tempfile.TemporaryDirectory()
|
||||
self.tmp = self._dir.name
|
||||
self.config_path = os.path.join(self.tmp, "profiles.json")
|
||||
session_ctx._reset_session_context_for_testing()
|
||||
gitea_config._active_profile_override = None
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
srv._MUTATION_AUTHORITY = None
|
||||
# The install checkout's remote is Gitea-Tools regardless of the
|
||||
# developer's layout; pin it so "install-derived" is deterministic.
|
||||
self._remote_url = patch.object(
|
||||
srv, "_local_git_remote_url", side_effect=self._install_remote_url
|
||||
)
|
||||
self._remote_url.start()
|
||||
|
||||
def tearDown(self):
|
||||
self._remote_url.stop()
|
||||
session_ctx._reset_session_context_for_testing()
|
||||
gitea_config._active_profile_override = None
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
srv._MUTATION_AUTHORITY = None
|
||||
self._dir.cleanup()
|
||||
|
||||
def _install_remote_url(self, remote_name):
|
||||
return INSTALL_URL if remote_name in ("prgs", "origin") else None
|
||||
|
||||
def _write_config(self, profiles: dict) -> None:
|
||||
with open(self.config_path, "w", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(_config(profiles)))
|
||||
|
||||
def _env(self, profile_name: str, **extra) -> dict:
|
||||
env = {
|
||||
"GITEA_MCP_CONFIG": self.config_path,
|
||||
"GITEA_MCP_PROFILE": profile_name,
|
||||
"GITEA_TOKEN_PRGS_AUTHOR": "t",
|
||||
"GITEA_TOKEN_PRGS_REVIEWER": "t",
|
||||
"GITEA_TOKEN_PRGS_MERGER": "t",
|
||||
"GITEA_TOKEN_PRGS_RECONCILER": "t",
|
||||
}
|
||||
env.update(extra)
|
||||
return env
|
||||
|
||||
def _api(self, method, url, header):
|
||||
return {
|
||||
"login": "jcwalker3",
|
||||
"full_name": "Test",
|
||||
"id": 1,
|
||||
"email": "[email protected]",
|
||||
}
|
||||
|
||||
def _live(self):
|
||||
return patch("gitea_mcp_server.api_request", side_effect=self._api)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# A. Remote initialization
|
||||
# ===========================================================================
|
||||
class TestRemoteInitializationFirstCall(_ServerHarness):
|
||||
"""A prgs namespace must never pin the dadeschools argument default."""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self._write_config(
|
||||
{"prgs-author": _profile("author", allowed_repositories=[INSTALL_SLUG])}
|
||||
)
|
||||
|
||||
def test_runtime_context_first_call_does_not_bind_default_remote(self):
|
||||
"""Runtime-context as the FIRST native call, relying on the default."""
|
||||
with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
|
||||
srv.gitea_get_runtime_context()
|
||||
ctx = session_ctx.get_session_context()
|
||||
self.assertIsNotNone(ctx, "first call must establish a session binding")
|
||||
self.assertEqual(ctx["remote"], "prgs")
|
||||
self.assertEqual(ctx["host"], "gitea.prgs.cc")
|
||||
self.assertEqual(ctx["org"], INSTALL_ORG)
|
||||
self.assertEqual(ctx["repository"], INSTALL_REPO)
|
||||
|
||||
def test_runtime_context_first_call_reports_effective_remote(self):
|
||||
with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
|
||||
result = srv.gitea_get_runtime_context()
|
||||
self.assertEqual(result["remote"], "prgs")
|
||||
|
||||
def test_whoami_first_call_remains_correct(self):
|
||||
"""Control: the already-normalizing path is unchanged."""
|
||||
with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
|
||||
srv.gitea_whoami()
|
||||
ctx = session_ctx.get_session_context()
|
||||
self.assertEqual(ctx["remote"], "prgs")
|
||||
self.assertEqual(ctx["host"], "gitea.prgs.cc")
|
||||
|
||||
def test_explicit_remote_argument_still_honoured(self):
|
||||
"""Normalization only fills the default; explicit values are untouched."""
|
||||
with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
|
||||
result = srv.gitea_get_runtime_context(remote="prgs")
|
||||
self.assertEqual(result["remote"], "prgs")
|
||||
|
||||
def test_mdcps_profile_default_is_not_rewritten_to_prgs(self):
|
||||
"""A dadeschools-hosted profile keeps the dadeschools remote."""
|
||||
profile = _profile("author", allowed_repositories=[INSTALL_SLUG])
|
||||
profile["context"] = "mdcps"
|
||||
profile["base_url"] = "https://gitea.dadeschools.net"
|
||||
self._write_config({"prgs-author": profile})
|
||||
with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
|
||||
result = srv.gitea_get_runtime_context()
|
||||
self.assertEqual(result["remote"], "dadeschools")
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# B. Remote/repository guard (intentional behaviour — regression fence)
|
||||
# ===========================================================================
|
||||
class TestCanonicalRootGuardBinding(_ServerHarness):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
|
||||
self.install_root = _init_repo(Path(self.tmp) / "install", INSTALL_URL)
|
||||
|
||||
def test_canonical_target_root_resolves_to_target_slug(self):
|
||||
got = crr.assess_canonical_repository_root(
|
||||
configured_value=self.target_root,
|
||||
source="profile.canonical_repository_root",
|
||||
expected_slug=None,
|
||||
process_project_root=self.install_root,
|
||||
remote="prgs",
|
||||
require_binding=True,
|
||||
)
|
||||
self.assertFalse(got.get("block"), got.get("reasons"))
|
||||
self.assertEqual(got["resolved_slug"], TARGET_SLUG)
|
||||
|
||||
def test_install_root_identity_remains_gitea_tools(self):
|
||||
got = crr.assess_canonical_repository_root(
|
||||
configured_value=None,
|
||||
source=None,
|
||||
expected_slug=None,
|
||||
process_project_root=self.install_root,
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertEqual(
|
||||
os.path.realpath(got["canonical_repo_root"]), self.install_root
|
||||
)
|
||||
|
||||
def test_guards_validate_against_canonical_target_root(self):
|
||||
ctx = nwb.resolve_namespace_mutation_context(
|
||||
role_kind="author",
|
||||
worktree_path=None,
|
||||
process_project_root=self.install_root,
|
||||
configured_canonical_root=self.target_root,
|
||||
)
|
||||
self.assertEqual(
|
||||
os.path.realpath(ctx["canonical_repo_root"]), self.target_root
|
||||
)
|
||||
|
||||
def test_explicit_coordinates_cannot_override_canonical_binding(self):
|
||||
"""Explicit org/repo may confirm, never authorize, a binding."""
|
||||
confirm = session_ctx.assess_repository_override(
|
||||
requested_org=TARGET_ORG,
|
||||
requested_repo=TARGET_REPO,
|
||||
bound_org=TARGET_ORG,
|
||||
bound_repo=TARGET_REPO,
|
||||
)
|
||||
self.assertFalse(confirm.get("block"))
|
||||
override = session_ctx.assess_repository_override(
|
||||
requested_org=TARGET_ORG,
|
||||
requested_repo=TARGET_REPO,
|
||||
bound_org=INSTALL_ORG,
|
||||
bound_repo=INSTALL_REPO,
|
||||
)
|
||||
self.assertTrue(override.get("block"))
|
||||
|
||||
def test_gitea_tools_rooted_namespace_cannot_reach_target_repository(self):
|
||||
"""No canonical root configured → session stays pinned to Gitea-Tools."""
|
||||
profile = _profile("author", allowed_repositories=[INSTALL_SLUG])
|
||||
with patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop(crr.CANONICAL_ROOT_ENV, None)
|
||||
resolved = srv._trusted_session_repository(
|
||||
profile, "prgs", for_mutation=True
|
||||
)
|
||||
self.assertEqual(resolved["repository"], INSTALL_REPO)
|
||||
self.assertNotEqual(resolved["repository"], TARGET_REPO)
|
||||
|
||||
def test_target_rooted_namespace_binds_target_repository(self):
|
||||
profile = _profile(
|
||||
"author",
|
||||
canonical_root=self.target_root,
|
||||
allowed_repositories=[TARGET_SLUG],
|
||||
)
|
||||
resolved = srv._trusted_session_repository(profile, "prgs", for_mutation=True)
|
||||
self.assertEqual(resolved["org"], TARGET_ORG)
|
||||
self.assertEqual(resolved["repository"], TARGET_REPO)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# C. Reconciler branch deletion
|
||||
# ===========================================================================
|
||||
class TestDeleteBranchRepositoryBinding(_ServerHarness):
|
||||
"""The binding guard must consult the canonical root, not PROJECT_ROOT.
|
||||
|
||||
No branch is ever deleted here: only the pre-deletion binding guard is
|
||||
exercised.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
|
||||
self._write_config(
|
||||
{
|
||||
"prgs-reconciler": _profile(
|
||||
"reconciler",
|
||||
canonical_root=self.target_root,
|
||||
allowed_repositories=[TARGET_SLUG],
|
||||
),
|
||||
"gt-reconciler": _profile(
|
||||
"reconciler", allowed_repositories=[INSTALL_SLUG]
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
def test_target_rooted_reconciler_validates_target_deletion(self):
|
||||
with patch.dict(os.environ, self._env("prgs-reconciler"), clear=False):
|
||||
block = srv._delete_branch_repository_binding_block(
|
||||
"prgs", org=TARGET_ORG, repo=TARGET_REPO
|
||||
)
|
||||
self.assertIsNone(block, block)
|
||||
|
||||
def test_target_rooted_reconciler_fails_closed_for_install_repository(self):
|
||||
with patch.dict(os.environ, self._env("prgs-reconciler"), clear=False):
|
||||
block = srv._delete_branch_repository_binding_block(
|
||||
"prgs", org=INSTALL_ORG, repo=INSTALL_REPO
|
||||
)
|
||||
self.assertIsNotNone(block)
|
||||
self.assertEqual(block["blocker_kind"], "repository_binding")
|
||||
self.assertFalse(block["performed"])
|
||||
|
||||
def test_install_rooted_reconciler_fails_closed_for_target_repository(self):
|
||||
with patch.dict(os.environ, self._env("gt-reconciler"), clear=False):
|
||||
os.environ.pop(crr.CANONICAL_ROOT_ENV, None)
|
||||
block = srv._delete_branch_repository_binding_block(
|
||||
"prgs", org=TARGET_ORG, repo=TARGET_REPO
|
||||
)
|
||||
self.assertIsNotNone(block)
|
||||
self.assertEqual(block["blocker_kind"], "repository_binding")
|
||||
|
||||
def test_env_configured_canonical_root_is_honoured(self):
|
||||
env = self._env("gt-reconciler")
|
||||
env[crr.CANONICAL_ROOT_ENV] = self.target_root
|
||||
with patch.dict(os.environ, env, clear=False):
|
||||
allowed = srv._delete_branch_repository_binding_block(
|
||||
"prgs", org=TARGET_ORG, repo=TARGET_REPO
|
||||
)
|
||||
blocked = srv._delete_branch_repository_binding_block(
|
||||
"prgs", org=INSTALL_ORG, repo=INSTALL_REPO
|
||||
)
|
||||
self.assertIsNone(allowed, allowed)
|
||||
self.assertIsNotNone(blocked)
|
||||
|
||||
def test_unresolvable_canonical_root_fails_closed(self):
|
||||
env = self._env("gt-reconciler")
|
||||
env[crr.CANONICAL_ROOT_ENV] = os.path.join(self.tmp, "does-not-exist")
|
||||
with patch.dict(os.environ, env, clear=False):
|
||||
block = srv._delete_branch_repository_binding_block(
|
||||
"prgs", org=TARGET_ORG, repo=TARGET_REPO
|
||||
)
|
||||
self.assertIsNotNone(block)
|
||||
self.assertEqual(block["blocker_kind"], "repository_binding")
|
||||
|
||||
def test_no_request_parameter_can_override_canonical_identity(self):
|
||||
"""Every explicit coordinate that is not the canonical target is
|
||||
refused; none of them can *establish* the binding."""
|
||||
# Both repositories share an org, so a wrong *org* case must name a
|
||||
# genuinely different owner to be meaningful.
|
||||
cases = [
|
||||
(INSTALL_ORG, INSTALL_REPO),
|
||||
(TARGET_ORG, INSTALL_REPO),
|
||||
(TARGET_ORG, "some-other-repo"),
|
||||
("someone-else", TARGET_REPO),
|
||||
]
|
||||
with patch.dict(os.environ, self._env("prgs-reconciler"), clear=False):
|
||||
for org, repo in cases:
|
||||
with self.subTest(org=org, repo=repo):
|
||||
block = srv._delete_branch_repository_binding_block(
|
||||
"prgs", org=org, repo=repo
|
||||
)
|
||||
self.assertIsNotNone(block, f"{org}/{repo} must fail closed")
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# D. Parity semantics
|
||||
# ===========================================================================
|
||||
class TestTargetRepositoryParityAssessment(unittest.TestCase):
|
||||
"""Server-implementation parity is preserved; target parity is additive."""
|
||||
|
||||
def setUp(self):
|
||||
self._dir = tempfile.TemporaryDirectory()
|
||||
self.tmp = self._dir.name
|
||||
|
||||
def tearDown(self):
|
||||
self._dir.cleanup()
|
||||
|
||||
def _target(self) -> str:
|
||||
return _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
|
||||
|
||||
def test_unconfigured_target_is_reported_not_stale(self):
|
||||
got = master_parity_gate.assess_target_repository_parity(
|
||||
canonical_root=None, source=None
|
||||
)
|
||||
self.assertFalse(got["configured"])
|
||||
self.assertFalse(got["stale"])
|
||||
self.assertIsNone(got["canonical_repository_root"])
|
||||
|
||||
def test_configured_target_reports_checkout_head_and_slug(self):
|
||||
root = self._target()
|
||||
head = _git(root, "rev-parse", "HEAD")
|
||||
got = master_parity_gate.assess_target_repository_parity(
|
||||
canonical_root=root, source="profile.canonical_repository_root"
|
||||
)
|
||||
self.assertTrue(got["configured"])
|
||||
self.assertEqual(got["canonical_repository_root"], root)
|
||||
self.assertEqual(got["checkout_head"], head)
|
||||
self.assertEqual(got["repository_slug"], TARGET_SLUG)
|
||||
|
||||
def test_target_stale_when_remote_tracking_ref_is_ahead(self):
|
||||
root = self._target()
|
||||
head = _git(root, "rev-parse", "HEAD")
|
||||
# Simulate a fetched remote-tracking ref that has advanced.
|
||||
_git(root, "checkout", "-q", "-b", "advanced")
|
||||
(Path(root) / "next.md").write_text("next\n")
|
||||
_git(root, "add", "next.md")
|
||||
_git(root, "commit", "-q", "-m", "advance")
|
||||
advanced = _git(root, "rev-parse", "HEAD")
|
||||
_git(root, "update-ref", "refs/remotes/origin/master", advanced)
|
||||
_git(root, "checkout", "-q", "--detach", head)
|
||||
got = master_parity_gate.assess_target_repository_parity(
|
||||
canonical_root=root, source="profile.canonical_repository_root"
|
||||
)
|
||||
self.assertEqual(got["checkout_head"], head)
|
||||
self.assertEqual(got["remote_tracking_head"], advanced)
|
||||
self.assertTrue(got["stale"])
|
||||
|
||||
def test_missing_root_is_not_determinable_and_fails_closed(self):
|
||||
got = master_parity_gate.assess_target_repository_parity(
|
||||
canonical_root=os.path.join(self.tmp, "absent"),
|
||||
source="profile.canonical_repository_root",
|
||||
)
|
||||
self.assertTrue(got["configured"])
|
||||
self.assertFalse(got["determinable"])
|
||||
self.assertTrue(got["reasons"])
|
||||
|
||||
|
||||
class TestParityToolEvidenceDimensions(_ServerHarness):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
|
||||
self._write_config(
|
||||
{
|
||||
"prgs-author": _profile(
|
||||
"author",
|
||||
canonical_root=self.target_root,
|
||||
allowed_repositories=[TARGET_SLUG],
|
||||
)
|
||||
}
|
||||
)
|
||||
|
||||
def test_existing_server_parity_semantics_are_unchanged(self):
|
||||
with patch.dict(os.environ, self._env("prgs-author"), clear=False):
|
||||
result = srv.gitea_assess_master_parity(remote="prgs")
|
||||
# startup_head/current_head keep their Gitea-Tools implementation
|
||||
# meaning and must not be redefined to the target repository.
|
||||
self.assertEqual(result["process_root"], srv.PROJECT_ROOT)
|
||||
self.assertEqual(
|
||||
result["startup_head"], srv._STARTUP_PARITY.get("startup_head")
|
||||
)
|
||||
self.assertIn("in_parity", result)
|
||||
|
||||
def test_evidence_distinguishes_server_and_target_dimensions(self):
|
||||
with patch.dict(os.environ, self._env("prgs-author"), clear=False):
|
||||
result = srv.gitea_assess_master_parity(remote="prgs")
|
||||
server = result["server_implementation"]
|
||||
self.assertEqual(server["installation_root"], srv.PROJECT_ROOT)
|
||||
self.assertEqual(server["current_head"], result["current_head"])
|
||||
self.assertIn("stale", server)
|
||||
|
||||
target = result["target_repository"]
|
||||
self.assertTrue(target["configured"])
|
||||
self.assertEqual(target["canonical_repository_root"], self.target_root)
|
||||
self.assertEqual(target["repository_slug"], TARGET_SLUG)
|
||||
self.assertEqual(
|
||||
target["checkout_head"], _git(self.target_root, "rev-parse", "HEAD")
|
||||
)
|
||||
self.assertIn("stale", target)
|
||||
|
||||
def test_unconfigured_namespace_reports_target_as_unconfigured(self):
|
||||
self._write_config(
|
||||
{"prgs-author": _profile("author", allowed_repositories=[INSTALL_SLUG])}
|
||||
)
|
||||
env = self._env("prgs-author")
|
||||
with patch.dict(os.environ, env, clear=False):
|
||||
os.environ.pop(crr.CANONICAL_ROOT_ENV, None)
|
||||
result = srv.gitea_assess_master_parity(remote="prgs")
|
||||
self.assertFalse(result["target_repository"]["configured"])
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# E. Startup / configuration validation for a candidate namespace set
|
||||
# ===========================================================================
|
||||
class TestCandidateNamespaceConfiguration(_ServerHarness):
|
||||
"""Four repository-specific profiles for the target repository."""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
|
||||
self.profiles = {
|
||||
f"mcpcp-{role}": _profile(
|
||||
role,
|
||||
canonical_root=self.target_root,
|
||||
allowed_repositories=[TARGET_SLUG],
|
||||
)
|
||||
for role in ("author", "reviewer", "merger", "reconciler")
|
||||
}
|
||||
self._write_config(self.profiles)
|
||||
|
||||
def test_configuration_audit_succeeds(self):
|
||||
with patch.dict(os.environ, self._env("mcpcp-author"), clear=False):
|
||||
audit = srv.gitea_audit_config()
|
||||
self.assertTrue(audit.get("configured"))
|
||||
names = {row["name"] for row in audit["profiles"]}
|
||||
self.assertEqual(names, set(self.profiles))
|
||||
|
||||
def test_no_inline_credentials_are_present(self):
|
||||
raw = json.loads(Path(self.config_path).read_text())
|
||||
for name, profile in raw["profiles"].items():
|
||||
with self.subTest(profile=name):
|
||||
self.assertNotIn("token", profile)
|
||||
self.assertNotIn("password", profile)
|
||||
self.assertNotIn("secret", profile)
|
||||
self.assertIn(profile["auth"]["type"], ("env", "keychain"))
|
||||
|
||||
def test_bind_time_validation_succeeds_for_target_repository(self):
|
||||
for name, profile in self.profiles.items():
|
||||
with self.subTest(profile=name):
|
||||
resolved = srv._trusted_session_repository(
|
||||
profile, "prgs", for_mutation=True
|
||||
)
|
||||
self.assertEqual(resolved["reasons"], [])
|
||||
self.assertEqual(resolved["org"], TARGET_ORG)
|
||||
self.assertEqual(resolved["repository"], TARGET_REPO)
|
||||
|
||||
def test_wrong_repository_root_fails_closed(self):
|
||||
wrong_root = _init_repo(Path(self.tmp) / "wrong", INSTALL_URL)
|
||||
profile = _profile(
|
||||
"author", canonical_root=wrong_root, allowed_repositories=[TARGET_SLUG]
|
||||
)
|
||||
resolved = srv._trusted_session_repository(profile, "prgs", for_mutation=True)
|
||||
self.assertIsNone(resolved["repository"])
|
||||
self.assertTrue(resolved["reasons"])
|
||||
|
||||
def test_existing_install_profiles_are_unchanged_in_behaviour(self):
|
||||
profile = _profile("author", allowed_repositories=[INSTALL_SLUG])
|
||||
resolved = srv._trusted_session_repository(profile, "prgs", for_mutation=True)
|
||||
self.assertEqual(resolved["org"], INSTALL_ORG)
|
||||
self.assertEqual(resolved["repository"], INSTALL_REPO)
|
||||
|
||||
def _denied(self, profile: dict, operation: str) -> bool:
|
||||
ok, _ = gitea_config.check_operation(
|
||||
operation,
|
||||
profile["allowed_operations"],
|
||||
profile["forbidden_operations"],
|
||||
)
|
||||
return not ok
|
||||
|
||||
def test_role_separation_is_enforced(self):
|
||||
expectations = {
|
||||
"author": [
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.merge",
|
||||
"gitea.pr.request_changes",
|
||||
],
|
||||
"reviewer": ["gitea.pr.create", "gitea.branch.push", "gitea.pr.merge"],
|
||||
"merger": [
|
||||
"gitea.pr.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.request_changes",
|
||||
],
|
||||
"reconciler": [
|
||||
"gitea.pr.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.merge",
|
||||
"gitea.pr.request_changes",
|
||||
],
|
||||
}
|
||||
for role, denied_ops in expectations.items():
|
||||
profile = self.profiles[f"mcpcp-{role}"]
|
||||
for op in denied_ops:
|
||||
with self.subTest(role=role, operation=op):
|
||||
self.assertTrue(
|
||||
self._denied(profile, op),
|
||||
f"{role} must not be permitted {op}",
|
||||
)
|
||||
|
||||
def test_each_role_retains_its_own_capability(self):
|
||||
permitted = {
|
||||
"author": "gitea.pr.create",
|
||||
"reviewer": "gitea.pr.approve",
|
||||
"merger": "gitea.pr.merge",
|
||||
"reconciler": "gitea.branch.delete",
|
||||
}
|
||||
for role, op in permitted.items():
|
||||
with self.subTest(role=role, operation=op):
|
||||
self.assertFalse(self._denied(self.profiles[f"mcpcp-{role}"], op))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,423 @@
|
||||
"""Issue #706: immutable startup/profile canonical_repository_root capability.
|
||||
|
||||
Cross-repository MCP namespaces (e.g. ``eagenda-author``) must bind their
|
||||
canonical repository root to the *configured target repository*, not to the
|
||||
Gitea-Tools install checkout the server script lives in. These tests prove:
|
||||
|
||||
* the configured binding is resolved from profile/env with env precedence,
|
||||
* the target repository identity and git common-directory membership are
|
||||
validated (AC3),
|
||||
* the branches-only guard (#274) is enforced *inside the target repo* (AC4),
|
||||
* missing / conflicting / forged bindings fail closed (AC5),
|
||||
* prgs and mdcps namespaces stay simultaneously isolated (AC6),
|
||||
* the session binding pins the canonical root immutably,
|
||||
* no weakening of the install-root single-repo default (AC8).
|
||||
|
||||
Uses two distinct real git repositories/worktrees (AC7).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import canonical_repository_root as crr # noqa: E402
|
||||
import gitea_config # noqa: E402
|
||||
import namespace_workspace_binding as nwb # noqa: E402
|
||||
import session_context_binding as session_ctx # noqa: E402
|
||||
|
||||
|
||||
def _git(cwd: str, *args: str) -> str:
|
||||
res = subprocess.run(
|
||||
["git", "-C", cwd, *args],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
return res.stdout.strip()
|
||||
|
||||
|
||||
def _init_repo(path: Path, remote_url: str, *, remote_name: str = "origin") -> str:
|
||||
"""Create a real git repo with one commit and a remote; return realpath root."""
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
_git(str(path), "init", "-q")
|
||||
_git(str(path), "config", "user.email", "[email protected]")
|
||||
_git(str(path), "config", "user.name", "Test")
|
||||
_git(str(path), "remote", "add", remote_name, remote_url)
|
||||
(path / "README.md").write_text("seed\n")
|
||||
_git(str(path), "add", "README.md")
|
||||
_git(str(path), "commit", "-q", "-m", "seed")
|
||||
return os.path.realpath(str(path))
|
||||
|
||||
|
||||
def _add_worktree(repo_root: str, worktree_path: Path, branch: str) -> str:
|
||||
_git(repo_root, "worktree", "add", "-q", "-b", branch, str(worktree_path))
|
||||
return os.path.realpath(str(worktree_path))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# configured_canonical_root: resolution + precedence
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestConfiguredCanonicalRoot(unittest.TestCase):
|
||||
def test_unconfigured_returns_none(self):
|
||||
value, source = crr.configured_canonical_root({}, {})
|
||||
self.assertIsNone(value)
|
||||
self.assertIsNone(source)
|
||||
|
||||
def test_profile_field_used(self):
|
||||
value, source = crr.configured_canonical_root(
|
||||
{"canonical_repository_root": "/repo/eAgenda"}, {}
|
||||
)
|
||||
self.assertEqual(value, "/repo/eAgenda")
|
||||
self.assertIn("profile", source)
|
||||
|
||||
def test_env_overrides_profile(self):
|
||||
value, source = crr.configured_canonical_root(
|
||||
{"canonical_repository_root": "/repo/eAgenda"},
|
||||
{crr.CANONICAL_ROOT_ENV: "/repo/other"},
|
||||
)
|
||||
self.assertEqual(value, "/repo/other")
|
||||
self.assertIn(crr.CANONICAL_ROOT_ENV, source)
|
||||
|
||||
def test_blank_values_ignored(self):
|
||||
value, source = crr.configured_canonical_root(
|
||||
{"canonical_repository_root": " "},
|
||||
{crr.CANONICAL_ROOT_ENV: ""},
|
||||
)
|
||||
self.assertIsNone(value)
|
||||
self.assertIsNone(source)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# assess_canonical_repository_root: default (single-repo) path
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestUnconfiguredFallback(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.tmp = self._tmp.name
|
||||
|
||||
def tearDown(self):
|
||||
self._tmp.cleanup()
|
||||
|
||||
def test_unconfigured_falls_back_to_process_root(self):
|
||||
root = _init_repo(
|
||||
Path(self.tmp) / "install",
|
||||
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
|
||||
)
|
||||
got = crr.assess_canonical_repository_root(
|
||||
configured_value=None,
|
||||
source=None,
|
||||
expected_slug=None,
|
||||
process_project_root=root,
|
||||
)
|
||||
self.assertTrue(got["proven"])
|
||||
self.assertFalse(got["block"])
|
||||
self.assertFalse(got["configured"])
|
||||
self.assertEqual(got["canonical_repo_root"], root)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# assess_canonical_repository_root: configured cross-repo path
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestConfiguredCrossRepo(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.tmp = Path(self._tmp.name)
|
||||
self.install = _init_repo(
|
||||
self.tmp / "Gitea-Tools",
|
||||
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
|
||||
)
|
||||
self.target = _init_repo(
|
||||
self.tmp / "mcp-control-plane",
|
||||
"https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git",
|
||||
)
|
||||
|
||||
def tearDown(self):
|
||||
self._tmp.cleanup()
|
||||
|
||||
def test_valid_configured_root_binds_to_target(self):
|
||||
got = crr.assess_canonical_repository_root(
|
||||
configured_value=self.target,
|
||||
source="profile canonical_repository_root",
|
||||
expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
|
||||
process_project_root=self.install,
|
||||
)
|
||||
self.assertTrue(got["proven"], got.get("reasons"))
|
||||
self.assertFalse(got["block"])
|
||||
self.assertTrue(got["configured"])
|
||||
self.assertEqual(got["canonical_repo_root"], self.target)
|
||||
self.assertNotEqual(got["canonical_repo_root"], self.install)
|
||||
|
||||
def test_missing_path_fails_closed(self):
|
||||
got = crr.assess_canonical_repository_root(
|
||||
configured_value=str(self.tmp / "does-not-exist"),
|
||||
source="profile canonical_repository_root",
|
||||
expected_slug=None,
|
||||
process_project_root=self.install,
|
||||
)
|
||||
self.assertTrue(got["block"])
|
||||
self.assertFalse(got["proven"])
|
||||
self.assertTrue(any("exist" in r for r in got["reasons"]))
|
||||
|
||||
def test_non_git_path_fails_closed(self):
|
||||
plain = self.tmp / "plain"
|
||||
plain.mkdir()
|
||||
got = crr.assess_canonical_repository_root(
|
||||
configured_value=str(plain),
|
||||
source="profile canonical_repository_root",
|
||||
expected_slug=None,
|
||||
process_project_root=self.install,
|
||||
)
|
||||
self.assertTrue(got["block"])
|
||||
self.assertTrue(any("git" in r for r in got["reasons"]))
|
||||
|
||||
def test_identity_mismatch_fails_closed(self):
|
||||
# Configured root is the install repo, but session expects the target
|
||||
# repo identity: a forged/conflicting binding.
|
||||
got = crr.assess_canonical_repository_root(
|
||||
configured_value=self.install,
|
||||
source="profile canonical_repository_root",
|
||||
expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
|
||||
process_project_root=self.install,
|
||||
)
|
||||
self.assertTrue(got["block"])
|
||||
self.assertTrue(any("identity" in r for r in got["reasons"]))
|
||||
|
||||
def test_unprovable_identity_blocks_when_required(self):
|
||||
noremote = _init_repo(self.tmp / "noremote-src", "x", remote_name="origin")
|
||||
_git(noremote, "remote", "remove", "origin")
|
||||
got = crr.assess_canonical_repository_root(
|
||||
configured_value=noremote,
|
||||
source="profile canonical_repository_root",
|
||||
expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
|
||||
process_project_root=self.install,
|
||||
require_binding=True,
|
||||
)
|
||||
self.assertTrue(got["block"])
|
||||
|
||||
def test_repository_identity_slug_reads_remote(self):
|
||||
self.assertEqual(
|
||||
crr.repository_identity_slug(self.target),
|
||||
"Scaled-Tech-Consulting/mcp-control-plane",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Workspace membership + branches-only enforced inside the TARGET repo (AC4)
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestNamespaceContextUsesConfiguredRoot(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.tmp = Path(self._tmp.name)
|
||||
self.install = _init_repo(
|
||||
self.tmp / "Gitea-Tools",
|
||||
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
|
||||
)
|
||||
self.target = _init_repo(
|
||||
self.tmp / "mcp-control-plane",
|
||||
"https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git",
|
||||
)
|
||||
(Path(self.target) / "branches").mkdir()
|
||||
self.target_wt = _add_worktree(
|
||||
self.target,
|
||||
Path(self.target) / "branches" / "author-issue-1",
|
||||
"feat/issue-1",
|
||||
)
|
||||
|
||||
def tearDown(self):
|
||||
self._tmp.cleanup()
|
||||
|
||||
def test_context_canonical_root_is_configured_target(self):
|
||||
ctx = nwb.resolve_namespace_mutation_context(
|
||||
role_kind="author",
|
||||
worktree_path=self.target_wt,
|
||||
process_project_root=self.install,
|
||||
env={},
|
||||
configured_canonical_root=self.target,
|
||||
)
|
||||
self.assertEqual(ctx["canonical_repo_root"], self.target)
|
||||
self.assertFalse(ctx["roots_aligned"])
|
||||
|
||||
def test_target_worktree_is_member_of_target_root(self):
|
||||
got = nwb.amw.assess_workspace_repo_membership(
|
||||
workspace_path=self.target_wt,
|
||||
canonical_repo_root=self.target,
|
||||
)
|
||||
self.assertTrue(got["proven"], got.get("reasons"))
|
||||
|
||||
def test_target_worktree_not_member_of_install_root(self):
|
||||
got = nwb.amw.assess_workspace_repo_membership(
|
||||
workspace_path=self.target_wt,
|
||||
canonical_repo_root=self.install,
|
||||
)
|
||||
self.assertTrue(got["block"])
|
||||
|
||||
def test_branches_guard_passes_inside_target(self):
|
||||
assessment = nwb.assess_namespace_mutation_workspace(
|
||||
role_kind="author",
|
||||
worktree_path=self.target_wt,
|
||||
worktree=None,
|
||||
process_project_root=self.install,
|
||||
env={},
|
||||
current_branch="feat/issue-1",
|
||||
configured_canonical_root=self.target,
|
||||
)
|
||||
self.assertFalse(assessment["block"], assessment.get("reasons"))
|
||||
self.assertEqual(assessment["canonical_repo_root"], self.target)
|
||||
|
||||
def test_target_control_checkout_blocks_branches_guard(self):
|
||||
assessment = nwb.assess_namespace_mutation_workspace(
|
||||
role_kind="author",
|
||||
worktree_path=self.target, # stable target checkout, not branches/
|
||||
worktree=None,
|
||||
process_project_root=self.install,
|
||||
env={},
|
||||
current_branch="master",
|
||||
configured_canonical_root=self.target,
|
||||
)
|
||||
self.assertTrue(assessment["block"])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Immutable session pin (AC1/AC5)
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestSessionCanonicalRootPin(unittest.TestCase):
|
||||
def setUp(self):
|
||||
os.environ["PYTEST_CURRENT_TEST"] = "t"
|
||||
session_ctx._reset_session_context_for_testing()
|
||||
|
||||
def tearDown(self):
|
||||
session_ctx._reset_session_context_for_testing()
|
||||
|
||||
def test_seed_stores_canonical_root(self):
|
||||
ctx = session_ctx.seed_session_context_if_unbound(
|
||||
profile_name="mcp-control-plane-author",
|
||||
remote="prgs",
|
||||
host="gitea.prgs.cc",
|
||||
identity="svc",
|
||||
repository="mcp-control-plane",
|
||||
org="Scaled-Tech-Consulting",
|
||||
canonical_repository_root="/repo/mcp-control-plane",
|
||||
)
|
||||
self.assertEqual(ctx["canonical_repository_root"], "/repo/mcp-control-plane")
|
||||
|
||||
def test_canonical_root_drift_fails_closed(self):
|
||||
session_ctx.seed_session_context_if_unbound(
|
||||
profile_name="mcp-control-plane-author",
|
||||
remote="prgs",
|
||||
host="gitea.prgs.cc",
|
||||
identity="svc",
|
||||
repository="mcp-control-plane",
|
||||
org="Scaled-Tech-Consulting",
|
||||
canonical_repository_root="/repo/mcp-control-plane",
|
||||
)
|
||||
got = session_ctx.assess_session_context(
|
||||
profile_name="mcp-control-plane-author",
|
||||
remote="prgs",
|
||||
canonical_repository_root="/repo/forged-elsewhere",
|
||||
)
|
||||
self.assertTrue(got["block"])
|
||||
self.assertTrue(any("canonical" in r for r in got["reasons"]))
|
||||
|
||||
def test_matching_canonical_root_passes(self):
|
||||
session_ctx.seed_session_context_if_unbound(
|
||||
profile_name="mcp-control-plane-author",
|
||||
remote="prgs",
|
||||
host="gitea.prgs.cc",
|
||||
identity="svc",
|
||||
repository="mcp-control-plane",
|
||||
org="Scaled-Tech-Consulting",
|
||||
canonical_repository_root="/repo/mcp-control-plane",
|
||||
)
|
||||
got = session_ctx.assess_session_context(
|
||||
profile_name="mcp-control-plane-author",
|
||||
remote="prgs",
|
||||
canonical_repository_root="/repo/mcp-control-plane",
|
||||
)
|
||||
self.assertFalse(got["block"], got["reasons"])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Simultaneous prgs / mdcps isolation (AC6)
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestSimultaneousIsolation(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.tmp = Path(self._tmp.name)
|
||||
self.prgs = _init_repo(
|
||||
self.tmp / "prgs-repo",
|
||||
"https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git",
|
||||
)
|
||||
self.mdcps = _init_repo(
|
||||
self.tmp / "mdcps-repo",
|
||||
"https://gitea.dadeschools.net/dadeschools/eAgenda.git",
|
||||
)
|
||||
|
||||
def tearDown(self):
|
||||
self._tmp.cleanup()
|
||||
|
||||
def test_each_namespace_binds_its_own_target(self):
|
||||
a = crr.assess_canonical_repository_root(
|
||||
configured_value=self.prgs,
|
||||
source="env",
|
||||
expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
|
||||
process_project_root=self.tmp.as_posix(),
|
||||
)
|
||||
b = crr.assess_canonical_repository_root(
|
||||
configured_value=self.mdcps,
|
||||
source="env",
|
||||
expected_slug="dadeschools/eAgenda",
|
||||
process_project_root=self.tmp.as_posix(),
|
||||
)
|
||||
self.assertEqual(a["canonical_repo_root"], self.prgs)
|
||||
self.assertEqual(b["canonical_repo_root"], self.mdcps)
|
||||
self.assertNotEqual(a["canonical_repo_root"], b["canonical_repo_root"])
|
||||
|
||||
def test_cross_wired_identity_blocks(self):
|
||||
# prgs path claimed under the mdcps identity → forged binding.
|
||||
got = crr.assess_canonical_repository_root(
|
||||
configured_value=self.prgs,
|
||||
source="env",
|
||||
expected_slug="dadeschools/eAgenda",
|
||||
process_project_root=self.tmp.as_posix(),
|
||||
)
|
||||
self.assertTrue(got["block"])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Config validation of the profile field (AC5: malformed bindings fail closed)
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestConfigValidation(unittest.TestCase):
|
||||
def test_absent_is_allowed(self):
|
||||
# single-repo default: no field configured
|
||||
gitea_config._validate_canonical_repository_root("p", None)
|
||||
|
||||
def test_valid_absolute_path_ok(self):
|
||||
gitea_config._validate_canonical_repository_root(
|
||||
"p", "/Users/x/Development/mcp-control-plane"
|
||||
)
|
||||
|
||||
def test_relative_path_rejected(self):
|
||||
with self.assertRaises(gitea_config.ConfigError):
|
||||
gitea_config._validate_canonical_repository_root(
|
||||
"p", "relative/path"
|
||||
)
|
||||
|
||||
def test_empty_string_rejected(self):
|
||||
with self.assertRaises(gitea_config.ConfigError):
|
||||
gitea_config._validate_canonical_repository_root("p", " ")
|
||||
|
||||
def test_non_string_rejected(self):
|
||||
with self.assertRaises(gitea_config.ConfigError):
|
||||
gitea_config._validate_canonical_repository_root("p", ["/x"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,221 @@
|
||||
"""Issue #706 F1 integration regression (review 457).
|
||||
|
||||
The unit tests in ``test_issue_706_canonical_repository_root.py`` exercise
|
||||
``crr.assess_canonical_repository_root`` / ``session_ctx.seed_session_context_if_unbound``
|
||||
with a *preselected* slug, so they never drive the real end-to-end path that
|
||||
review 457 found broken:
|
||||
|
||||
_seed_session_context -> _trusted_session_repository -> _workspace_repository_slug
|
||||
-> _local_git_remote_url (cwd=PROJECT_ROOT, always Gitea-Tools)
|
||||
|
||||
Before the fix, the session repository identity was pinned from the *install*
|
||||
checkout remote even when a cross-repository ``canonical_repository_root`` was
|
||||
configured to an external repository (e.g. mcp-control-plane). The mutation
|
||||
preflight then derived ``expected_slug`` from that install-derived pin and
|
||||
``_enforce_canonical_repository_root`` failed closed on a self-inflicted
|
||||
identity mismatch, so the stated cross-repo namespaces stayed blocked.
|
||||
|
||||
These tests use two *real* temporary git repositories and drive the live
|
||||
``mcp_server._seed_session_context`` and ``mcp_server._enforce_canonical_repository_root``
|
||||
functions, asserting the session pins the *configured target* identity and that
|
||||
enforcement accepts the target worktree — while every fail-closed property is
|
||||
preserved.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import mcp_server # noqa: E402 # loads gitea_mcp_server.py into this namespace
|
||||
import canonical_repository_root as crr # noqa: E402
|
||||
import session_context_binding as session_ctx # noqa: E402
|
||||
|
||||
INSTALL_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
|
||||
TARGET_SLUG = "Scaled-Tech-Consulting/mcp-control-plane"
|
||||
TARGET_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git"
|
||||
OTHER_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/other-repo.git"
|
||||
|
||||
|
||||
def _git(cwd: str, *args: str) -> str:
|
||||
return subprocess.run(
|
||||
["git", "-C", cwd, *args], capture_output=True, text=True, check=True
|
||||
).stdout.strip()
|
||||
|
||||
|
||||
def _init_repo(path: Path, remote_url: str) -> str:
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
_git(str(path), "init", "-q")
|
||||
_git(str(path), "config", "user.email", "[email protected]")
|
||||
_git(str(path), "config", "user.name", "Test")
|
||||
_git(str(path), "remote", "add", "prgs", remote_url)
|
||||
(path / "README.md").write_text("seed\n")
|
||||
_git(str(path), "add", "README.md")
|
||||
_git(str(path), "commit", "-q", "-m", "seed")
|
||||
return os.path.realpath(str(path))
|
||||
|
||||
|
||||
def _add_worktree(repo_root: str, wt: Path, branch: str) -> str:
|
||||
_git(repo_root, "worktree", "add", "-q", "-b", branch, str(wt))
|
||||
return os.path.realpath(str(wt))
|
||||
|
||||
|
||||
def _profile(role: str, *, canonical: str | None = None,
|
||||
allowed=(TARGET_SLUG,)) -> dict:
|
||||
p = {
|
||||
"profile_name": f"mcp-control-plane-{role}",
|
||||
"role": role,
|
||||
"username": "svc",
|
||||
"allowed_operations": ["gitea.read"],
|
||||
"forbidden_operations": [],
|
||||
"allowed_repositories": list(allowed),
|
||||
}
|
||||
if canonical is not None:
|
||||
p["canonical_repository_root"] = canonical
|
||||
return p
|
||||
|
||||
|
||||
class _Base(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
tmp = Path(self._tmp.name)
|
||||
self.install = _init_repo(tmp / "Gitea-Tools", INSTALL_URL)
|
||||
self.target = _init_repo(tmp / "mcp-control-plane", TARGET_URL)
|
||||
(Path(self.target) / "branches").mkdir()
|
||||
self.target_wt = _add_worktree(
|
||||
self.target, Path(self.target) / "branches" / "author-issue-1",
|
||||
"feat/issue-1",
|
||||
)
|
||||
# PROJECT_ROOT and the install git remote are the Gitea-Tools install
|
||||
# checkout — exactly the source that (mis)seeded the session before.
|
||||
self._p_root = mock.patch.object(mcp_server, "PROJECT_ROOT", self.install)
|
||||
self._p_url = mock.patch.object(
|
||||
mcp_server, "_local_git_remote_url", return_value=INSTALL_URL
|
||||
)
|
||||
self._p_root.start()
|
||||
self._p_url.start()
|
||||
session_ctx._reset_session_context_for_testing()
|
||||
|
||||
def tearDown(self):
|
||||
mock.patch.stopall()
|
||||
session_ctx._reset_session_context_for_testing()
|
||||
|
||||
def _seed(self, profile, env=None):
|
||||
session_ctx._reset_session_context_for_testing()
|
||||
with mock.patch.object(mcp_server, "get_profile", return_value=profile), \
|
||||
mock.patch.dict(os.environ, env or {}, clear=False):
|
||||
return mcp_server._seed_session_context(
|
||||
profile=profile, remote="prgs", host="gitea.prgs.cc",
|
||||
identity="svc",
|
||||
)
|
||||
|
||||
def _enforce(self, profile, env=None):
|
||||
with mock.patch.object(mcp_server, "get_profile", return_value=profile), \
|
||||
mock.patch.dict(os.environ, env or {}, clear=False):
|
||||
mcp_server._enforce_canonical_repository_root(
|
||||
self.target_wt, remote="prgs"
|
||||
)
|
||||
|
||||
|
||||
class TestF1SeedsConfiguredTargetIdentity(_Base):
|
||||
def test_env_config_seeds_target_not_install(self):
|
||||
ctx = self._seed(
|
||||
_profile("reviewer"),
|
||||
env={crr.CANONICAL_ROOT_ENV: self.target},
|
||||
)
|
||||
self.assertEqual(ctx["org"], "Scaled-Tech-Consulting")
|
||||
self.assertEqual(ctx["repository"], "mcp-control-plane")
|
||||
# Regression guard: must NOT be the install repo.
|
||||
self.assertNotEqual(ctx["repository"], "Gitea-Tools")
|
||||
|
||||
def test_profile_field_seeds_target_not_install(self):
|
||||
ctx = self._seed(_profile("merger", canonical=self.target))
|
||||
self.assertEqual(ctx["org"], "Scaled-Tech-Consulting")
|
||||
self.assertEqual(ctx["repository"], "mcp-control-plane")
|
||||
self.assertNotEqual(ctx["repository"], "Gitea-Tools")
|
||||
|
||||
def test_enforce_accepts_target_after_seed_reviewer(self):
|
||||
prof = _profile("reviewer", canonical=self.target)
|
||||
self._seed(prof)
|
||||
# Would raise RuntimeError on the self-inflicted identity mismatch
|
||||
# before the fix.
|
||||
self._enforce(prof)
|
||||
|
||||
def test_enforce_accepts_target_after_seed_merger(self):
|
||||
prof = _profile("merger", canonical=self.target)
|
||||
self._seed(prof)
|
||||
self._enforce(prof)
|
||||
|
||||
def test_env_overrides_profile_for_seed(self):
|
||||
# profile points at install; env points at the real target → env wins.
|
||||
prof = _profile("reviewer", canonical=self.install,
|
||||
allowed=(TARGET_SLUG,))
|
||||
ctx = self._seed(prof, env={crr.CANONICAL_ROOT_ENV: self.target})
|
||||
self.assertEqual(ctx["repository"], "mcp-control-plane")
|
||||
|
||||
|
||||
class TestUnconfiguredDefaultUnchanged(_Base):
|
||||
def test_unconfigured_keeps_install_identity(self):
|
||||
prof = _profile("author", allowed=("Scaled-Tech-Consulting/Gitea-Tools",))
|
||||
ctx = self._seed(prof) # no env, no profile canonical field
|
||||
self.assertEqual(ctx["repository"], "Gitea-Tools")
|
||||
self.assertEqual(ctx["org"], "Scaled-Tech-Consulting")
|
||||
|
||||
|
||||
class TestFailClosed(_Base):
|
||||
def _trusted(self, profile, env=None, *, for_mutation=True):
|
||||
with mock.patch.object(mcp_server, "get_profile", return_value=profile), \
|
||||
mock.patch.dict(os.environ, env or {}, clear=False):
|
||||
return mcp_server._trusted_session_repository(
|
||||
profile, "prgs", for_mutation=for_mutation
|
||||
)
|
||||
|
||||
def test_nonexistent_configured_root_fails_closed(self):
|
||||
res = self._trusted(
|
||||
_profile("reviewer"),
|
||||
env={crr.CANONICAL_ROOT_ENV: self.target + "-missing"},
|
||||
)
|
||||
self.assertIsNone(res["repository"])
|
||||
self.assertTrue(res["reasons"])
|
||||
|
||||
def test_non_git_configured_root_fails_closed(self):
|
||||
plain = Path(self._tmp.name) / "plain"
|
||||
plain.mkdir()
|
||||
res = self._trusted(
|
||||
_profile("reviewer"), env={crr.CANONICAL_ROOT_ENV: str(plain)}
|
||||
)
|
||||
self.assertIsNone(res["repository"])
|
||||
self.assertTrue(any("git" in r for r in res["reasons"]))
|
||||
|
||||
def test_unallowlisted_target_identity_fails_closed(self):
|
||||
# Configured target is valid, but the profile does not authorize it.
|
||||
res = self._trusted(
|
||||
_profile("reviewer", canonical=self.target,
|
||||
allowed=("Scaled-Tech-Consulting/Gitea-Tools",)),
|
||||
)
|
||||
self.assertIsNone(res["repository"])
|
||||
self.assertTrue(any("scope" in r.lower() for r in res["reasons"]))
|
||||
|
||||
def test_forged_identity_conflict_fails_closed_at_enforce(self):
|
||||
# Seed the target, then present a *different* configured root at
|
||||
# enforcement time: the session pin no longer matches → fail closed.
|
||||
other = _init_repo(Path(self._tmp.name) / "other", OTHER_URL)
|
||||
prof_seed = _profile("reviewer", canonical=self.target)
|
||||
self._seed(prof_seed)
|
||||
prof_drift = _profile(
|
||||
"reviewer", canonical=other,
|
||||
allowed=(TARGET_SLUG, "Scaled-Tech-Consulting/other-repo"),
|
||||
)
|
||||
with self.assertRaises(RuntimeError):
|
||||
self._enforce(prof_drift)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,280 @@
|
||||
"""Regression matrix: session-bound mutation preflight resolves the workspace repo.
|
||||
|
||||
Root-cause follow-up to #733. ``_resolve`` (the actual API-targeting resolver)
|
||||
already prefers the workspace-aligned git remote over the ``REMOTES`` default
|
||||
(bare ``prgs`` → ``Scaled-Tech-Consulting/Timesheet``). The shared #604
|
||||
anti-stomp preflight (``_run_anti_stomp_preflight``) did **not** mirror that: for
|
||||
every mutation task whose call site omitted ``org``/``repo`` it filled the
|
||||
remote-wide ``REMOTES`` default, so the #530 repo guard false-positived
|
||||
``wrong_repo`` against a ``Gitea-Tools`` checkout and blocked native
|
||||
issue/PR/review/merge/lock/cleanup mutations.
|
||||
|
||||
The fix makes ``_run_anti_stomp_preflight`` prefer the trusted, git-remote-
|
||||
derived workspace identity for omitted coordinates on **session-bound** mutation
|
||||
tasks (marking those filled sides explicit, exactly like ``_resolve``), while
|
||||
``delete_branch`` keeps its explicit cross-repository *target* contract (#733)
|
||||
and continues to fail closed on omitted coordinates.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import mcp_server
|
||||
import remote_repo_guard
|
||||
|
||||
# Workspace is bound to Gitea-Tools even though the prgs REMOTES default repo is
|
||||
# Timesheet (the exact cross-repo scenario).
|
||||
LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
|
||||
|
||||
RECONCILER = {
|
||||
"profile_name": "prgs-reconciler",
|
||||
"role": "reconciler",
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.issue.comment", "gitea.pr.comment",
|
||||
"gitea.pr.close", "gitea.branch.delete",
|
||||
],
|
||||
"forbidden_operations": ["gitea.pr.create", "gitea.pr.merge"],
|
||||
"audit_label": "prgs-reconciler",
|
||||
}
|
||||
|
||||
# Session-bound mutation tasks (every MUTATION_TASK except the explicit-target
|
||||
# delete_branch) must auto-resolve the workspace repository for omitted coords.
|
||||
SESSION_BOUND_TASKS = [
|
||||
"create_issue",
|
||||
"comment_issue",
|
||||
"close_issue",
|
||||
"mark_issue",
|
||||
"lock_issue",
|
||||
"set_issue_labels",
|
||||
"create_label",
|
||||
"create_pr",
|
||||
"close_pr",
|
||||
"edit_pr",
|
||||
"commit_files",
|
||||
"cleanup_merged_pr_branch",
|
||||
"cleanup_stale_claims",
|
||||
"reconcile_merged_cleanups",
|
||||
"reconcile_already_landed_pr",
|
||||
"reconcile_close_superseded_pr",
|
||||
"post_heartbeat",
|
||||
"review_pr",
|
||||
"submit_pr_review",
|
||||
"merge_pr",
|
||||
]
|
||||
|
||||
|
||||
class _AntiStompResolutionBase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._remotes = patch.dict(mcp_server.REMOTES, {
|
||||
"prgs": {
|
||||
"host": "gitea.prgs.cc",
|
||||
"org": "Scaled-Tech-Consulting",
|
||||
"repo": "Timesheet",
|
||||
},
|
||||
})
|
||||
self._remotes.start()
|
||||
|
||||
def tearDown(self):
|
||||
patch.stopall()
|
||||
|
||||
def _capture_resolution(self, task, org, repo, *, local_url=LOCAL_GITEA_TOOLS_URL):
|
||||
"""Drive the live anti-stomp runner and capture the org/repo it resolved
|
||||
and passed to the pure assessor."""
|
||||
passthrough = {
|
||||
"allowed": True, "block": False, "blockers": [], "reasons": [],
|
||||
"exact_next_action": "proceed", "blocker_kind": None, "checks": {},
|
||||
}
|
||||
with patch.dict(
|
||||
os.environ, {"GITEA_TEST_FORCE_ANTI_STOMP": "1"}, clear=False
|
||||
), patch(
|
||||
"mcp_server._local_git_remote_url", return_value=local_url
|
||||
), patch(
|
||||
"mcp_server.get_profile", return_value=RECONCILER
|
||||
), patch.object(
|
||||
mcp_server.anti_stomp_preflight,
|
||||
"assess_anti_stomp_preflight",
|
||||
return_value=passthrough,
|
||||
) as m:
|
||||
mcp_server._run_anti_stomp_preflight(
|
||||
task, remote="prgs", org=org, repo=repo
|
||||
)
|
||||
self.assertTrue(m.called)
|
||||
return m.call_args.kwargs
|
||||
|
||||
|
||||
class TestSessionBoundResolution(_AntiStompResolutionBase):
|
||||
def test_omitted_coords_resolve_workspace_repo_and_mark_explicit(self):
|
||||
for task in SESSION_BOUND_TASKS:
|
||||
with self.subTest(task=task):
|
||||
kw = self._capture_resolution(task, None, None)
|
||||
self.assertEqual(
|
||||
kw["resolved_org"], "Scaled-Tech-Consulting", task
|
||||
)
|
||||
self.assertEqual(kw["resolved_repo"], "Gitea-Tools", task)
|
||||
# Workspace-filled sides are intentional alignment → explicit.
|
||||
self.assertTrue(kw["org_explicit"], task)
|
||||
self.assertTrue(kw["repo_explicit"], task)
|
||||
|
||||
def test_resolved_workspace_repo_passes_the_530_guard(self):
|
||||
kw = self._capture_resolution("create_pr", None, None)
|
||||
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||
remote="prgs",
|
||||
resolved_org=kw["resolved_org"],
|
||||
resolved_repo=kw["resolved_repo"],
|
||||
local_remote_url=LOCAL_GITEA_TOOLS_URL,
|
||||
org_explicit=kw["org_explicit"],
|
||||
repo_explicit=kw["repo_explicit"],
|
||||
)
|
||||
self.assertFalse(assessment["block"], assessment)
|
||||
|
||||
def test_explicit_caller_coords_are_preserved(self):
|
||||
kw = self._capture_resolution(
|
||||
"create_pr", "Scaled-Tech-Consulting", "Gitea-Tools"
|
||||
)
|
||||
self.assertEqual(kw["resolved_org"], "Scaled-Tech-Consulting")
|
||||
self.assertEqual(kw["resolved_repo"], "Gitea-Tools")
|
||||
self.assertTrue(kw["org_explicit"])
|
||||
self.assertTrue(kw["repo_explicit"])
|
||||
|
||||
def test_no_workspace_identity_falls_back_to_remotes_default(self):
|
||||
# No derivable local remote → best-effort REMOTES default preserved,
|
||||
# not marked explicit (so the guard still corroborates when able).
|
||||
kw = self._capture_resolution("create_pr", None, None, local_url=None)
|
||||
self.assertEqual(kw["resolved_repo"], "Timesheet")
|
||||
self.assertFalse(kw["org_explicit"])
|
||||
self.assertFalse(kw["repo_explicit"])
|
||||
|
||||
|
||||
class TestDeleteBranchContractPreserved(_AntiStompResolutionBase):
|
||||
"""delete_branch keeps its explicit-target contract (#733): omitted coords
|
||||
still resolve the remote-wide default and are NOT marked explicit, so the
|
||||
guard fails closed against a mismatched workspace."""
|
||||
|
||||
def test_omitted_coords_still_remote_default_not_explicit(self):
|
||||
kw = self._capture_resolution("delete_branch", None, None)
|
||||
self.assertEqual(kw["resolved_repo"], "Timesheet")
|
||||
self.assertFalse(kw["org_explicit"])
|
||||
self.assertFalse(kw["repo_explicit"])
|
||||
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||
remote="prgs",
|
||||
resolved_org=kw["resolved_org"],
|
||||
resolved_repo=kw["resolved_repo"],
|
||||
local_remote_url=LOCAL_GITEA_TOOLS_URL,
|
||||
org_explicit=kw["org_explicit"],
|
||||
repo_explicit=kw["repo_explicit"],
|
||||
)
|
||||
self.assertTrue(assessment["block"])
|
||||
|
||||
def test_explicit_coords_still_forwarded(self):
|
||||
kw = self._capture_resolution(
|
||||
"delete_branch", "Scaled-Tech-Consulting", "Gitea-Tools"
|
||||
)
|
||||
self.assertEqual(kw["resolved_repo"], "Gitea-Tools")
|
||||
self.assertTrue(kw["org_explicit"])
|
||||
self.assertTrue(kw["repo_explicit"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
|
||||
class TestAuthorCallSiteForwardsOrgRepo(_AntiStompResolutionBase):
|
||||
"""#735: author mutation entrypoints must pass org/repo into preflight."""
|
||||
|
||||
def _assert_call_forwards(self, fn_name: str, *args, **kwargs):
|
||||
import inspect
|
||||
fn = getattr(mcp_server, fn_name)
|
||||
# Force preflight path to raise with the kwargs it received so we can
|
||||
# prove org/repo were forwarded without performing the full mutation.
|
||||
captured = {}
|
||||
|
||||
def _capture(*a, **kw):
|
||||
captured.update(kw)
|
||||
captured["_args"] = a
|
||||
raise RuntimeError("capture-only")
|
||||
|
||||
with patch.object(mcp_server, "verify_preflight_purity", side_effect=_capture), \
|
||||
patch.object(mcp_server, "get_profile", return_value={
|
||||
"profile_name": "prgs-author",
|
||||
"role": "author",
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.issue.create", "gitea.issue.comment",
|
||||
"gitea.pr.create", "gitea.repo.commit", "gitea.branch.push",
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
"audit_label": "prgs-author",
|
||||
}), patch.object(
|
||||
mcp_server, "_resolve",
|
||||
return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
|
||||
), patch.object(
|
||||
mcp_server, "_auth", return_value="token fake",
|
||||
), patch.object(
|
||||
mcp_server.role_session_router,
|
||||
"check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []),
|
||||
), patch.object(
|
||||
mcp_server, "_namespace_mutation_block", return_value=None
|
||||
), patch.object(
|
||||
mcp_server, "_profile_permission_block", return_value=None
|
||||
), patch.object(
|
||||
mcp_server, "_production_guard_block_from_exc", return_value=None
|
||||
):
|
||||
try:
|
||||
fn(*args, **kwargs)
|
||||
except RuntimeError as exc:
|
||||
if "capture-only" not in str(exc):
|
||||
# Some tools re-raise; still require capture when preflight ran.
|
||||
if not captured:
|
||||
raise
|
||||
self.assertTrue(captured, f"{fn_name} never called verify_preflight_purity")
|
||||
self.assertEqual(captured.get("org"), "Scaled-Tech-Consulting", fn_name)
|
||||
self.assertEqual(captured.get("repo"), "Gitea-Tools", fn_name)
|
||||
|
||||
def test_create_issue_forwards_explicit_org_repo(self):
|
||||
self._assert_call_forwards(
|
||||
"gitea_create_issue",
|
||||
"title",
|
||||
body="body with acceptance criteria for gate",
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
worktree_path="/tmp/fake-wt",
|
||||
)
|
||||
|
||||
def test_create_pr_forwards_explicit_org_repo(self):
|
||||
self._assert_call_forwards(
|
||||
"gitea_create_pr",
|
||||
title="t",
|
||||
head="fix/issue-735-x",
|
||||
base="master",
|
||||
body="Closes #735",
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
worktree_path="/tmp/fake-wt",
|
||||
)
|
||||
|
||||
def test_mark_issue_forwards_explicit_org_repo(self):
|
||||
self._assert_call_forwards(
|
||||
"gitea_mark_issue",
|
||||
issue_number=735,
|
||||
action="start",
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
worktree_path="/tmp/fake-wt",
|
||||
)
|
||||
|
||||
def test_commit_files_forwards_explicit_org_repo(self):
|
||||
self._assert_call_forwards(
|
||||
"gitea_commit_files",
|
||||
files=[{"operation": "update", "path": "x", "content_plain": "y"}],
|
||||
message="m",
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
)
|
||||
Reference in New Issue
Block a user