sysadmin
dc899d23c8
Close the #714 remediation gap left at 943d402 where env-only mutation
profiles, REMOTES Timesheet defaults treated as explicit caller input, and
machine-dependent Git remotes produced false greens.
- Fail closed when mutations lack a config-backed profile with non-empty
allowed_repositories; env ops cannot invent mutation authority.
- Preserve omitted-vs-explicit org/repo provenance through the mutation
gate; omitted targets bind to the verified workspace repository.
- Prefer workspace-aligned Git remotes over historical Timesheet defaults
in MCP _resolve and CLI resolve_remote.
- Centralize deterministic config-backed mutation fixtures; migrate
mutation tests off env-only authority without weakening security
assertions.
Full suite: 2744 passed, 6 skipped.
510 lines
21 KiB
Python
510 lines
21 KiB
Python
"""#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": "[email protected]",
|
|
}
|
|
|
|
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",
|
|
org_explicit=True,
|
|
repo_explicit=True,
|
|
)
|
|
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_for_read_diagnostics(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_require_scope_blocks_empty_or_missing_allowlist(self):
|
|
scope = session_ctx.assess_repository_scope(
|
|
workspace_slug=WORKSPACE_SLUG,
|
|
allowed=[],
|
|
profile_name="legacy",
|
|
require_scope=True,
|
|
)
|
|
self.assertTrue(scope["block"])
|
|
self.assertTrue(any("allowed_repositories" in r for r in scope["reasons"]))
|
|
|
|
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_in_non_strict_mode(self):
|
|
self.assertEqual(
|
|
session_ctx.declared_allowed_repositories(
|
|
{"allowed_repositories": ["not-a-slug", 17, None, WORKSPACE_SLUG]}
|
|
),
|
|
[WORKSPACE_SLUG],
|
|
)
|
|
|
|
def test_strict_mode_rejects_malformed_allowlist_entries(self):
|
|
with self.assertRaises(ValueError):
|
|
session_ctx.declared_allowed_repositories(
|
|
{"allowed_repositories": ["not-a-slug", WORKSPACE_SLUG]},
|
|
strict=True,
|
|
)
|
|
|
|
|
|
class TestRemotesDefaultNotCallerOverride(_ProductionOrderBase):
|
|
def test_resolved_timesheet_default_does_not_block_when_bound_to_workspace(self):
|
|
"""Tools historically pass REMOTES-filled Timesheet after _resolve; that
|
|
must not be treated as an explicit override against Gitea-Tools."""
|
|
with patch.dict(os.environ, self._env, clear=False), self._live():
|
|
srv.gitea_whoami(remote="prgs")
|
|
# Simulate create_issue after resolve: org/repo are REMOTES defaults
|
|
blocked = srv._session_context_mutation_block(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Timesheet",
|
|
)
|
|
self.assertIsNone(blocked, getattr(blocked, "get", lambda *_: blocked)("reasons") if blocked else None)
|
|
|
|
def test_git_unavailable_blocks_mutation(self):
|
|
def no_remote(_name):
|
|
return None
|
|
with patch.dict(os.environ, self._env, clear=False), self._live():
|
|
with patch.object(srv, "_local_git_remote_url", side_effect=no_remote):
|
|
blocked = srv._session_context_mutation_block(remote="prgs")
|
|
self.assertIsNotNone(blocked)
|
|
self.assertTrue(
|
|
any(
|
|
"workspace" in r.lower() or "allowed_repositories" in r or "unverified" in r
|
|
for r in blocked["reasons"]
|
|
),
|
|
blocked["reasons"],
|
|
)
|
|
|
|
def test_explicit_mismatch_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="prgs",
|
|
org=WORKSPACE_ORG,
|
|
repo="Other-Tools",
|
|
)
|
|
self.assertIsNotNone(blocked)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|