Merge pull request 'feat(guard): block direct stable-branch pushes from MCP workflow sessions (Closes #671)' (#677) from fix/issue-671-block-stable-branch-push into master

tests/test_stable_branch_contamination_server.py

@@ -0,0 +1,188 @@
+"""Server wiring for stable-branch push contamination (#671).
+
+Exercises the MCP tools and the pre-flight enforcement gate against the durable
+contamination marker (isolated per-test state dir from conftest).
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import patch
+
+import gitea_mcp_server as srv
+
+
+def _clear_marker(remote="prgs"):
+    srv._clear_stable_contamination_marker(remote=remote)
+
+
+def teardown_function():
+    _clear_marker()
+
+
+# ── record tool marks a real direct push ─────────────────────────────────────
+
+def test_record_tool_marks_direct_master_push():
+    _clear_marker()
+    res = srv.gitea_record_stable_branch_push_attempt(
+        command="git push prgs master", remote="prgs"
+    )
+    assert res["contaminated"] is True
+    assert res["marked"] is True
+    assert res["marker"] is not None
+    assert res["marker"]["reason_class"] == "stable_branch_push"
+    # loaded back through the durable store
+    loaded = srv._load_stable_contamination_marker("prgs")
+    assert loaded is not None
+    assert loaded["ref"] == "master"
+
+
+def test_record_tool_dry_run_still_marks():
+    _clear_marker()
+    res = srv.gitea_record_stable_branch_push_attempt(
+        command="git push --dry-run prgs master", remote="prgs"
+    )
+    assert res["contaminated"] is True
+    assert res["marked"] is True
+
+
+def test_record_tool_feature_branch_does_not_mark():
+    _clear_marker()
+    res = srv.gitea_record_stable_branch_push_attempt(
+        command="git push prgs fix/issue-671-block-stable-branch-push",
+        remote="prgs",
+    )
+    assert res["contaminated"] is False
+    assert res["marked"] is False
+    assert srv._load_stable_contamination_marker("prgs") is None
+
+
+def test_record_tool_fetch_only_does_not_mark():
+    _clear_marker()
+    res = srv.gitea_record_stable_branch_push_attempt(
+        command="git fetch prgs", remote="prgs"
+    )
+    assert res["contaminated"] is False
+    assert res["marked"] is False
+
+
+def test_record_tool_mark_false_is_readonly():
+    _clear_marker()
+    res = srv.gitea_record_stable_branch_push_attempt(
+        command="git push prgs master", remote="prgs", mark=False
+    )
+    assert res["contaminated"] is True
+    assert res["marked"] is False
+    assert srv._load_stable_contamination_marker("prgs") is None
+
+
+def test_record_tool_redacts_secret_in_marker():
+    _clear_marker()
+    res = srv.gitea_record_stable_branch_push_attempt(
+        command="git push https://u:supersecret@host/o/r.git master",
+        remote="prgs",
+    )
+    assert res["marked"] is True
+    assert "supersecret" not in res["marker"]["command_summary"]
+
+
+def test_record_tool_root_checkout_local_commit_marks():
+    _clear_marker()
+    res = srv.gitea_record_stable_branch_push_attempt(
+        command=None,
+        remote="prgs",
+        current_branch="master",
+        head_sha="a" * 40,
+        remote_master_sha="b" * 40,
+        is_under_branches=False,
+    )
+    assert res["contaminated"] is True
+    assert res["marked"] is True
+    assert res["marker"]["reason_class"] == "root_checkout_commit"
+
+
+# ── audit tool: inspect + reconciler-only clear ──────────────────────────────
+
+def test_audit_inspect_reports_marker():
+    _clear_marker()
+    srv.gitea_record_stable_branch_push_attempt(
+        command="git push prgs master", remote="prgs"
+    )
+    out = srv.gitea_audit_stable_branch_contamination(action="inspect", remote="prgs")
+    assert out["contaminated"] is True
+    assert out["read_only"] is True
+
+
+def test_audit_clear_refused_for_non_reconciler():
+    _clear_marker()
+    srv.gitea_record_stable_branch_push_attempt(
+        command="git push prgs master", remote="prgs"
+    )
+    with patch.object(srv, "_actual_profile_role", return_value="author"):
+        out = srv.gitea_audit_stable_branch_contamination(action="clear", remote="prgs")
+    assert out["success"] is False
+    assert out["reasons"]
+    # marker survives a refused clear
+    assert srv._load_stable_contamination_marker("prgs") is not None
+
+
+def test_audit_clear_allowed_for_reconciler():
+    _clear_marker()
+    srv.gitea_record_stable_branch_push_attempt(
+        command="git push prgs master", remote="prgs"
+    )
+    identity = srv._stable_contamination_profile_identity()
+    with patch.object(srv, "_actual_profile_role", return_value="reconciler"):
+        out = srv.gitea_audit_stable_branch_contamination(
+            action="clear", remote="prgs", profile_identity=identity
+        )
+    assert out["success"] is True
+    assert srv._load_stable_contamination_marker("prgs") is None
+
+
+# ── pre-flight enforcement gate ──────────────────────────────────────────────
+
+def _force_gate_env():
+    return patch.dict(os.environ, {"GITEA_TEST_FORCE_STABLE_CONTAMINATION": "1"})
+
+
+def test_gate_blocks_gated_mutation_when_contaminated():
+    _clear_marker()
+    srv.gitea_record_stable_branch_push_attempt(
+        command="git push prgs master", remote="prgs"
+    )
+    with _force_gate_env(), patch.object(srv, "_actual_profile_role", return_value="author"):
+        for task in ("merge_pr", "review_pr", "close_issue", "create_pr"):
+            try:
+                srv._enforce_stable_branch_contamination_gate(task, "prgs")
+                raised = False
+            except RuntimeError as exc:
+                raised = True
+                assert "#671" in str(exc)
+            assert raised, task
+
+
+def test_gate_allows_comment_for_handoff_when_contaminated():
+    _clear_marker()
+    srv.gitea_record_stable_branch_push_attempt(
+        command="git push prgs master", remote="prgs"
+    )
+    with _force_gate_env(), patch.object(srv, "_actual_profile_role", return_value="author"):
+        # must not raise — worker can still post the durable audit comment
+        srv._enforce_stable_branch_contamination_gate("comment_issue", "prgs")
+        srv._enforce_stable_branch_contamination_gate("lock_issue", "prgs")
+
+
+def test_gate_exempts_reconciler_when_contaminated():
+    _clear_marker()
+    srv.gitea_record_stable_branch_push_attempt(
+        command="git push prgs master", remote="prgs"
+    )
+    with _force_gate_env(), patch.object(srv, "_actual_profile_role", return_value="reconciler"):
+        srv._enforce_stable_branch_contamination_gate("merge_pr", "prgs")
+
+
+def test_gate_noop_when_not_contaminated():
+    _clear_marker()
+    with _force_gate_env(), patch.object(srv, "_actual_profile_role", return_value="author"):
+        srv._enforce_stable_branch_contamination_gate("merge_pr", "prgs")

tests/test_stable_branch_push_guard.py

@@ -0,0 +1,341 @@
+"""Tests for the direct stable-branch push guard (#671).
+
+Covers the acceptance criteria:
+1. detect shell ``git push <remote> master`` equivalents
+2. detect root/control-checkout local commits not on an issue branch
+3. contamination record shape
+4. fail-closed gate on review/merge/close/completion (reconciler-exempt)
+5. AC5 case matrix: no-op dry-run push, real direct push, sanctioned Gitea
+   merge, fetch-only, root-checkout local commit; plus feature-branch push
+   still allowed and sanctioned merge still allowed
+6. redaction of credentials in logged command summaries
+"""
+
+import stable_branch_push_guard as guard
+
+
+# ── AC1: detect direct stable-branch push equivalents ────────────────────────
+
+def test_plain_git_push_remote_master_is_contamination():
+    res = guard.classify_push_command("git push prgs master")
+    assert res["is_git_push"] is True
+    assert res["targets_stable"] is True
+    assert res["stable_refs"] == ["master"]
+    assert res["contamination"] is True
+    assert res["reasons"]
+
+
+def test_push_main_and_dev_detected():
+    for ref in ("main", "dev", "develop", "development"):
+        res = guard.classify_push_command(f"git push origin {ref}")
+        assert res["contamination"] is True, ref
+        assert res["stable_refs"] == [ref]
+
+
+def test_head_colon_master_refspec_detected():
+    res = guard.classify_push_command("git push prgs HEAD:master")
+    assert res["targets_stable"] is True
+    assert res["stable_refs"] == ["master"]
+    assert res["contamination"] is True
+
+
+def test_full_refspec_force_plus_detected():
+    res = guard.classify_push_command(
+        "git push prgs +refs/heads/tmp:refs/heads/master"
+    )
+    assert res["contamination"] is True
+    assert res["is_force"] is True
+    assert res["stable_refs"] == ["master"]
+
+
+def test_force_flag_to_master_detected():
+    res = guard.classify_push_command("git push --force prgs master")
+    assert res["contamination"] is True
+    assert res["is_force"] is True
+
+
+def test_delete_refspec_master_detected():
+    res = guard.classify_push_command("git push prgs :master")
+    assert res["contamination"] is True
+    assert res["is_delete"] is True
+
+
+def test_delete_flag_master_detected():
+    res = guard.classify_push_command("git push --delete prgs master")
+    assert res["contamination"] is True
+    assert res["is_delete"] is True
+
+
+def test_push_detected_inside_compound_command():
+    res = guard.classify_push_command("cd repo && git push prgs master && echo ok")
+    assert res["contamination"] is True
+    assert res["stable_refs"] == ["master"]
+
+
+# ── AC5: no-op / dry-run push still proves intent ────────────────────────────
+
+def test_dry_run_push_to_master_proves_intent():
+    res = guard.classify_push_command("git push --dry-run prgs master")
+    assert res["is_dry_run"] is True
+    assert res["contamination"] is True
+    assert res["proves_intent"] is True
+    assert "intent" in " ".join(res["reasons"]).lower()
+
+
+def test_short_dry_run_flag_n_detected():
+    res = guard.classify_push_command("git push -n prgs master")
+    assert res["is_dry_run"] is True
+    assert res["contamination"] is True
+
+
+# ── AC5 negative: feature-branch push still allowed ──────────────────────────
+
+def test_feature_branch_push_not_flagged():
+    res = guard.classify_push_command(
+        "git push prgs fix/issue-671-block-stable-branch-push"
+    )
+    assert res["is_git_push"] is True
+    assert res["targets_stable"] is False
+    assert res["contamination"] is False
+    assert res["reasons"] == []
+
+
+def test_feature_branch_head_refspec_not_flagged():
+    res = guard.classify_push_command(
+        "git push prgs HEAD:fix/issue-671-block-stable-branch-push"
+    )
+    assert res["contamination"] is False
+    assert res["targets_stable"] is False
+
+
+def test_branch_named_like_master_substring_not_flagged():
+    # 'master-notes' is not the stable 'master'.
+    res = guard.classify_push_command("git push prgs master-notes")
+    assert res["contamination"] is False
+    assert res["targets_stable"] is False
+
+
+# ── AC5 negative: fetch-only operations ──────────────────────────────────────
+
+def test_git_fetch_not_a_push():
+    res = guard.classify_push_command("git fetch prgs")
+    assert res["is_git_push"] is False
+    assert res["is_fetch_or_pull"] is True
+    assert res["contamination"] is False
+
+
+def test_git_pull_ff_only_master_not_a_push():
+    res = guard.classify_push_command("git pull --ff-only prgs master")
+    assert res["is_git_push"] is False
+    assert res["is_fetch_or_pull"] is True
+    assert res["contamination"] is False
+
+
+# ── AC5 negative: sanctioned Gitea merge is not a push ───────────────────────
+
+def test_gitea_merge_pr_tool_is_not_a_push():
+    res = guard.classify_push_command("gitea_merge_pr(pr_number=671, remote='prgs')")
+    assert res["is_git_push"] is False
+    assert res["contamination"] is False
+
+
+def test_gitea_api_merge_is_not_a_push():
+    res = guard.classify_push_command(
+        "curl -X POST https://gitea.example/api/v1/repos/o/r/pulls/671/merge"
+    )
+    assert res["is_git_push"] is False
+    assert res["contamination"] is False
+
+
+# ── ambiguous bare push: reported, not auto-contaminating ────────────────────
+
+def test_bare_push_is_ambiguous_not_contaminating():
+    res = guard.classify_push_command("git push prgs")
+    assert res["is_git_push"] is True
+    assert res["ambiguous_target"] is True
+    assert res["contamination"] is False
+    assert res["reasons"]  # surfaced as a warning
+
+
+# ── AC2: root/control-checkout local commit detection ────────────────────────
+
+def test_root_checkout_commit_ahead_of_master_flagged():
+    res = guard.assess_root_checkout_local_commit(
+        current_branch="master",
+        head_sha="a" * 40,
+        remote_master_sha="b" * 40,
+        is_under_branches=False,
+    )
+    assert res["contamination"] is True
+    assert res["reasons"]
+
+
+def test_root_checkout_clean_at_master_not_flagged():
+    sha = "c" * 40
+    res = guard.assess_root_checkout_local_commit(
+        current_branch="master",
+        head_sha=sha,
+        remote_master_sha=sha,
+        is_under_branches=False,
+    )
+    assert res["contamination"] is False
+    assert res["unknown"] is False
+
+
+def test_branches_worktree_commit_is_exempt():
+    res = guard.assess_root_checkout_local_commit(
+        current_branch="fix/issue-671-block-stable-branch-push",
+        head_sha="a" * 40,
+        remote_master_sha="b" * 40,
+        is_under_branches=True,
+    )
+    assert res["contamination"] is False
+
+
+def test_root_checkout_ahead_count_flagged():
+    res = guard.assess_root_checkout_local_commit(
+        current_branch="master",
+        head_sha="",
+        remote_master_sha="",
+        is_under_branches=False,
+        ahead_count=2,
+    )
+    assert res["contamination"] is True
+
+
+def test_root_checkout_unknown_when_state_missing():
+    res = guard.assess_root_checkout_local_commit(
+        current_branch="master",
+        head_sha="",
+        remote_master_sha="",
+        is_under_branches=False,
+    )
+    assert res["contamination"] is False
+    assert res["unknown"] is True
+
+
+def test_root_checkout_non_stable_branch_out_of_scope():
+    res = guard.assess_root_checkout_local_commit(
+        current_branch="feature/x",
+        head_sha="a" * 40,
+        remote_master_sha="b" * 40,
+        is_under_branches=False,
+    )
+    assert res["contamination"] is False
+
+
+# ── AC3: contamination record shape ──────────────────────────────────────────
+
+def test_build_contamination_record_shape_and_redaction():
+    rec = guard.build_contamination_record(
+        reason_class="stable_branch_push",
+        command_redacted="git push https://user:[email protected]/o/r.git master",
+        session_id="prgs-author-123",
+        remote="prgs",
+        ref="master",
+        role="author",
+    )
+    assert rec["kind"] == guard.CONTAMINATION_KIND
+    assert rec["reason_class"] == "stable_branch_push"
+    assert rec["remote"] == "prgs"
+    assert rec["ref"] == "master"
+    assert rec["cleared_by_reconciler"] is False
+    # secret must never survive into the durable record
+    assert "tok@" not in rec["command_summary"]
+    assert "***@" in rec["command_summary"]
+
+
+# ── AC4: fail-closed gate on gated mutations ─────────────────────────────────
+
+def _marker():
+    return guard.build_contamination_record(
+        reason_class="stable_branch_push",
+        command_redacted="git push prgs master",
+        remote="prgs",
+        ref="master",
+        role="author",
+    )
+
+
+def test_gate_blocks_gated_mutations_for_author():
+    marker = _marker()
+    for task in ("create_pr", "merge_pr", "close_issue", "review_pr",
+                 "submit_pr_review", "commit_files", "close_pr"):
+        gate = guard.assess_contamination_gate(marker, task=task, actual_role="author")
+        assert gate["block"] is True, task
+        assert gate["reasons"]
+
+
+def test_gate_allows_comment_and_lock_for_handoff():
+    marker = _marker()
+    for task in ("comment_issue", "lock_issue", "create_issue", "mark_issue"):
+        gate = guard.assess_contamination_gate(marker, task=task, actual_role="author")
+        assert gate["block"] is False, task
+
+
+def test_gate_exempts_reconciler_audit_path():
+    marker = _marker()
+    gate = guard.assess_contamination_gate(marker, task="close_pr", actual_role="reconciler")
+    assert gate["block"] is False
+
+
+def test_gate_no_marker_allows_everything():
+    gate = guard.assess_contamination_gate(None, task="merge_pr", actual_role="merger")
+    assert gate["block"] is False
+
+
+def test_gate_cleared_marker_allows_everything():
+    marker = _marker()
+    marker["cleared_by_reconciler"] = True
+    gate = guard.assess_contamination_gate(marker, task="merge_pr", actual_role="merger")
+    assert gate["block"] is False
+
+
+def test_same_worker_cannot_self_clear_by_role():
+    # A merger/reviewer/author role is still gated — only reconciler is exempt.
+    marker = _marker()
+    for role in ("author", "reviewer", "merger"):
+        gate = guard.assess_contamination_gate(marker, task="merge_pr", actual_role=role)
+        assert gate["block"] is True, role
+
+
+def test_format_gate_error_mentions_issue():
+    marker = _marker()
+    gate = guard.assess_contamination_gate(marker, task="merge_pr", actual_role="merger")
+    msg = guard.format_contamination_gate_error(gate)
+    assert "#671" in msg
+    assert "contaminat" in msg.lower()
+
+
+# ── redaction unit coverage ──────────────────────────────────────────────────
+
+def test_redact_url_userinfo():
+    out = guard.redact_command("git push https://bob:secretpat@host/o/r.git master")
+    assert "secretpat" not in out
+    assert "***@" in out
+    assert "master" in out  # structure preserved for audit
+
+
+def test_redact_token_assignment():
+    out = guard.redact_command("GITEA_TOKEN=abcdef123456 git push prgs master")
+    assert "abcdef123456" not in out
+    assert "GITEA_TOKEN=***" in out
+
+
+def test_redact_empty():
+    assert guard.redact_command(None) == ""
+    assert guard.redact_command("") == ""
+
+
+# ── detect_stable_push over iterables ────────────────────────────────────────
+
+def test_detect_stable_push_iterable_finds_first_contamination():
+    cmds = ["git status", "git push prgs master", "echo done"]
+    res = guard.detect_stable_push(cmds)
+    assert res["contamination"] is True
+
+
+def test_detect_stable_push_iterable_all_safe():
+    cmds = ["git status", "git push prgs feature/x", "git fetch prgs"]
+    res = guard.detect_stable_push(cmds)
+    assert res["contamination"] is False