Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 93 Pull Requests 15 Actions Packages Projects Releases Wiki Activity

Merge pull request 'fix(scope): wire author bootstrap scope into workflow scope guard' (#942) from fix/issue-941-scope-guard-bootstrap-wiring into master

Browse Source
This commit was merged in pull request #942.
This commit is contained in:
sysadmin
2026-07-26 06:20:00 -05:00
parent 8c1d22a658 a09c485fc0
commit aab54d4825
3 changed files with 408 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+32 -1
View File
@@ -1546,6 +1546,7 @@ def verify_preflight_purity(
task=task,
target_issue_number=target_issue_number,
require_author_lock=require_author_lock,
bootstrap_assessment=bootstrap_assessment,
)
# #604: common anti-stomp preflight after legacy + #683 enforcers.
_run_anti_stomp_preflight(
@@ -1585,6 +1586,7 @@ def verify_preflight_purity(
task=task,
target_issue_number=target_issue_number,
require_author_lock=require_author_lock,
bootstrap_assessment=bootstrap_assessment,
)
if force_anti_stomp:
_run_anti_stomp_preflight(
@@ -1652,8 +1654,15 @@ def _enforce_issue_scope_guard(
task: str | None = None,
target_issue_number: int | None = None,
require_author_lock: bool = False,
bootstrap_assessment: object = _BOOTSTRAP_UNSET,
) -> None:
"""#683: fail closed on missing/out-of-scope issue ownership for mutations."""
"""#683: fail closed on missing/out-of-scope issue ownership for mutations.
#941: the shared bootstrap assessment is threaded in so this guard judges
the author issue-worktree bootstrap on the same server-derived evidence as
the #274 branches-only and #604 anti-stomp guards. Callers that supply
none fall back to computing it here, which preserves behaviour.
"""
ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
git_state = issue_lock_worktree.read_worktree_git_state(workspace)
@@ -1703,11 +1712,32 @@ def _enforce_issue_scope_guard(
import create_issue_bootstrap as _cib
is_create_issue = _cib.is_create_issue_task(task)
# #941: consume the caller-computed bootstrap assessment when one was
# threaded in, so this guard and the #274/#604 guards judge identical
# evidence. Falling back preserves behaviour for callers that supply none.
bootstrap = (
_create_issue_bootstrap_assessment(task, worktree_path)
if bootstrap_assessment is _BOOTSTRAP_UNSET
else bootstrap_assessment
)
# #941: the author issue-worktree bootstrap is pre-ownership for the same
# reason create_issue is — it exists to break the lock<->worktree cycle, so
# no owning lock can exist yet. The exemption is granted by the canonical
# shared decision over server-derived evidence, never by a task-name list,
# and fails closed on missing, malformed, cross-scope, dirty, drifted, or
# wrongly bound evidence.
bootstrap_waives_ownership = _cib.bootstrap_permits_control_checkout(
bootstrap,
task=task,
workspace_path=workspace,
canonical_repo_root=ctx["canonical_repo_root"],
)
require_lock = bool(require_author_lock) or (
authorish
and workflow_scope_guard.production_guards_forced()
and role == "author"
and not is_create_issue
and not bootstrap_waives_ownership
)
assessment = workflow_scope_guard.assess_production_mutation_guards(
workspace_path=workspace,
@@ -1720,6 +1750,7 @@ def _enforce_issue_scope_guard(
require_author_lock=require_lock,
in_test_mode=_preflight_in_test_mode(),
mutation_task=task,
bootstrap_assessment=bootstrap,
)
workflow_scope_guard.raise_if_blocked(assessment)
tests/test_issue_941_scope_guard_bootstrap_wiring.py
+346
View File
@@ -0,0 +1,346 @@
"""Regression: author bootstrap scope reaches workflow_scope_guard (#941).
PR #926 (#892) made ``bootstrap_permits_control_checkout`` accept
``task_scope=author_issue_bootstrap`` and wired that canonical decision into
the #274 branches-only enforcer and the #604 anti-stomp preflight. A third
enforcement path was left unwired.
``workflow_scope_guard.assess_root_source_mutation`` kept its own copy of the
clean-root author decision, gated on ``create_issue_bootstrap.is_create_issue_task``
— a task-name allowlist that never contained ``bootstrap_author_issue_worktree``.
So the real call path
gitea_bootstrap_author_issue_worktree
-> verify_preflight_purity
-> _enforce_issue_scope_guard
-> workflow_scope_guard.assess_production_mutation_guards
raised ProductionGuardError(missing_issue_worktree) before
``assess_author_issue_bootstrap`` was ever consulted.
These tests drive the real enforcer, not the authorization helper in
isolation. A helper-only test cannot observe this defect: #892's own predicate
tests all passed while the live bootstrap stayed blocked.
"""
from __future__ import annotations
import os
import subprocess
import tempfile
import unittest
from unittest import mock
import author_issue_bootstrap as aib
import create_issue_bootstrap as cib
import workflow_scope_guard
BOOTSTRAP_TASK = "bootstrap_author_issue_worktree"
BOOTSTRAP_TOOL = "gitea_bootstrap_author_issue_worktree"
def _make_control_repo(tmp: str) -> tuple[str, str]:
"""Create a clean control checkout on master and return (path, head)."""
repo = os.path.join(tmp, "repo")
os.makedirs(os.path.join(repo, "branches"))
subprocess.check_call(
["git", "init", "-b", "master", repo],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
subprocess.check_call(
[
"git", "-C", repo,
"-c", "user.email=t@t", "-c", "user.name=t",
"commit", "--allow-empty", "-m", "init",
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
head = subprocess.check_output(
["git", "-C", repo, "rev-parse", "HEAD"], text=True
).strip()
return repo, head
def _assessment(
repo: str,
head: str,
*,
task: str = BOOTSTRAP_TASK,
porcelain: str = "",
remote: str | None = None,
) -> dict:
return aib.assess_author_issue_bootstrap(
workspace_path=repo,
canonical_repo_root=repo,
current_branch="master",
head_sha=head,
porcelain_status=porcelain,
remote_master_sha=head if remote is None else remote,
task=task,
)
class _ControlCheckoutHarness(unittest.TestCase):
"""Drive the real server guard against a temporary clean control checkout."""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.repo, self.head = _make_control_repo(self._tmp.name)
# #683 force-on: production guards must execute under pytest.
patcher = mock.patch.dict(
os.environ,
{workflow_scope_guard.FORCE_PRODUCTION_GUARDS_ENV: "1"},
)
patcher.start()
self.addCleanup(patcher.stop)
def _enforce(
self,
task: str,
*,
porcelain: str = "",
assessment: object = "auto",
role_kind: str = "author",
):
"""Call the real _enforce_issue_scope_guard for *task*."""
import gitea_mcp_server as srv
if assessment == "auto":
assessment = _assessment(
self.repo, self.head, task=task, porcelain=porcelain
)
ctx = {
"workspace_path": self.repo,
"canonical_repo_root": self.repo,
"workspace_role_kind": role_kind,
"workspace_binding_source": "process_project_root",
}
git_state = {
"current_branch": "master",
"head_sha": self.head,
"porcelain_status": porcelain,
}
with mock.patch.object(
srv, "_resolve_namespace_mutation_context", return_value=ctx
), mock.patch.object(
srv.issue_lock_worktree,
"read_worktree_git_state",
return_value=git_state,
), mock.patch.object(
srv,
"_session_issue_lock_snapshot",
return_value={
"locked_issue_number": None,
"lock_branch_name": None,
"worktrees_match": False,
},
), mock.patch.object(
srv, "_actual_profile_role", return_value=role_kind
), mock.patch.object(
srv, "_effective_workspace_role", return_value=role_kind
), mock.patch.object(
srv, "_create_issue_bootstrap_assessment", return_value=assessment
):
srv._enforce_issue_scope_guard(None, task=task)
class TestRealPathBootstrapReachesGuard(_ControlCheckoutHarness):
"""The defect and its fix, observed through the real enforcer."""
def test_bootstrap_task_passes_scope_guard_from_clean_control(self):
# Pre-fix this raises ProductionGuardError(missing_issue_worktree)
# because the guard consulted a task-name allowlist instead of the
# canonical authorization decision.
self._enforce(BOOTSTRAP_TASK)
def test_bootstrap_tool_alias_passes_scope_guard(self):
self._enforce(BOOTSTRAP_TOOL)
def test_guard_consults_canonical_predicate(self):
"""The guard must reach bootstrap_permits_control_checkout, not a name list."""
real = cib.bootstrap_permits_control_checkout
seen: list[str | None] = []
def _spy(assessment, *, task, workspace_path, canonical_repo_root):
seen.append(task)
return real(
assessment,
task=task,
workspace_path=workspace_path,
canonical_repo_root=canonical_repo_root,
)
with mock.patch.object(
cib, "bootstrap_permits_control_checkout", side_effect=_spy
):
self._enforce(BOOTSTRAP_TASK)
self.assertIn(
BOOTSTRAP_TASK,
seen,
"workflow_scope_guard did not consult the canonical bootstrap "
"authorization decision",
)
class TestFailClosedOnBadEvidence(_ControlCheckoutHarness):
"""Missing, malformed, or mismatched scope evidence must still block."""
def _assert_blocked(self, **kwargs):
with self.assertRaises(workflow_scope_guard.ProductionGuardError):
self._enforce(BOOTSTRAP_TASK, **kwargs)
def test_missing_assessment_fails_closed(self):
self._assert_blocked(assessment=None)
def test_malformed_assessment_fails_closed(self):
self._assert_blocked(assessment={"allowed": True})
def test_non_dict_assessment_fails_closed(self):
self._assert_blocked(assessment="allowed")
def test_wrong_task_scope_fails_closed(self):
bad = dict(_assessment(self.repo, self.head))
bad["task_scope"] = "create_issue_only"
self._assert_blocked(assessment=bad)
def test_nonempty_reasons_fail_closed(self):
bad = dict(_assessment(self.repo, self.head), reasons=["note"])
self._assert_blocked(assessment=bad)
def test_mismatched_base_tips_fail_closed(self):
bad = dict(_assessment(self.repo, self.head))
bad["remote_master_sha"] = "b" * 40
self._assert_blocked(assessment=bad)
def test_unverified_base_tips_fail_closed(self):
bad = dict(_assessment(self.repo, self.head), base_tips_verified=False)
self._assert_blocked(assessment=bad)
def test_mismatched_workspace_binding_fails_closed(self):
bad = dict(_assessment(self.repo, self.head))
bad["workspace_path"] = os.path.join(self.repo, "elsewhere")
self._assert_blocked(assessment=bad)
def test_mismatched_repo_root_binding_fails_closed(self):
bad = dict(_assessment(self.repo, self.head))
bad["canonical_repo_root"] = os.path.join(self.repo, "other-root")
self._assert_blocked(assessment=bad)
def test_blocked_assessment_fails_closed(self):
bad = dict(_assessment(self.repo, self.head), block=True, allowed=False)
self._assert_blocked(assessment=bad)
class TestOrdinaryControlCheckoutMutationStillForbidden(_ControlCheckoutHarness):
"""The waiver must not leak to ordinary author work."""
def test_ordinary_author_task_still_blocked(self):
with self.assertRaises(workflow_scope_guard.ProductionGuardError):
self._enforce("commit_files", assessment=None)
def test_lock_issue_still_blocked_from_control(self):
with self.assertRaises(workflow_scope_guard.ProductionGuardError):
self._enforce("lock_issue", assessment=None)
def test_bootstrap_assessment_cannot_license_other_task(self):
# Cross-task smuggling: valid bootstrap evidence must not waive a
# different author mutation.
good = _assessment(self.repo, self.head)
with self.assertRaises(workflow_scope_guard.ProductionGuardError):
self._enforce("commit_files", assessment=good)
def test_dirty_control_checkout_still_blocked_for_bootstrap(self):
with self.assertRaises(workflow_scope_guard.ProductionGuardError):
self._enforce(BOOTSTRAP_TASK, porcelain=" M gitea_mcp_server.py\n")
class TestCreateIssueBehaviorUnchanged(_ControlCheckoutHarness):
"""#749 create_issue keeps its own sanctioned path."""
def test_create_issue_still_allowed_from_clean_control(self):
self._enforce("create_issue", assessment=None)
def test_create_issue_tool_alias_still_allowed(self):
self._enforce("gitea_create_issue", assessment=None)
def test_create_issue_blocked_when_control_dirty(self):
with self.assertRaises(workflow_scope_guard.ProductionGuardError):
self._enforce(
"create_issue",
porcelain=" M gitea_mcp_server.py\n",
assessment=None,
)
class TestGuardUnitLevelWiring(unittest.TestCase):
"""assess_root_source_mutation itself must accept and honour the evidence."""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.repo, self.head = _make_control_repo(self._tmp.name)
patcher = mock.patch.dict(
os.environ,
{workflow_scope_guard.FORCE_PRODUCTION_GUARDS_ENV: "1"},
)
patcher.start()
self.addCleanup(patcher.stop)
def _assess(self, *, task=BOOTSTRAP_TASK, bootstrap_assessment="auto"):
if bootstrap_assessment == "auto":
bootstrap_assessment = _assessment(self.repo, self.head, task=task)
return workflow_scope_guard.assess_root_source_mutation(
workspace_path=self.repo,
canonical_repo_root=self.repo,
porcelain_status="",
current_branch="master",
role_kind="author",
mutation_task=task,
bootstrap_assessment=bootstrap_assessment,
)
def test_valid_evidence_unblocks(self):
result = self._assess()
self.assertFalse(result["block"])
self.assertIsNone(result["blocker_kind"])
def test_absent_evidence_blocks(self):
result = self._assess(bootstrap_assessment=None)
self.assertTrue(result["block"])
self.assertEqual(
result["blocker_kind"], workflow_scope_guard.BLOCKER_MISSING_WORKTREE
)
def test_reconciler_exemption_preserved(self):
result = workflow_scope_guard.assess_root_source_mutation(
workspace_path=self.repo,
canonical_repo_root=self.repo,
porcelain_status="",
current_branch="master",
role_kind="reconciler",
mutation_task=BOOTSTRAP_TASK,
)
self.assertFalse(result["block"])
def test_signature_accepts_evidence_without_it_being_required(self):
# Callers that supply no evidence keep the pre-existing behaviour.
result = workflow_scope_guard.assess_root_source_mutation(
workspace_path=self.repo,
canonical_repo_root=self.repo,
porcelain_status="",
current_branch="master",
role_kind="author",
mutation_task="create_issue",
)
self.assertFalse(result["block"])
if __name__ == "__main__":
unittest.main()
workflow_scope_guard.py
+30 -1
View File
@@ -279,6 +279,7 @@ def assess_root_source_mutation(
locked_issue_number: int | None = None,
role_kind: str | None = None,
mutation_task: str | None = None,
bootstrap_assessment: Any | None = None,
) -> dict[str, Any]:
"""Fail closed for diagnostic/source edits on the control/root checkout.
@@ -286,6 +287,13 @@ def assess_root_source_mutation(
tracked source/test files on the control checkout always block, including
temporary/diagnostic/test-only intent.
#941: ``bootstrap_author_issue_worktree`` is judged by the canonical
``create_issue_bootstrap.bootstrap_permits_control_checkout`` decision over
*bootstrap_assessment* — the same server-derived evidence the #274 and
#604 guards consume — instead of a task-name allowlist local to this
module. Evidence that is absent, malformed, wrongly scoped, or bound to
another workspace leaves the ordinary block in force.
#749: ``create_issue`` is a pure remote mutation with no local tree write.
When *mutation_task* is create_issue and the control checkout has no dirty
source/test files, the missing-worktree signal is suppressed so the
@@ -336,6 +344,19 @@ def assess_root_source_mutation(
if _cib is not None and _cib.is_create_issue_task(mutation_task):
# #749: clean-root create_issue is the sanctioned bootstrap path.
create_issue_bootstrap = True
elif _cib is not None and _cib.bootstrap_permits_control_checkout(
bootstrap_assessment,
task=mutation_task,
workspace_path=workspace,
canonical_repo_root=root,
):
# #941: the author issue-worktree bootstrap is authorized by the
# canonical shared decision over server-derived task-scope
# evidence, never by a task-name allowlist kept in this module.
# The predicate fails closed on missing, malformed, cross-scope,
# dirty, drifted, or wrongly bound evidence, so this arm cannot
# widen the waiver beyond the one sanctioned bootstrap task.
create_issue_bootstrap = True
else:
# Explicit missing-worktree signal for force-on author entrypoints.
reasons.append(
@@ -393,8 +414,15 @@ def assess_production_mutation_guards(
require_author_lock: bool = False,
in_test_mode: bool = False,
mutation_task: str | None = None,
bootstrap_assessment: Any | None = None,
) -> dict[str, Any]:
"""Compose root + scope production guards when they must be active (#683)."""
"""Compose root + scope production guards when they must be active (#683).
#941: *bootstrap_assessment* is the server-derived author-bootstrap
evidence, forwarded unchanged to :func:`assess_root_source_mutation` so
this guard reaches the same canonical decision as the #274 and #604
guards. Omitting it preserves the pre-existing behaviour.
"""
if not production_guards_active(in_test_mode=in_test_mode):
return {
"proven": True,
@@ -414,6 +442,7 @@ def assess_production_mutation_guards(
locked_issue_number=locked_issue_number,
role_kind=role_kind,
mutation_task=mutation_task,
bootstrap_assessment=bootstrap_assessment,
)
if root_assess["block"]:
return {**root_assess, "skipped": False}