From 943d40270e9ecdee48707eb09454f0321e8e4e14 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 15 Jul 2026 16:50:41 -0400 Subject: [PATCH] fix(session): bind workspace-verified repository scope at first bind (#714) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The independent review reproduced a real hole: the production first-bind path never pinned repository/org, so the drift checks it feeds were skipped. Root cause ---------- gitea_whoami, gitea_get_runtime_context, gitea_resolve_task_capability and gitea_activate_profile all seeded the session context without repository or org. First-bind-wins then made the mutation gate's later seed a no-op, and assess_session_context only compares repository/org when the bound fields are non-null. Result, reproduced in production order with the real REMOTES: after whoami: repository=None org=None same-host repo=Other-Tools -> NOT BLOCKED same-host org=Other-Org -> NOT BLOCKED cross-host -> blocked (this part always worked) Binding REMOTES defaults would not fix it: REMOTES['prgs'].repo is 'Timesheet' (a default *target*, not an authorization scope, see #530), so pinning it fails closed on every legitimate Gitea-Tools mutation. There was no trusted source for repository scope, so this adds one. Design ------ - New optional profile field `allowed_repositories`: canonical owner/repository slugs. An authorization boundary, never the binding itself. Config-only — no env var can widen or forge it. - The session repository is derived from the verified, workspace-aligned git remote, never from caller-supplied org/repo, and validated against that allowlist. The session binds immutably to exactly one canonical slug; org is derived from it. A profile authorizing several repositories still binds to the single verified one. - Every first-bind entry point now pins the same complete context. Activation rejects an unauthorized/unverifiable workspace. Mutations fail closed when repository/org is unverified, and a tool-level org/repo override that disagrees with the binding is rejected before the write. - A mutation request can no longer establish, complete, or replace the binding (it previously seeded from `org or REMOTES[...]`, i.e. caller-controlled). - Enforcement is opt-in per profile: a profile without the field keeps prior behaviour, so existing static-profile namespaces stay functional. First-bind-wins, immutability, the RLock, profile isolation, host/identity validation and the private pytest-only reset are unchanged. gitea_auth.get_profile() built an explicit whitelist dict and silently dropped `allowed_repositories`; the new integration tests caught that the scope was never enforced through the real path. Tests ----- New tests/test_issue_714_production_first_bind.py drives the real entry points in production order rather than constructing _SessionContext directly. Covers complete first bind via each entry point, same-host repo/org drift, Timesheet and other-repo/other-org rejection, unverified and unauthorized workspaces, caller values unable to establish the binding, concurrent init selecting one repository, and the PR #715 author path staying functional. Mutation-verified: disabling trusted derivation, override validation, the scope check, or the get_profile passthrough each fails these tests (9/8/4/5 failures). No security assertion was weakened, removed, skipped, or rewritten. Focused: 26 passed. Issue #714 total: 46 passed. Prior failing modules: 240 passed. Config/profile/remote modules: 130 passed. Full suite: 2739 passed, 6 skipped, 1 pre-existing warning, 161 subtests in 25.80s. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/gitea-execution-profiles.md | 42 ++ gitea-mcp.v2-contexts.example.json | 1 + gitea_auth.py | 4 + gitea_config.py | 28 ++ gitea_mcp_server.py | 102 +++- session_context_binding.py | 133 +++++- tests/test_issue_714_production_first_bind.py | 448 ++++++++++++++++++ 7 files changed, 748 insertions(+), 10 deletions(-) create mode 100644 tests/test_issue_714_production_first_bind.py diff --git a/docs/gitea-execution-profiles.md b/docs/gitea-execution-profiles.md index 970f5d4..5cf4325 100644 --- a/docs/gitea-execution-profiles.md +++ b/docs/gitea-execution-profiles.md @@ -45,6 +45,7 @@ authenticated capability set. Each profile defines the following fields: | `authenticated_username` | string | The Gitea login this profile authenticates as (verified at runtime via `gitea_whoami`, not trusted from config). | | `allowed_operations` | list | Operation categories this profile may perform. | | `forbidden_operations` | list | Operation categories this profile must never perform. | +| `allowed_repositories` | list | Optional. Canonical `owner/repository` slugs this profile may bind to. An authorization boundary, not the binding itself — see [Repository scope](#repository-scope-714). | | `token_source_name` | string | The *name* of the secret source (e.g. env var name or secret key). **Never the token value.** | | `audit_label` | string | Short label attached to audit records for actions by this profile. | | `can_approve_prs` | bool | May submit an approving PR review. | @@ -57,6 +58,47 @@ authenticated capability set. Each profile defines the following fields: name), never the token itself. Token values are never part of a profile object, never logged, never returned by a tool, and never committed. +## Repository scope (#714) + +`allowed_repositories` declares the canonical `owner/repository` slugs a profile +may operate on: + +```json +"prgs-author": { + "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"] +} +``` + +It is an **authorization boundary, not the session binding**. The binding is +derived and enforced like this: + +1. The session repository is derived from the **verified, workspace-aligned git + remote** — never from a caller-supplied `org`/`repo` argument, and never from + the `REMOTES` table (whose entries are default *targets*: `prgs` defaults to + `Timesheet`, which is not this project). +2. That workspace-derived slug is validated against `allowed_repositories`. +3. The session binds immutably to that one canonical `owner/repository`. The + organization is derived from the slug; there is no independent, + caller-controlled organization value. +4. If a profile authorizes several repositories, the verified workspace still + selects exactly one. A session never binds to the whole list and never + switches between entries. + +Fail-closed rules: + +- Activation is rejected when the workspace repository is absent from the + allowlist. +- A mutation is rejected when no verified workspace repository can be + established. +- A tool-level `org`/`repo` override that disagrees with the binding is rejected + before the mutation runs. +- A mutation request can never establish, complete, or replace the binding. + +The field is **config-only**: no environment variable can widen or forge it. It +is optional — a profile that omits it keeps the previous behaviour, so existing +static-profile namespaces stay functional until an operator provisions the +scope. Provisioning it is what activates enforcement for that profile. + ## Example profiles The following are the reference profiles. Booleans express intended capability diff --git a/gitea-mcp.v2-contexts.example.json b/gitea-mcp.v2-contexts.example.json index bd5b1ee..77c9d64 100644 --- a/gitea-mcp.v2-contexts.example.json +++ b/gitea-mcp.v2-contexts.example.json @@ -41,6 +41,7 @@ "execution_profile": "example-author", "audit_label": "example-author", "auth": { "type": "keychain", "id": "example-gitea-author-token" }, + "allowed_repositories": ["Example-Org/Example-Repo"], "allowed_operations": ["read", "branch", "commit", "push", "open_pr", "comment", "issue.comment"], "forbidden_operations": ["approve", "request_changes", "merge"] }, diff --git a/gitea_auth.py b/gitea_auth.py index a02d92a..385db61 100644 --- a/gitea_auth.py +++ b/gitea_auth.py @@ -570,6 +570,10 @@ def get_profile(): "profile_name": name, "allowed_operations": ops, "forbidden_operations": forbidden, + # #714 repository authorization boundary. Config-only on purpose: an + # environment variable must never widen or forge the set of + # repositories a session may bind to. + "allowed_repositories": _json_list("allowed_repositories"), "audit_label": audit_label, "token_source_name": token_source, "auth_source_type": auth_type, diff --git a/gitea_config.py b/gitea_config.py index ffbf7ea..3e0adef 100644 --- a/gitea_config.py +++ b/gitea_config.py @@ -475,6 +475,33 @@ def _require_enabled(kind, name, obj): return enabled +_REPO_SCOPE_RE = re.compile(r"^[^/\s]+/[^/\s]+$") + + +def _validate_allowed_repositories(name, raw): + """Validate the optional per-profile repository authorization scope (#714). + + ``allowed_repositories`` is an authorization boundary of canonical + ``owner/repository`` slugs. It is not the session binding: the verified + workspace repository selects exactly one entry at bind time. Absent means + "no repository scope configured" and is allowed, so existing profiles keep + working until an operator provisions the field. + """ + if raw is None: + return + if not isinstance(raw, list): + raise ConfigError( + f"profile '{name}' allowed_repositories must be a list of " + "'owner/repository' strings" + ) + for entry in raw: + if not isinstance(entry, str) or not _REPO_SCOPE_RE.match(entry.strip()): + raise ConfigError( + f"profile '{name}' allowed_repositories entry {entry!r} is not " + "a canonical 'owner/repository' slug" + ) + + def _reject_inline_secrets(kind, name, obj): for key in _INLINE_SECRET_KEYS: if key in obj: @@ -549,6 +576,7 @@ def _load_v2_contexts(data, path): forbidden = raw.get("forbidden_operations") or [] 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")) 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. diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 4fa3e42..823b52b 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1121,25 +1121,71 @@ import issue_claim_heartbeat # noqa: E402 import session_context_binding as session_ctx # noqa: E402 # #714 immutable session context +def _workspace_repository_slug(remote: str | None) -> str | None: + """Trusted ``owner/repository`` derived from the workspace git remote (#714). + + This is the only source the session binding accepts for repository identity: + it comes from the verified, workspace-aligned checkout rather than from any + caller-supplied tool argument. ``REMOTES`` is deliberately not consulted — + its ``org``/``repo`` entries are default *targets*, not an authorization + scope (#530). + """ + if not remote: + return None + parsed = remote_repo_guard.parse_org_repo_from_remote_url( + _local_git_remote_url(remote) + ) + if not parsed: + return None + return session_ctx.format_repository_slug(parsed[0], parsed[1]) + + +def _trusted_session_repository(profile: dict, remote: str | None) -> dict: + """Resolve and authorize the session repository from the verified workspace. + + Returns ``org`` / ``repository`` / ``reasons``. A profile's + ``allowed_repositories`` is an authorization boundary: the verified + workspace selects exactly one entry from it, and the session never binds to + the list itself nor switches between entries. + """ + slug = _workspace_repository_slug(remote) + scope = session_ctx.assess_repository_scope( + workspace_slug=slug, + allowed=session_ctx.declared_allowed_repositories(profile), + profile_name=profile.get("profile_name"), + ) + if scope.get("block"): + return {"org": None, "repository": None, "reasons": list(scope["reasons"])} + parts = session_ctx.parse_repository_slug(slug) if slug else None + if not parts: + return {"org": None, "repository": None, "reasons": []} + return {"org": parts[0], "repository": parts[1], "reasons": []} + + def _seed_session_context( *, profile: dict, remote: str | None, host: str | None, identity: str | None, - repository: str | None = None, - org: str | None = None, source: str = "seed", ) -> dict: - """Seed immutable session context once for the current process (#714).""" + """Seed immutable session context once for the current process (#714). + + Repository and organization are always derived from the verified workspace, + never from caller arguments, so every first-bind entry point (whoami, + runtime context, capability preflight, activation, mutation gate) pins the + same complete context. + """ expected = (profile.get("username") or "").strip() or None + trusted = _trusted_session_repository(profile, remote) return session_ctx.seed_session_context_if_unbound( profile_name=profile.get("profile_name") or "", remote=remote, host=host, identity=identity, - repository=repository, - org=org, + repository=trusted["repository"], + org=trusted["org"], role_kind=_profile_role_kind(profile), expected_username=expected, source=source, @@ -7621,8 +7667,12 @@ def _session_context_mutation_block( expected = (profile.get("username") or "").strip() or None remote_config = REMOTES.get(remote or "", {}) if remote else {} h = host or remote_config.get("host") - resolved_org = org or remote_config.get("org") - resolved_repo = repo or remote_config.get("repo") + # #714: repository identity comes from the verified workspace only. The + # caller-supplied org/repo are a *request target* to be checked against the + # binding — they must never establish, complete, or replace it. + trusted = _trusted_session_repository(profile, remote) + resolved_org = trusted["org"] + resolved_repo = trusted["repository"] identity = username # Legacy env-only profiles do not declare an expected identity. Their # first mutation still pins remote/host/repository, while existing audit @@ -7650,14 +7700,15 @@ def _session_context_mutation_block( ) reasons.extend(id_gate.get("reasons") or []) + # Repository scope denial (unauthorized workspace) fails closed here. + reasons.extend(trusted.get("reasons") or []) + # Seed if unbound so subsequent tools share one context; then assess. _seed_session_context( profile=profile, remote=remote, host=h, identity=identity, - repository=resolved_repo, - org=resolved_org, source="mutation-gate-seed", ) ctx_gate = session_ctx.assess_session_context( @@ -7669,8 +7720,21 @@ def _session_context_mutation_block( org=resolved_org, expected_username=expected, require_bound=True, + require_complete=bool( + session_ctx.declared_allowed_repositories(profile) + ), ) reasons.extend(ctx_gate.get("reasons") or []) + + # The request may only be checked against the immutable binding (#714). + bound_ctx = session_ctx.get_session_context() or {} + override_gate = session_ctx.assess_repository_override( + requested_org=org, + requested_repo=repo, + bound_org=bound_ctx.get("org"), + bound_repo=bound_ctx.get("repository"), + ) + reasons.extend(override_gate.get("reasons") or []) # De-dupe while preserving order seen: set[str] = set() uniq: list[str] = [] @@ -10835,11 +10899,31 @@ def gitea_activate_profile( os.environ[SESSION_PROFILE_LOCK_ENV] = after_profile # 4.6 #714: re-bind immutable session context after explicit activation. + # Activation is the sanctioned logical-session boundary, so it establishes + # the complete binding — including the workspace-verified repository. An + # unauthorized workspace fails closed here rather than at first mutation. + activated_trusted = _trusted_session_repository(after_profile_data, remote) + if activated_trusted.get("reasons"): + return { + "success": False, + "performed": False, + "reasons": list(activated_trusted["reasons"]), + "blocker_kind": "repository_scope", + "exact_next_action": ( + "BLOCKED + DIAGNOSE: the verified workspace repository is not " + "authorized by this profile's allowed_repositories. Run from a " + "workspace whose git remote matches an authorized " + "owner/repository, or have the operator provision the profile " + "scope. Do not pass org/repo to bypass this gate." + ), + } session_ctx.bind_session_context( profile_name=after_profile or profile_name, remote=remote, host=h, identity=after_identity, + repository=activated_trusted["repository"], + org=activated_trusted["org"], role_kind=_profile_role_kind(after_profile_data), expected_username=expected_username, source="gitea_activate_profile", diff --git a/session_context_binding.py b/session_context_binding.py index 57f2f1c..2336449 100644 --- a/session_context_binding.py +++ b/session_context_binding.py @@ -11,6 +11,7 @@ substitution is forbidden. from __future__ import annotations import os +import re import threading import urllib.parse from dataclasses import dataclass @@ -232,12 +233,16 @@ def assess_session_context( org: str | None = None, expected_username: str | None = None, require_bound: bool = False, + require_complete: bool = False, ) -> dict[str, Any]: """Compare live values against the bound session context. Returns ``proven`` / ``block`` / ``reasons``. When unbound and ``require_bound`` is false, does not block (caller may seed). When - unbound and ``require_bound`` is true, fails closed. + unbound and ``require_bound`` is true, fails closed. ``require_complete`` + additionally fails closed when the binding carries no verified + repository/organization identity, so a mutation can never run against an + unknown repository (#714). """ reasons: list[str] = [] with _SESSION_CONTEXT_LOCK: @@ -252,6 +257,15 @@ def assess_session_context( return _assessment(False, reasons, ctx) return _assessment(True, reasons, ctx) + if require_complete and not (ctx.get("repository") and ctx.get("org")): + reasons.append( + "session repository/organization identity is unverified " + f"(repository={ctx.get('repository')!r}, org={ctx.get('org')!r}); " + "a mutation cannot proceed without a verified workspace " + "repository (fail closed)" + ) + return _assessment(False, reasons, ctx) + live_profile = (profile_name or "").strip() or None live_remote = (remote or "").strip() or None live_host = (host or "").strip().lower() or None @@ -327,6 +341,123 @@ def assess_identity_match( } +_REPO_SLUG_RE = re.compile(r"^\s*(?P[^/\s]+)\s*/\s*(?P[^/\s]+?)(?:\.git)?\s*$") + + +def parse_repository_slug(slug: str | None) -> tuple[str, str] | None: + """Split a canonical ``owner/repository`` slug, or None when unparseable.""" + match = _REPO_SLUG_RE.match(slug or "") + if not match: + return None + return match.group("org"), match.group("repo") + + +def format_repository_slug(org: str | None, repo: str | None) -> str | None: + """``owner/repository`` from parts, or None when either side is missing.""" + left = (org or "").strip() + right = (repo or "").strip() + if not left or not right: + return None + return f"{left}/{right}" + + +def declared_allowed_repositories(profile: Mapping[str, Any] | None) -> list[str]: + """Canonical ``owner/repository`` authorization boundary declared by *profile*. + + This list is an authorization boundary, never the session binding itself: + the verified workspace selects exactly one entry (see + :func:`assess_repository_scope`). Returns ``[]`` when the profile declares + no scope — enforcement is opt-in per profile so existing static-profile + namespaces stay functional until an operator provisions the field. + """ + if not profile: + return [] + raw = profile.get("allowed_repositories") + if not isinstance(raw, (list, tuple)): + return [] + slugs: list[str] = [] + for entry in raw: + if not isinstance(entry, str): + continue + parsed = parse_repository_slug(entry) + if parsed: + slugs.append(f"{parsed[0]}/{parsed[1]}") + return slugs + + +def assess_repository_scope( + *, + workspace_slug: str | None, + allowed: list[str] | None, + profile_name: str | None = None, +) -> dict[str, Any]: + """Authorize the verified workspace repository against the profile allowlist. + + The workspace-derived slug is the only candidate: a profile that authorizes + several repositories still binds to the single verified one, and never to + the list as a whole. + """ + scope = list(allowed or []) + reasons: list[str] = [] + if not scope: + return { + "proven": True, + "block": False, + "reasons": reasons, + "scope_enforced": False, + "workspace_slug": workspace_slug, + "allowed_repositories": [], + } + name = profile_name or "(active profile)" + if not workspace_slug: + reasons.append( + f"no verified workspace repository could be established for profile " + f"'{name}'; repository scope cannot be authorized (fail closed)" + ) + elif workspace_slug.lower() not in {entry.lower() for entry in scope}: + reasons.append( + f"repository scope denial: workspace repository '{workspace_slug}' " + f"is not authorized by profile '{name}' allowed_repositories " + f"{sorted(scope)} (fail closed)" + ) + return { + "proven": not reasons, + "block": bool(reasons), + "reasons": reasons, + "scope_enforced": True, + "workspace_slug": workspace_slug, + "allowed_repositories": sorted(scope), + } + + +def assess_repository_override( + *, + requested_org: str | None, + requested_repo: str | None, + bound_org: str | None, + bound_repo: str | None, +) -> dict[str, Any]: + """Caller-supplied org/repo must agree with the immutable binding. + + A mutation request is never allowed to establish, complete, or replace the + binding — it may only be checked against it. + """ + reasons: list[str] = [] + req_org = (requested_org or "").strip() or None + req_repo = (requested_repo or "").strip() or None + if req_repo and bound_repo and req_repo.lower() != bound_repo.lower(): + reasons.append( + f"repository override denial: request targets '{req_repo}' but the " + f"session is bound to '{bound_repo}' (fail closed)" + ) + if req_org and bound_org and req_org.lower() != bound_org.lower(): + reasons.append( + f"organization override denial: request targets '{req_org}' but the " + f"session is bound to '{bound_org}' (fail closed)" + ) + return {"proven": not reasons, "block": bool(reasons), "reasons": reasons} + + def filter_profiles_for_remote( config: dict | None, remote: str | None, diff --git a/tests/test_issue_714_production_first_bind.py b/tests/test_issue_714_production_first_bind.py new file mode 100644 index 0000000..28e20d0 --- /dev/null +++ b/tests/test_issue_714_production_first_bind.py @@ -0,0 +1,448 @@ +"""#714: production first-bind pins the workspace-verified repository scope. + +These tests drive the real entry points (``gitea_whoami``, +``gitea_get_runtime_context``, ``gitea_resolve_task_capability``, +``gitea_activate_profile``) in production order. They deliberately do not +construct a ``_SessionContext`` directly: the defect they cover is that the +production first-bind path left ``repository``/``org`` unbound, so +``assess_session_context`` skipped its repository/org drift checks and a +same-host mutation against another repository was not blocked. + +The trusted repository identity comes from the workspace-aligned git remote +(``Scaled-Tech-Consulting/Gitea-Tools`` for this checkout), never from +``REMOTES`` (whose ``prgs`` default repo is ``Timesheet``) and never from a +caller-supplied ``org``/``repo`` argument. +""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import gitea_config # noqa: E402 +import gitea_mcp_server as srv # noqa: E402 +import mcp_server # noqa: E402 +import session_context_binding as session_ctx # noqa: E402 + +WORKSPACE_ORG = "Scaled-Tech-Consulting" +WORKSPACE_REPO = "Gitea-Tools" +WORKSPACE_SLUG = f"{WORKSPACE_ORG}/{WORKSPACE_REPO}" +WORKSPACE_URL = f"https://gitea.prgs.cc/{WORKSPACE_SLUG}.git" + +_BASE_AUTHOR_OPS = [ + "gitea.read", + "gitea.issue.create", + "gitea.issue.comment", + "gitea.pr.create", + "gitea.pr.comment", + "gitea.branch.push", + "gitea.repo.commit", +] + + +def _config(allowed_repositories=None): + profile = { + "enabled": True, + "context": "prgs", + "role": "author", + "username": "jcwalker3", + "base_url": "https://gitea.prgs.cc", + "auth": {"type": "env", "name": "GITEA_TOKEN_PRGS_AUTHOR"}, + "allowed_operations": list(_BASE_AUTHOR_OPS), + "forbidden_operations": [ + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.request_changes", + ], + "execution_profile": "prgs-author", + } + if allowed_repositories is not None: + profile["allowed_repositories"] = list(allowed_repositories) + return { + "version": 2, + # v2-contexts normalizes the config and keeps only rules.* — a + # top-level allow_runtime_switching would be dropped. + "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": {"prgs-author": profile}, + } + + +class _ProductionOrderBase(unittest.TestCase): + """Real entry points, pinned workspace remote, mocked network only.""" + + allowed_repositories = [WORKSPACE_SLUG] + + def setUp(self): + self._dir = tempfile.TemporaryDirectory() + self.config_path = os.path.join(self._dir.name, "profiles.json") + with open(self.config_path, "w", encoding="utf-8") as fh: + fh.write(json.dumps(_config(self.allowed_repositories))) + self._env = { + "GITEA_MCP_CONFIG": self.config_path, + "GITEA_MCP_PROFILE": "prgs-author", + "GITEA_TOKEN_PRGS_AUTHOR": "prgs-author-token", + } + gitea_config._active_profile_override = None + mcp_server._IDENTITY_CACHE.clear() + srv._MUTATION_AUTHORITY = None + # Pin the workspace git remote so the trusted source is deterministic + # and never depends on the developer's checkout layout. + self._remote_url = patch.object( + srv, "_local_git_remote_url", side_effect=self._local_remote_url + ) + self._remote_url.start() + + def tearDown(self): + self._remote_url.stop() + gitea_config._active_profile_override = None + mcp_server._IDENTITY_CACHE.clear() + srv._MUTATION_AUTHORITY = None + self._dir.cleanup() + + def _local_remote_url(self, remote_name): + return WORKSPACE_URL if remote_name == "prgs" else None + + def _api(self, method, url, header): + return { + "login": "jcwalker3", + "full_name": "Test", + "id": 1, + "email": "t@example.com", + } + + def _live(self): + return patch("gitea_mcp_server.api_request", side_effect=self._api) + + +class TestProductionFirstBindPinsRepository(_ProductionOrderBase): + def test_whoami_first_bind_is_complete(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_whoami(remote="prgs") + ctx = session_ctx.get_session_context() + self.assertIsNotNone(ctx) + self.assertEqual(ctx["org"], WORKSPACE_ORG) + self.assertEqual(ctx["repository"], WORKSPACE_REPO) + self.assertEqual(ctx["remote"], "prgs") + self.assertEqual(ctx["host"], "gitea.prgs.cc") + + def test_runtime_context_first_bind_is_complete(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_get_runtime_context(remote="prgs") + ctx = session_ctx.get_session_context() + self.assertEqual(ctx["org"], WORKSPACE_ORG) + self.assertEqual(ctx["repository"], WORKSPACE_REPO) + + def test_capability_preflight_first_bind_is_complete(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_resolve_task_capability(task="comment_issue", remote="prgs") + ctx = session_ctx.get_session_context() + self.assertEqual(ctx["org"], WORKSPACE_ORG) + self.assertEqual(ctx["repository"], WORKSPACE_REPO) + + def test_activate_profile_binds_complete_context(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_activate_profile(profile_name="prgs-author", remote="prgs") + ctx = session_ctx.get_session_context() + self.assertEqual(ctx["org"], WORKSPACE_ORG) + self.assertEqual(ctx["repository"], WORKSPACE_REPO) + + +class TestProductionOrderRepositoryDriftBlocks(_ProductionOrderBase): + def test_whoami_then_same_host_other_repository_blocks(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_whoami(remote="prgs") + blocked = srv._session_context_mutation_block( + remote="prgs", org=WORKSPACE_ORG, repo="Other-Tools" + ) + self.assertIsNotNone(blocked) + self.assertTrue( + any("Other-Tools" in r for r in blocked["reasons"]), blocked["reasons"] + ) + + def test_whoami_then_timesheet_blocks(self): + """REMOTES['prgs'].repo is Timesheet — a default target, not a scope.""" + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_whoami(remote="prgs") + blocked = srv._session_context_mutation_block( + remote="prgs", org=WORKSPACE_ORG, repo="Timesheet" + ) + self.assertIsNotNone(blocked) + self.assertTrue( + any("Timesheet" in r for r in blocked["reasons"]), blocked["reasons"] + ) + + def test_whoami_then_same_host_other_org_blocks(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_whoami(remote="prgs") + blocked = srv._session_context_mutation_block( + remote="prgs", org="Other-Org", repo=WORKSPACE_REPO + ) + self.assertIsNotNone(blocked) + self.assertTrue( + any("Other-Org" in r for r in blocked["reasons"]), blocked["reasons"] + ) + + def test_runtime_context_then_other_repository_blocks(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_get_runtime_context(remote="prgs") + blocked = srv._session_context_mutation_block( + remote="prgs", org=WORKSPACE_ORG, repo="Other-Tools" + ) + self.assertIsNotNone(blocked) + + def test_capability_preflight_then_other_repository_blocks(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_resolve_task_capability(task="comment_issue", remote="prgs") + blocked = srv._session_context_mutation_block( + remote="prgs", org=WORKSPACE_ORG, repo="Other-Tools" + ) + self.assertIsNotNone(blocked) + + def test_activate_profile_then_other_repository_blocks(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_activate_profile(profile_name="prgs-author", remote="prgs") + blocked = srv._session_context_mutation_block( + remote="prgs", org="Other-Org", repo="Other-Tools" + ) + self.assertIsNotNone(blocked) + + def test_cross_host_still_blocks(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_whoami(remote="prgs") + blocked = srv._session_context_mutation_block(remote="dadeschools") + self.assertIsNotNone(blocked) + self.assertTrue( + any("cross-host" in r for r in blocked["reasons"]), blocked["reasons"] + ) + + def test_authorized_repository_mutation_remains_allowed(self): + """Normal PR #715 author comment/push path must stay functional.""" + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_whoami(remote="prgs") + self.assertIsNone( + srv._session_context_mutation_block( + remote="prgs", org=WORKSPACE_ORG, repo=WORKSPACE_REPO + ) + ) + # A bare call with no override resolves to the same bound scope. + self.assertIsNone(srv._session_context_mutation_block(remote="prgs")) + + +class TestMutationRequestCannotEstablishBinding(_ProductionOrderBase): + def test_caller_values_cannot_establish_binding_on_first_mutation(self): + """A mutation-first session must not be pinned by request values.""" + with patch.dict(os.environ, self._env, clear=False), self._live(): + self.assertIsNone(session_ctx.get_session_context()) + srv._session_context_mutation_block( + remote="prgs", org="Attacker-Org", repo="Attacker-Repo" + ) + ctx = session_ctx.get_session_context() + self.assertIsNotNone(ctx) + # Bound to the verified workspace, never to the request values. + self.assertEqual(ctx["org"], WORKSPACE_ORG) + self.assertEqual(ctx["repository"], WORKSPACE_REPO) + + def test_mutation_first_with_attacker_values_is_blocked(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + blocked = srv._session_context_mutation_block( + remote="prgs", org="Attacker-Org", repo="Attacker-Repo" + ) + self.assertIsNotNone(blocked) + + def test_later_call_cannot_overwrite_bound_repository(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_whoami(remote="prgs") + before = session_ctx.get_session_context() + for _ in range(3): + srv._session_context_mutation_block( + remote="prgs", org="Other-Org", repo="Other-Tools" + ) + self.assertEqual(session_ctx.get_session_context(), before) + + +class TestUnverifiedWorkspaceFailsClosed(_ProductionOrderBase): + def _local_remote_url(self, remote_name): + return None # no verifiable workspace repository + + def test_mutation_blocks_when_workspace_repository_unverified(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_whoami(remote="prgs") + ctx = session_ctx.get_session_context() + self.assertIsNone(ctx["repository"]) + blocked = srv._session_context_mutation_block( + remote="prgs", org=WORKSPACE_ORG, repo=WORKSPACE_REPO + ) + self.assertIsNotNone(blocked) + self.assertTrue( + any( + "no verified workspace repository" in r or "unverified" in r + for r in blocked["reasons"] + ), + blocked["reasons"], + ) + + def test_activate_profile_rejects_unverifiable_workspace(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + res = srv.gitea_activate_profile(profile_name="prgs-author", remote="prgs") + self.assertFalse(res.get("success", True)) + self.assertEqual(res.get("blocker_kind"), "repository_scope") + + +class TestUnauthorizedWorkspaceFailsClosed(_ProductionOrderBase): + allowed_repositories = ["Scaled-Tech-Consulting/Some-Other-Project"] + + def test_workspace_absent_from_allowlist_blocks_mutation(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + blocked = srv._session_context_mutation_block(remote="prgs") + self.assertIsNotNone(blocked) + self.assertTrue( + any("not authorized" in r for r in blocked["reasons"]), + blocked["reasons"], + ) + + def test_workspace_absent_from_allowlist_blocks_activation(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + res = srv.gitea_activate_profile(profile_name="prgs-author", remote="prgs") + self.assertFalse(res.get("success", True)) + self.assertEqual(res.get("blocker_kind"), "repository_scope") + + +class TestTimesheetNotAuthorizedInThisPhase(_ProductionOrderBase): + """Cross-project use is paused: only Gitea-Tools is authorized.""" + + def test_timesheet_workspace_is_rejected(self): + def timesheet_remote(remote_name): + return "https://gitea.prgs.cc/Scaled-Tech-Consulting/Timesheet.git" + + with patch.dict(os.environ, self._env, clear=False), self._live(): + with patch.object( + srv, "_local_git_remote_url", side_effect=timesheet_remote + ): + blocked = srv._session_context_mutation_block(remote="prgs") + self.assertIsNotNone(blocked) + self.assertTrue( + any("not authorized" in r for r in blocked["reasons"]), + blocked["reasons"], + ) + + +class TestConcurrentFirstBindSelectsOneRepository(_ProductionOrderBase): + def test_concurrent_initialization_cannot_change_selected_repository(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + self.assertIsNone(session_ctx.get_session_context()) + thread_count = 8 + barrier = threading.Barrier(thread_count) + results = [] + lock = threading.Lock() + + def racer(index: int) -> None: + barrier.wait() + srv._session_context_mutation_block( + remote="prgs", org=f"Racer-Org-{index}", repo=f"Racer-Repo-{index}" + ) + with lock: + results.append(session_ctx.get_session_context()) + + threads = [ + threading.Thread(target=racer, args=(i,)) for i in range(thread_count) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + + self.assertFalse(any(t.is_alive() for t in threads)) + winner = session_ctx.get_session_context() + self.assertEqual(winner["org"], WORKSPACE_ORG) + self.assertEqual(winner["repository"], WORKSPACE_REPO) + self.assertEqual(results, [winner] * thread_count) + + +class TestRepositoryScopeUnits(unittest.TestCase): + def test_absent_scope_is_not_enforced(self): + scope = session_ctx.assess_repository_scope( + workspace_slug=WORKSPACE_SLUG, allowed=[], profile_name="legacy" + ) + self.assertTrue(scope["proven"]) + self.assertFalse(scope["scope_enforced"]) + + def test_declared_scope_authorizes_only_listed_repository(self): + allowed = session_ctx.declared_allowed_repositories( + {"allowed_repositories": [WORKSPACE_SLUG]} + ) + self.assertEqual(allowed, [WORKSPACE_SLUG]) + self.assertTrue( + session_ctx.assess_repository_scope( + workspace_slug=WORKSPACE_SLUG, allowed=allowed + )["proven"] + ) + self.assertTrue( + session_ctx.assess_repository_scope( + workspace_slug="Scaled-Tech-Consulting/Timesheet", allowed=allowed + )["block"] + ) + + def test_missing_workspace_slug_blocks_when_scope_declared(self): + scope = session_ctx.assess_repository_scope( + workspace_slug=None, allowed=[WORKSPACE_SLUG] + ) + self.assertTrue(scope["block"]) + + def test_override_must_match_binding(self): + self.assertTrue( + session_ctx.assess_repository_override( + requested_org=WORKSPACE_ORG, + requested_repo="Other", + bound_org=WORKSPACE_ORG, + bound_repo=WORKSPACE_REPO, + )["block"] + ) + self.assertTrue( + session_ctx.assess_repository_override( + requested_org="Other-Org", + requested_repo=WORKSPACE_REPO, + bound_org=WORKSPACE_ORG, + bound_repo=WORKSPACE_REPO, + )["block"] + ) + self.assertTrue( + session_ctx.assess_repository_override( + requested_org=WORKSPACE_ORG, + requested_repo=WORKSPACE_REPO, + bound_org=WORKSPACE_ORG, + bound_repo=WORKSPACE_REPO, + )["proven"] + ) + + def test_malformed_allowlist_entries_are_ignored(self): + self.assertEqual( + session_ctx.declared_allowed_repositories( + {"allowed_repositories": ["not-a-slug", 17, None, WORKSPACE_SLUG]} + ), + [WORKSPACE_SLUG], + ) + + +if __name__ == "__main__": + unittest.main()