"""#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()