diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 70835ab..fd5db07 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -21155,6 +21155,14 @@ def gitea_adopt_workflow_lease( holds the required role (independent consume without sharing the controller session). Expired leases may be reclaimed; provenance records adopted_from/by. Terminal (abandoned/released) leases cannot be adopted. + + #843 F1: the adopter role is derived authoritatively from the active + authenticated profile — never from caller input. A supplied ``role`` that + does not exactly match the profile-derived role is rejected (no silent + accept or reinterpretation), and handoff provenance ``required_profile`` / + ``required_namespace`` restrictions are validated against the same + authoritative caller context. Caller-supplied role/profile/namespace can + never grant authority. """ read_block = _profile_operation_gate("gitea.read") if read_block: @@ -21163,25 +21171,64 @@ def gitea_adopt_workflow_lease( "reasons": read_block, "permission_report": _permission_block_report("gitea.read"), } + profile = get_profile() + profile_name = (profile.get("profile_name") or "").strip() or "session" + active_role = (_profile_role_kind(profile) or "").strip().lower() + if not active_role: + return { + "success": False, + "outcome": "blocked", + "mutation_performed": False, + "reasons": [ + "active profile role could not be derived authoritatively; " + "refusing lease adoption (fail closed, #843)" + ], + "lease_id": lease_id, + "authoritative_source": "control_plane_db", + "file_lock_only": False, + "comment_lease_only": False, + } + if role is not None and str(role).strip(): + supplied_role = str(role).strip().lower() + if supplied_role != active_role: + return { + "success": False, + "outcome": "blocked", + "mutation_performed": False, + "profile_role_kind": active_role, + "supplied_role": supplied_role, + "reasons": [ + f"caller-supplied role '{supplied_role}' does not match " + f"the authenticated profile-derived role '{active_role}'; " + "caller-supplied role/profile/namespace can never grant " + "authority (fail closed, #843)" + ], + "lease_id": lease_id, + "authoritative_source": "control_plane_db", + "file_lock_only": False, + "comment_lease_only": False, + } db, errs = _control_plane_db_or_error() if db is None: return {"success": False, "reasons": errs} - profile = get_profile() - profile_name = (profile.get("profile_name") or "").strip() or "session" - active_role = _profile_role_kind(profile) or "author" sid = (session_id or "").strip() or ( f"{profile_name}-{os.getpid()}-{uuid.uuid4().hex[:8]}" ) + adopter_namespace = allocator_service.DEFAULT_ROLE_NAMESPACES.get( + active_role, f"gitea-{active_role}" + ) try: return lease_lifecycle.adopt_lease( db, lease_id=lease_id, adopter_session_id=sid, - role=(role or active_role).strip() or "author", + role=active_role, worktree_path=worktree_path, expected_head_sha=expected_head_sha, owner_pid=os.getpid(), operator_authorized=bool(operator_authorized), + adopter_profile_name=profile_name, + adopter_namespace=adopter_namespace, ) except (lease_lifecycle.LeaseLifecycleError, control_plane_db.ControlPlaneError) as exc: return { diff --git a/lease_lifecycle.py b/lease_lifecycle.py index e081070..c42be14 100644 --- a/lease_lifecycle.py +++ b/lease_lifecycle.py @@ -564,8 +564,17 @@ def adopt_lease( expected_head_sha: str | None = None, owner_pid: int | None = None, operator_authorized: bool = False, + adopter_profile_name: str | None = None, + adopter_namespace: str | None = None, ) -> dict[str, Any]: - """Sanctioned adopt path with provenance; never silent foreign steal.""" + """Sanctioned adopt path with provenance; never silent foreign steal. + + #843 F1: for a pending cross-role handoff, ``role`` must be the + authoritative profile-derived role supplied by the MCP boundary — never + caller-asserted authority. When the handoff provenance declares + ``required_profile`` / ``required_namespace`` and the caller context is + provided, both are validated exactly; a mismatch fails closed. + """ state = db.get_lease_workflow_state(lease_id) if not state: raise LeaseLifecycleError( @@ -607,6 +616,30 @@ def adopt_lease( f"required={required} adopter={adopter_role or 'none'} " "(fail closed)" ) + # #843 F1: provenance profile/namespace restrictions are validated + # against the authoritative caller context when declared. Caller + # input can never widen authority; a mismatch fails closed. + handoff_prov = handoff.get("provenance") or {} + required_profile = str( + handoff_prov.get("required_profile") or "" + ).strip() + if required_profile and adopter_profile_name is not None: + if str(adopter_profile_name).strip() != required_profile: + raise LeaseLifecycleError( + f"wrong profile for cross-role handoff consume of " + f"{lease_id}: required_profile={required_profile} " + f"adopter_profile={adopter_profile_name} (fail closed)" + ) + required_namespace = str( + handoff_prov.get("required_namespace") or "" + ).strip() + if required_namespace and adopter_namespace is not None: + if str(adopter_namespace).strip() != required_namespace: + raise LeaseLifecycleError( + f"wrong namespace for cross-role handoff consume of " + f"{lease_id}: required_namespace={required_namespace} " + f"adopter_namespace={adopter_namespace} (fail closed)" + ) reason = "cross-role-handoff-consume" elif freshness["freshness"] == "active" and not same_owner: raise LeaseLifecycleError( diff --git a/tests/test_issue_843_cross_role_handoff.py b/tests/test_issue_843_cross_role_handoff.py index d616f13..4fe0028 100644 --- a/tests/test_issue_843_cross_role_handoff.py +++ b/tests/test_issue_843_cross_role_handoff.py @@ -445,5 +445,238 @@ class CrossRoleHandoffTest(unittest.TestCase): self.assertEqual(state["lease"]["session_id"], "author-worker") +class MCPBoundaryAdoptRoleBindingTest(unittest.TestCase): + """#843 F1: MCP-boundary role binding for ``gitea_adopt_workflow_lease``. + + The library-level wrong-role test calls ``lease_lifecycle.adopt_lease`` + directly. These tests prove the MCP entry point derives the adopter role + authoritatively from the active authenticated profile and rejects any + caller-supplied role that disagrees, so a reviewer/merger profile cannot + consume an author handoff by passing ``role="author"``. + """ + + AUTHOR_PROFILE = { + "profile_name": "prgs-author", + "role": "author", + "allowed_operations": [ + "gitea.read", + "gitea.pr.create", + "gitea.branch.push", + ], + "forbidden_operations": [], + } + REVIEWER_PROFILE = { + "profile_name": "prgs-reviewer", + "role": "reviewer", + "allowed_operations": [ + "gitea.read", + "gitea.pr.review", + "gitea.pr.approve", + "gitea.pr.request_changes", + ], + "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"], + } + MERGER_PROFILE = { + "profile_name": "prgs-merger", + "role": "merger", + "allowed_operations": ["gitea.read", "gitea.pr.merge"], + "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"], + } + FOREIGN_AUTHOR_PROFILE = { + "profile_name": "dadeschools-author", + "role": "author", + "allowed_operations": [ + "gitea.read", + "gitea.pr.create", + "gitea.branch.push", + ], + "forbidden_operations": [], + } + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self._tmp.name, "cp.sqlite3") + self.db = ControlPlaneDB(self.db_path) + self.db.upsert_session( + session_id="ctrl-session", + role="controller", + profile="prgs-controller", + pid=99999999, + ) + self.db.upsert_session( + session_id="author-worker", + role="author", + profile="prgs-author", + pid=os.getpid(), + ) + self.wt = self._tmp.name + + def tearDown(self) -> None: + self._tmp.cleanup() + + def _ready_issue(self, number: int) -> WorkCandidate: + return WorkCandidate( + kind="issue", + number=number, + labels=("status:ready", "type:bug"), + title="handoff target", + priority=20, + ) + + def _handoff_lease(self, number: int = 843) -> str: + res = allocate_next_work( + db=self.db, + session_id="ctrl-session", + role=ROLE_CONTROLLER, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + candidates=[self._ready_issue(number)], + apply=True, + profile_name="prgs-controller", + username="controller-user", + allocation_mode=ALLOCATION_MODE_CROSS_ROLE, + ) + self.assertEqual(res["outcome"], OUTCOME_ASSIGNED) + self.assertEqual(res["required_role"], ROLE_AUTHOR) + return res["assignment"]["lease_id"] + + def _call_adopt_tool(self, profile: dict, **kwargs): + import gitea_mcp_server as mcp_server + + with ( + patch.object(mcp_server, "get_profile", return_value=profile), + patch.object( + mcp_server, + "_control_plane_db_or_error", + return_value=(self.db, []), + ), + ): + return mcp_server.gitea_adopt_workflow_lease( + remote="prgs", **kwargs + ) + + def _assert_handoff_untouched(self, lease_id: str) -> None: + state = self.db.get_lease_workflow_state(lease_id) + self.assertEqual(state["lease"]["session_id"], "ctrl-session") + self.assertIsNone(state["lease"].get("adopted_by_session_id") or None) + self.assertEqual(state["lease"]["status"], "active") + self.assertEqual(state["provenance"]["handoff_status"], "pending") + + def test_reviewer_profile_cannot_consume_author_handoff_via_role_author( + self, + ) -> None: + lid = self._handoff_lease(920) + result = self._call_adopt_tool( + self.REVIEWER_PROFILE, + lease_id=lid, + session_id="reviewer-worker", + role="author", + worktree_path=self.wt, + ) + self.assertFalse(result["success"]) + self.assertEqual(result["outcome"], "blocked") + self.assertEqual(result["profile_role_kind"], "reviewer") + self.assertEqual(result["supplied_role"], "author") + self.assertIn("does not match", result["reasons"][0]) + self._assert_handoff_untouched(lid) + + def test_merger_profile_cannot_consume_author_handoff_via_role_author( + self, + ) -> None: + lid = self._handoff_lease(921) + result = self._call_adopt_tool( + self.MERGER_PROFILE, + lease_id=lid, + session_id="merger-worker", + role="author", + worktree_path=self.wt, + ) + self.assertFalse(result["success"]) + self.assertEqual(result["outcome"], "blocked") + self.assertEqual(result["profile_role_kind"], "merger") + self._assert_handoff_untouched(lid) + + def test_reviewer_profile_rejected_without_role_argument(self) -> None: + """Even without a spoofed role, the profile-derived role binds.""" + lid = self._handoff_lease(922) + result = self._call_adopt_tool( + self.REVIEWER_PROFILE, + lease_id=lid, + session_id="reviewer-worker", + worktree_path=self.wt, + ) + self.assertFalse(result["success"]) + self.assertEqual(result["outcome"], "blocked") + self.assertIn("wrong role", result["reasons"][0].lower()) + self._assert_handoff_untouched(lid) + + def test_author_profile_mismatching_supplied_role_rejected(self) -> None: + lid = self._handoff_lease(923) + result = self._call_adopt_tool( + self.AUTHOR_PROFILE, + lease_id=lid, + session_id="author-worker", + role="reviewer", + worktree_path=self.wt, + ) + self.assertFalse(result["success"]) + self.assertEqual(result["outcome"], "blocked") + self.assertEqual(result["profile_role_kind"], "author") + self.assertEqual(result["supplied_role"], "reviewer") + self._assert_handoff_untouched(lid) + + def test_foreign_profile_name_rejected_for_author_handoff(self) -> None: + """Provenance required_profile binds even when the role matches.""" + lid = self._handoff_lease(924) + result = self._call_adopt_tool( + self.FOREIGN_AUTHOR_PROFILE, + lease_id=lid, + session_id="foreign-author-worker", + worktree_path=self.wt, + ) + self.assertFalse(result["success"]) + self.assertEqual(result["outcome"], "blocked") + self.assertIn("wrong profile", result["reasons"][0].lower()) + self._assert_handoff_untouched(lid) + + def test_author_profile_consumes_author_handoff(self) -> None: + lid = self._handoff_lease(925) + result = self._call_adopt_tool( + self.AUTHOR_PROFILE, + lease_id=lid, + session_id="author-worker", + role="author", + worktree_path=self.wt, + ) + self.assertTrue(result["success"]) + self.assertEqual(result["outcome"], "adopted_cross_role_handoff") + self.assertEqual(result["adopted_by_session_id"], "author-worker") + self.assertEqual(result["adopted_from_session_id"], "ctrl-session") + state = self.db.get_lease_workflow_state(lid) + self.assertEqual(state["lease"]["session_id"], "author-worker") + self.assertEqual( + state["lease"]["adopted_by_session_id"], "author-worker" + ) + self.assertEqual(state["assignment"]["session_id"], "author-worker") + self.assertEqual(state["provenance"]["handoff_status"], "adopted") + + def test_author_profile_consumes_author_handoff_without_role_argument( + self, + ) -> None: + lid = self._handoff_lease(926) + result = self._call_adopt_tool( + self.AUTHOR_PROFILE, + lease_id=lid, + session_id="author-worker", + worktree_path=self.wt, + ) + self.assertTrue(result["success"]) + self.assertEqual(result["outcome"], "adopted_cross_role_handoff") + state = self.db.get_lease_workflow_state(lid) + self.assertEqual(state["lease"]["session_id"], "author-worker") + self.assertEqual(state["lease"]["role"], "author") + + if __name__ == "__main__": unittest.main()