From 5933d87647656643a67a50331c4c7b06ea751dad Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Sat, 11 Jul 2026 20:07:40 -0400 Subject: [PATCH] feat(guard): block direct stable-branch pushes from MCP workflow sessions (Closes #671) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Prevention hardening for the #670 incident (bare direct-to-master commit 2fa97c26 and the PR #654 merger stable-branch push attempt). Worker sessions must never publish stable branches directly; stable updates land only through sanctioned Gitea merge tooling or an authorized reconciler path. New pure module stable_branch_push_guard.py: - classify_push_command: detects git push equivalents including refspecs (HEAD:, +refs/heads/x:refs/heads/), --force, --delete/:, and --dry-run/-n no-op probes (dry-run still proves intent). Fetch/pull and feature-branch pushes are never flagged; sanctioned gitea_merge_pr / API merge is not a push. - assess_root_checkout_local_commit: detects control-checkout commits not on an issue feature branch (branches/ worktrees exempt). - redact_command: strips URL userinfo/token assignments before logging. - build_contamination_record + assess_contamination_gate: durable marker shape and fail-closed gate (reconciler-exempt; comment/lock stay allowed for handoff). Server wiring (gitea_mcp_server.py, mcp_session_state.py): - KIND_STABLE_BRANCH_CONTAMINATION durable session marker (per profile identity). - _enforce_stable_branch_contamination_gate wired into verify_preflight_purity so review/merge/close/completion mutations fail closed while contaminated. - gitea_record_stable_branch_push_attempt: classify + mark on detection. - gitea_audit_stable_branch_contamination: reconciler-only inspect/clear; a worker session can never self-clear. Tests (AC5): no-op dry-run push, real direct push, sanctioned Gitea merge, fetch-only, root-checkout local commit, feature-branch push allowed, gate block/reconciler-exempt, redaction. 51 new tests; full suite 2596 passed. Docs: llm-project-workflow SKILL.md — universal rule, blocker class, and a dedicated "Stable Branch Push Protection" section (worker sessions must never push stable branches directly). Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 233 +++++++++ mcp_session_state.py | 5 + skills/llm-project-workflow/SKILL.md | 50 ++ stable_branch_push_guard.py | 478 ++++++++++++++++++ ...test_stable_branch_contamination_server.py | 188 +++++++ tests/test_stable_branch_push_guard.py | 341 +++++++++++++ 6 files changed, 1295 insertions(+) create mode 100644 stable_branch_push_guard.py create mode 100644 tests/test_stable_branch_contamination_server.py create mode 100644 tests/test_stable_branch_push_guard.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index bed2098..f7f1277 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -653,6 +653,10 @@ def verify_preflight_purity( f"'{task}' (fail closed)" ) + # #671: block review/merge/close/completion mutations while the session is + # contaminated by a direct stable-branch push attempt (reconciler-exempt). + _enforce_stable_branch_contamination_gate(task, remote) + ctx = _resolve_namespace_mutation_context(worktree_path) workspace = ctx["workspace_path"] canonical_root = ctx["canonical_repo_root"] @@ -809,6 +813,80 @@ def _enforce_root_checkout_guard(worktree_path: str | None = None) -> None: raise RuntimeError(root_checkout_guard.format_root_checkout_guard_error(assessment)) +# ── stable-branch push contamination (#671) ────────────────────────────────── +# A worker session that attempts a direct stable-branch push (or a +# root-checkout local commit) is workflow-contaminated. The marker is durable +# (survives daemon process pools like the other session proofs) and keyed per +# profile identity. It fails closed on gated mutations until a reconciler +# audits and clears it. + +def _stable_contamination_profile_identity() -> str: + return mcp_session_state.current_profile_identity( + profile_name=get_profile().get("profile_name"), + ) + + +def _load_stable_contamination_marker(remote: str | None = None) -> dict | None: + return mcp_session_state.load_state( + kind=mcp_session_state.KIND_STABLE_BRANCH_CONTAMINATION, + remote=remote, + profile_identity=_stable_contamination_profile_identity(), + ) + + +def _save_stable_contamination_marker( + record: dict, + *, + remote: str | None = None, +) -> dict | None: + return mcp_session_state.save_state( + kind=mcp_session_state.KIND_STABLE_BRANCH_CONTAMINATION, + payload=record, + remote=remote, + profile_identity=_stable_contamination_profile_identity(), + ) + + +def _clear_stable_contamination_marker( + *, + remote: str | None = None, + profile_identity: str | None = None, +) -> None: + mcp_session_state.clear_state( + kind=mcp_session_state.KIND_STABLE_BRANCH_CONTAMINATION, + remote=remote, + profile_identity=profile_identity or _stable_contamination_profile_identity(), + ) + + +def _enforce_stable_branch_contamination_gate( + task: str | None, + remote: str | None = None, +) -> None: + """#671 AC4: fail closed on gated mutations while contaminated. + + Reconciler role is exempt (the sanctioned audit/clear path). Non-gated + tasks (comment_issue, lock_issue) stay allowed so a contaminated worker can + still post the durable audit comment and hand off. + """ + if _preflight_in_test_mode() and not os.environ.get( + "GITEA_TEST_FORCE_STABLE_CONTAMINATION" + ): + return + marker = _load_stable_contamination_marker(remote) + if not marker: + return + gate = stable_branch_push_guard.assess_contamination_gate( + marker, + task=task, + actual_role=_actual_profile_role(), + ) + if gate["block"]: + raise RuntimeError( + stable_branch_push_guard.format_contamination_gate_error(gate) + ) + + from mcp.server.fastmcp import FastMCP # noqa: E402 from gitea_auth import ( # noqa: E402 @@ -849,6 +927,7 @@ import merge_approval_gate # noqa: E402 import already_landed_reconcile # noqa: E402 import author_mutation_worktree # noqa: E402 import root_checkout_guard # noqa: E402 +import stable_branch_push_guard # noqa: E402 import remote_repo_guard # noqa: E402 import issue_claim_heartbeat # noqa: E402 import issue_work_duplicate_gate # noqa: E402 @@ -9273,6 +9352,160 @@ def gitea_diagnose_terminal( } +@mcp.tool() +def gitea_record_stable_branch_push_attempt( + command: str | None = None, + remote: str = "dadeschools", + ref: str | None = None, + session_id: str | None = None, + mark: bool = True, + current_branch: str | None = None, + head_sha: str | None = None, + remote_master_sha: str | None = None, + is_under_branches: bool | None = None, + ahead_count: int | None = None, +) -> dict: + """Classify a proposed command for direct stable-branch push intent (#671). + + Worker sessions must never publish stable branches (``master``/``main``/ + ``dev``/...) directly. This tool detects ``git push master`` + equivalents (refspecs, ``HEAD:master``, ``--force``, ``--dry-run`` no-op + intent, ``:master`` delete) and — separately — a root/control-checkout + local commit not carried by an issue feature branch. + + When ``mark`` is true and contamination is detected, a durable + ``stable_branch_contamination`` marker is written for the active profile + identity. Subsequent review/merge/close/completion mutations then fail + closed (via the pre-flight gate) until a reconciler audits and clears it. + The stored command summary is redacted; secrets never persist. + + Read-only when nothing is detected (or ``mark`` is false). Returns the + classification, any root-checkout assessment, and the marker state. + """ + classification = stable_branch_push_guard.classify_push_command(command) + + root_checkout = None + if any(v is not None for v in (current_branch, head_sha, remote_master_sha, + is_under_branches, ahead_count)): + root_checkout = stable_branch_push_guard.assess_root_checkout_local_commit( + current_branch=current_branch, + head_sha=head_sha, + remote_master_sha=remote_master_sha, + is_under_branches=bool(is_under_branches), + ahead_count=ahead_count, + ) + + push_contam = bool(classification.get("contamination")) + root_contam = bool(root_checkout and root_checkout.get("contamination")) + contaminated = push_contam or root_contam + + marker = None + marked = False + if contaminated and mark: + if push_contam: + reason_class = "stable_branch_push" + command_redacted = classification.get("redacted_command") + detail = "; ".join(classification.get("reasons") or []) + resolved_ref = ref or ( + (classification.get("stable_refs") or [None])[0] + ) + else: + reason_class = "root_checkout_commit" + command_redacted = None + detail = "; ".join(root_checkout.get("reasons") or []) + resolved_ref = ref or root_checkout.get("current_branch") + record = stable_branch_push_guard.build_contamination_record( + reason_class=reason_class, + command_redacted=command_redacted, + session_id=session_id, + remote=remote, + ref=resolved_ref, + role=_actual_profile_role(), + detail=detail, + ) + marker = _save_stable_contamination_marker(record, remote=remote) + marked = marker is not None + + return { + "classification": classification, + "root_checkout": root_checkout, + "contaminated": contaminated, + "marked": marked, + "marker": marker, + "profile_identity": _stable_contamination_profile_identity(), + "remediation": stable_branch_push_guard.REMEDIATION if contaminated else None, + } + + +@mcp.tool() +def gitea_audit_stable_branch_contamination( + action: str = "inspect", + remote: str = "dadeschools", + profile_identity: str | None = None, +) -> dict: + """Reconciler audit of a stable-branch contamination marker (#671). + + ``action='inspect'`` (default, read-only) loads the durable marker for the + given ``profile_identity`` (or the active session's) and reports it. + + ``action='clear'`` removes the marker so the contaminated session may + resume gated mutations. Clearing is the reconciler audit path and requires + an active reconciler profile — a worker session must never self-clear + (#671 security requirement). ``profile_identity`` targets the contaminated + worker's marker (a reconciler runs under its own identity). + """ + act = (action or "inspect").strip().lower() + target_identity = (profile_identity or "").strip() or _stable_contamination_profile_identity() + + marker = mcp_session_state.load_state( + kind=mcp_session_state.KIND_STABLE_BRANCH_CONTAMINATION, + remote=remote, + profile_identity=target_identity, + ) + + if act == "inspect": + return { + "action": "inspect", + "profile_identity": target_identity, + "contaminated": marker is not None, + "marker": marker, + "read_only": True, + } + + if act == "clear": + role = _actual_profile_role() + if role != "reconciler": + return { + "action": "clear", + "success": False, + "performed": False, + "profile_identity": target_identity, + "reasons": [ + "stable-branch contamination may only be cleared by a " + f"reconciler audit; active role is '{role}' (fail closed). " + "The contaminated worker session must not self-clear." + ], + } + _clear_stable_contamination_marker( + remote=remote, + profile_identity=target_identity, + ) + return { + "action": "clear", + "success": True, + "performed": True, + "profile_identity": target_identity, + "was_contaminated": marker is not None, + } + + return { + "action": act, + "success": False, + "performed": False, + "reasons": [f"unknown action '{act}'; use 'inspect' or 'clear'"], + } + + @mcp.tool() def gitea_validate_review_final_report( report_text: str, diff --git a/mcp_session_state.py b/mcp_session_state.py index 343b7fb..d7192ea 100644 --- a/mcp_session_state.py +++ b/mcp_session_state.py @@ -31,6 +31,11 @@ DEFAULT_TTL_HOURS = 4.0 KIND_WORKFLOW_LOAD = "review_workflow_load" KIND_DECISION_LOCK = "review_decision_lock" KIND_REVIEW_DRAFT = "review_draft" +# Durable marker set when a worker session attempts a direct stable-branch push +# or a root-checkout local commit (#671). Keyed per profile identity like the +# other session proofs; a contaminated session fails closed on gated mutations +# until a reconciler audits and clears it. +KIND_STABLE_BRANCH_CONTAMINATION = "stable_branch_contamination" diff --git a/skills/llm-project-workflow/SKILL.md b/skills/llm-project-workflow/SKILL.md index 71b4133..5720069 100644 --- a/skills/llm-project-workflow/SKILL.md +++ b/skills/llm-project-workflow/SKILL.md @@ -32,6 +32,15 @@ workflow file. mutation. - A nearby capability does not count. - Do not self-review or self-merge. +- **Never push a stable branch directly.** Worker sessions (author/reviewer/merger) + must never run `git push master` — or `main`/`dev`/other stable refs, + including refspecs (`HEAD:master`), `--force`, `--delete`, or `--dry-run` no-op + probes — and must never commit on the root/control checkout outside an issue + feature branch. Stable-branch updates land ONLY through sanctioned Gitea merge + tooling (`gitea_merge_pr`) or an explicitly authorized reconciler path. A + detected attempt marks the session workflow-contaminated (#671) and fails + closed on review/merge/close/completion until a reconciler audits and clears + it. `git fetch` / `git pull --ff-only` and feature-branch pushes stay allowed. - Do not mix modes in one run. - **BLOCKED + DIAGNOSE default rule (required):** If any required workflow step, skill, tool, capability, preflight, instruction, profile, worktree binding, or terminal/MCP operation cannot be performed or loaded (including the canonical ones listed in this skill and its loaded workflow), immediately enter `BLOCKED + DIAGNOSE`. Stop before any git or Gitea mutation. Diagnose using the standard template in [`templates/blocked-diagnose-report.md`](templates/blocked-diagnose-report.md). Attempt *only* safe non-mutating recovery. Report using the template. Do not continue, use fallbacks, or treat the missing requirement as harmless. - If the required workflow cannot be loaded, stop and produce a recovery handoff @@ -48,6 +57,11 @@ workflow file. - wrong profile or role for the requested operation - dirty or misbound worktree (root checkout or non-branches/ path) - root checkout mutation risk +- stable-branch push contamination (#671): a direct `git push master` + (or `main`/`dev`) equivalent, or a root/control-checkout commit not on an issue + feature branch, marks the session workflow-contaminated; review/merge/close/ + completion mutations then fail closed until a reconciler audits and clears it + (`gitea_audit_stable_branch_contamination action=clear`, reconciler-only) - mutation guard failure (e.g. branches-only guard) - missing required MCP tool/schema or operation - stale or inconsistent runtime state (e.g. lease vs actual, dirty state disagreement) @@ -118,6 +132,42 @@ The main project checkout is a stable control checkout on `master`, `main`, or If `cwd` is not inside `branches/`, stop before any file edit, test write, commit, merge, rebase, or cleanup. The main checkout is orchestration-only. +## Stable Branch Push Protection (#671) + +Worker sessions must **never** publish a stable branch directly. This is the +prevention hardening for the #670 incident (a bare direct-to-master commit +`2fa97c26` and a PR #654 merger `git push prgs master` attempt). + +**Forbidden for author/reviewer/merger sessions:** + +- `git push master` (and `main`, `dev`, `develop`, `development`), + including refspecs (`HEAD:master`, `+refs/heads/x:refs/heads/master`), + `--force`, `--delete` / `:master`, and `--dry-run`/`-n` no-op probes (a + dry-run still proves intent and contaminates the session). +- Local commits on the root/control checkout that are not carried by an issue + feature branch under `branches/`. + +**Allowed (never blocked):** + +- `git fetch` and `git pull --ff-only` of master into the control checkout when + authorized for sync. +- Feature-branch pushes to non-stable refs (`git push fix/issue-N-...`). +- Sanctioned merges via `gitea_merge_pr` / the Gitea API merge endpoint — the + **only** way stable branches advance. + +**What happens on a detected attempt:** the session is marked +workflow-contaminated (durable `stable_branch_contamination` marker, redacted +command summary + session id + remote + ref). While contaminated, all +review / merge / close / issue-completion mutations fail closed. `comment_issue` +and `lock_issue` remain allowed so the contaminated worker can post the durable +audit comment and hand off. Contamination **cannot be self-cleared** — only a +reconciler audit (`gitea_audit_stable_branch_contamination action=clear`) may +clear it. + +Tooling: call `gitea_record_stable_branch_push_attempt` to classify/record a +proposed push before running it; `gitea_audit_stable_branch_contamination` to +inspect or (reconciler-only) clear the marker. + ## Shell Spawn Hard-Stop Rule `exit_code: -1` with empty stdout/stderr means the shell failed to spawn — not a diff --git a/stable_branch_push_guard.py b/stable_branch_push_guard.py new file mode 100644 index 0000000..aa15b83 --- /dev/null +++ b/stable_branch_push_guard.py @@ -0,0 +1,478 @@ +"""Fail-closed guard against direct pushes to stable branches (#671). + +MCP workflow sessions must never publish to a stable branch (``master``, +``main``, ``dev``, ...) directly. Stable-branch updates land only through +sanctioned Gitea merge tooling or an explicitly authorized reconciler path. + +Incident origin (#670): a bare direct-to-master commit ``2fa97c26`` landed on +``prgs/master`` without PR/review provenance, and a PR #654 merger run +attempted ``git push prgs master`` (audited as a no-op, but the *intent* is the +hazard). Partial protections already existed (``author_proofs.PROTECTED_BRANCHES``, +``root_checkout_guard``, branch-push proofs) but did not detect shell push +equivalents, mark the session contaminated after such an attempt, or fail +closed on subsequent review/merge/close/completion mutations. + +This module is the detection + contamination-policy core. Like +``author_proofs`` and ``root_checkout_guard`` it is **pure**: callers gather +the raw facts (the proposed command line, git state, the durable contamination +marker) and pass them in, so the same logic serves prompts, MCP gates, and +tests. Nothing here performs git or network calls or reads durable state; the +server wires these helpers to ``mcp_session_state`` and ``verify_preflight_purity``. + +Design rules honoured (from the #671 acceptance criteria): + +* Detect ``git push master`` shell equivalents, including refspecs + (``HEAD:master``, ``+refs/heads/x:refs/heads/master``), ``--force``, + ``--dry-run``/no-op intent, and delete refspecs (``:master``). +* Never false-block a feature-branch push (``git push prgs fix/issue-671-...``). +* Never treat ``git fetch`` / ``git pull --ff-only`` as a push. +* Never treat sanctioned ``gitea_merge_pr`` / Gitea API merge as a push. +* Fail closed on ambiguous targets that *may* resolve to a stable branch, but + surface ambiguity as its own signal rather than silently blocking feature work. +* Contamination must not be clearable by the same worker session — only a + reconciler (audit) role may clear or bypass the gate. +""" + +from __future__ import annotations + +import re +from typing import Any, Iterable + +from author_proofs import PROTECTED_BRANCHES + +# Single source of truth for the stable branch set: the same protected set the +# author branch-identity proofs already enforce (#177), so the two guards can +# never drift apart. +STABLE_BRANCHES = PROTECTED_BRANCHES + +# Mutations that must fail closed once the session is contaminated by a direct +# stable-branch push attempt (#671 AC4: review, merge, close, completion). +# ``comment_issue`` / ``lock_issue`` / ``create_issue`` are deliberately NOT +# gated so a contaminated worker can still post the durable audit comment and +# hand off. Only a reconciler (audit) role bypasses this set. +CONTAMINATION_GATED_TASKS = frozenset({ + "create_pr", + "commit_files", + "gitea_commit_files", + "close_pr", + "close_issue", + "review_pr", + "approve_pr", + "request_changes_pr", + "submit_pr_review", + "merge_pr", + "delete_branch", + "complete_issue", +}) + +CONTAMINATION_KIND = "stable_branch_push" + +REMEDIATION = ( + "Direct stable-branch publication is forbidden for worker sessions. Stop, " + "leave the stable branch untouched, and route the change through sanctioned " + "Gitea merge tooling (gitea_merge_pr) or an explicitly authorized reconciler " + "audit. This session is workflow-contaminated until a reconciler audits it." +) + +# ── command tokenising ──────────────────────────────────────────────────────── + +# Split a compound command line into individual simple commands on shell +# separators so ``a && git push prgs master`` is analysed segment by segment. +_SEGMENT_SPLIT_RE = re.compile(r"(?:\|\||&&|\||;|\n)") + +_GIT_PUSH_RE = re.compile(r"\bgit\b[\w\s\-]*?\bpush\b", re.IGNORECASE) +_GIT_FETCH_RE = re.compile(r"\bgit\b[\w\s\-]*?\b(?:fetch|pull)\b", re.IGNORECASE) + +# Flags that take a value argument in ``git push`` (so the following token is +# consumed as the flag's value, not a refspec). +_VALUE_FLAGS = frozenset({ + "--repo", + "-o", + "--push-option", + "--receive-pack", + "--exec", +}) + +_FORCE_FLAGS = frozenset({"-f", "--force"}) +_DRY_RUN_FLAGS = frozenset({"-n", "--dry-run"}) +_DELETE_FLAGS = frozenset({"-d", "--delete"}) + + +def _clean(value: str | None) -> str: + return (value or "").strip() + + +def _strip_ref_prefix(ref: str) -> str: + ref = ref.strip() + for prefix in ("refs/heads/", "heads/"): + if ref.startswith(prefix): + return ref[len(prefix):] + return ref + + +def is_stable_ref(ref: str | None) -> bool: + """True when ``ref`` names a configured stable branch.""" + name = _strip_ref_prefix(_clean(ref)) + return bool(name) and name in STABLE_BRANCHES + + +# ── redaction ───────────────────────────────────────────────────────────────── + +_URL_USERINFO_RE = re.compile(r"(https?://)[^/\s:@]+(?::[^/\s@]+)?@", re.IGNORECASE) +_SECRET_ASSIGN_RE = re.compile( + r"\b((?:GITEA_)?(?:TOKEN|PASSWORD|PASS|PAT|SECRET|API_KEY|AUTH))\s*=\s*\S+", + re.IGNORECASE, +) +_BEARER_RE = re.compile(r"\b(Bearer|token)\s+[A-Za-z0-9._\-]{8,}", re.IGNORECASE) + + +def redact_command(command: str | None) -> str: + """Strip credentials/URLs from a command line before logging it (#671). + + Redacts URL userinfo (``https://user:tok@host`` → ``https://***@host``), + ``TOKEN=...`` style assignments, and bearer/token headers. Leaves the + structural parts (``git push master``) intact for audit value. + """ + text = _clean(command) + if not text: + return "" + text = _URL_USERINFO_RE.sub(r"\1***@", text) + text = _SECRET_ASSIGN_RE.sub(lambda m: f"{m.group(1)}=***", text) + text = _BEARER_RE.sub(lambda m: f"{m.group(1)} ***", text) + return text + + +# ── push classification ─────────────────────────────────────────────────────── + +def _analyse_push_segment(segment: str) -> dict[str, Any]: + """Classify one ``git push ...`` command segment.""" + tokens = segment.split() + # Drop everything up to and including the ``push`` verb. + try: + push_idx = next( + i for i, tok in enumerate(tokens) + if tok.lower() == "push" + ) + except StopIteration: + push_idx = -1 + args = tokens[push_idx + 1:] if push_idx >= 0 else [] + + is_force = False + is_dry_run = False + is_delete = False + positionals: list[str] = [] + + skip_next = False + for tok in args: + if skip_next: + skip_next = False + continue + if tok in _FORCE_FLAGS or tok.startswith("--force-with-lease") or tok.startswith("--force-if-includes"): + is_force = True + continue + if tok in _DRY_RUN_FLAGS: + is_dry_run = True + continue + if tok in _DELETE_FLAGS: + is_delete = True + continue + if tok in _VALUE_FLAGS: + skip_next = True + continue + if tok.startswith("--") and "=" in tok: + # e.g. --repo=... : self-contained, not a refspec. + continue + if tok.startswith("-"): + # Unknown/combined short flag; ignore for ref detection. + continue + positionals.append(tok) + + # First positional after push is the remote (when any positional exists); + # the rest are refspecs / branch names. + remote = positionals[0] if positionals else None + refspecs = positionals[1:] + + stable_refs: list[str] = [] + force_to_stable = False + delete_stable = False + for spec in refspecs: + force_spec = spec.startswith("+") + body = spec[1:] if force_spec else spec + if ":" in body: + src, _, dst = body.partition(":") + dest = dst + if src == "" and dst: + # ``:master`` — delete refspec. + is_delete = True + else: + dest = body + if is_stable_ref(dest): + stable_refs.append(_strip_ref_prefix(dest)) + if force_spec or is_force: + force_to_stable = True + if is_delete: + delete_stable = True + + targets_stable = bool(stable_refs) + # Ambiguous: ``git push `` (or bare ``git push``) with no refspec — + # resolves to the current branch's upstream, which we cannot see from the + # command alone. Flag it, but do not assert stable (would false-block + # feature-branch pushes). + ambiguous = not refspecs + + return { + "is_git_push": True, + "remote": remote, + "refspecs": refspecs, + "targets_stable": targets_stable, + "stable_refs": stable_refs, + "is_force": is_force or force_to_stable, + "is_dry_run": is_dry_run, + "is_delete": is_delete or delete_stable, + "ambiguous_target": ambiguous, + } + + +def classify_push_command(command: str | None) -> dict[str, Any]: + """Classify a shell command line for direct stable-branch push intent (#671). + + Returns a dict describing whether the command is a ``git push`` targeting a + stable branch. Fetch/pull are never pushes. A sanctioned Gitea merge (which + is not a ``git push`` at all) classifies as non-push. Dry-run and no-op + pushes still count as *intent* and are reported as contamination + (``proves_intent``) so a ``--dry-run`` cannot be used to probe the gate. + """ + text = _clean(command) + result: dict[str, Any] = { + "command": text, + "redacted_command": redact_command(text), + "is_git_push": False, + "is_fetch_or_pull": False, + "targets_stable": False, + "stable_refs": [], + "is_force": False, + "is_dry_run": False, + "is_delete": False, + "ambiguous_target": False, + "contamination": False, + "proves_intent": False, + "reasons": [], + } + if not text: + return result + + stable_refs: list[str] = [] + saw_push = False + for segment in _SEGMENT_SPLIT_RE.split(text): + seg = segment.strip() + if not seg: + continue + if _GIT_FETCH_RE.search(seg) and not _GIT_PUSH_RE.search(seg): + result["is_fetch_or_pull"] = True + continue + if not _GIT_PUSH_RE.search(seg): + continue + saw_push = True + analysis = _analyse_push_segment(seg) + result["is_git_push"] = True + result["is_force"] = result["is_force"] or analysis["is_force"] + result["is_dry_run"] = result["is_dry_run"] or analysis["is_dry_run"] + result["is_delete"] = result["is_delete"] or analysis["is_delete"] + result["ambiguous_target"] = result["ambiguous_target"] or analysis["ambiguous_target"] + stable_refs.extend(analysis["stable_refs"]) + + result["stable_refs"] = sorted(set(stable_refs)) + result["targets_stable"] = bool(stable_refs) + + reasons: list[str] = [] + if result["targets_stable"]: + refs = ", ".join(result["stable_refs"]) + verb = "delete" if result["is_delete"] else ("force-push" if result["is_force"] else "push") + qualifier = " (dry-run/no-op still proves intent)" if result["is_dry_run"] else "" + reasons.append( + f"direct stable-branch {verb} detected targeting: {refs}{qualifier}; " + "worker sessions must never publish stable branches directly" + ) + result["contamination"] = True + result["proves_intent"] = True + elif saw_push and result["ambiguous_target"]: + # Bare ``git push`` with no refspec: ambiguous. Report, but do not + # contaminate on the command alone — the root-checkout / branch-state + # detector resolves whether the current branch is stable. + reasons.append( + "ambiguous 'git push' with no refspec: resolve the current branch " + "before pushing; if it is a stable branch this is forbidden" + ) + + result["reasons"] = reasons + return result + + +def detect_stable_push(commands: str | Iterable[str] | None) -> dict[str, Any]: + """Classify one command or an iterable of commands; contamination if any hit.""" + if commands is None: + return classify_push_command(None) + if isinstance(commands, str): + return classify_push_command(commands) + worst: dict[str, Any] | None = None + for cmd in commands: + res = classify_push_command(cmd) + if res["contamination"]: + return res + if worst is None or (res["is_git_push"] and not worst["is_git_push"]): + worst = res + return worst or classify_push_command(None) + + +# ── root/control checkout local-commit detection ────────────────────────────── + +def assess_root_checkout_local_commit( + *, + current_branch: str | None, + head_sha: str | None, + remote_master_sha: str | None, + is_under_branches: bool, + ahead_count: int | None = None, +) -> dict[str, Any]: + """Detect a local commit on the root/control checkout not on a feature branch. + + #671 AC2. An isolated ``branches/...`` worktree is exempt (that is where + feature work belongs). On the control checkout, a commit is illegitimate + when the checkout sits on a stable branch (or detached) and its HEAD has + advanced past the tracking ``prgs/master`` — i.e. a commit was made that is + not carried by an issue feature branch/PR. + + Positive evidence only: missing state reports ``unknown`` rather than + asserting contamination, but an unreadable *branch* on the control checkout + (detached HEAD) with an advanced HEAD is still flagged. + """ + branch = _clean(current_branch) + head = _clean(head_sha).lower() + master = _clean(remote_master_sha).lower() + + if is_under_branches: + return { + "contamination": False, + "unknown": False, + "reasons": [], + "detail": "isolated branches/ worktree — feature commits are expected here", + } + + reasons: list[str] = [] + unknown = False + + on_stable = (not branch) or (branch in STABLE_BRANCHES) + if not on_stable: + # Control checkout is on some non-stable branch: out of scope for this + # detector (root_checkout_guard #475 handles that contamination class). + return { + "contamination": False, + "unknown": False, + "reasons": [], + "detail": f"control checkout on non-stable branch '{branch}' — not this detector's class", + } + + advanced = False + if ahead_count is not None and ahead_count > 0: + advanced = True + if head and master and head != master: + advanced = True + if (not head or not master) and ahead_count is None: + unknown = True + + if advanced: + where = f"branch '{branch}'" if branch else "detached HEAD" + reasons.append( + f"control checkout ({where}) has local commits not on an issue " + "feature branch (HEAD advanced past prgs/master); worker sessions " + "must commit only on issue branches under branches/" + ) + + return { + "contamination": bool(reasons), + "unknown": unknown and not reasons, + "reasons": reasons, + "current_branch": branch or None, + "head_sha": head or None, + "remote_master_sha": master or None, + } + + +# ── contamination record + gate ─────────────────────────────────────────────── + +def build_contamination_record( + *, + reason_class: str, + command_redacted: str | None = None, + session_id: str | None = None, + remote: str | None = None, + ref: str | None = None, + role: str | None = None, + detail: str | None = None, +) -> dict[str, Any]: + """Build the durable contamination marker payload (redacted, audit-safe). + + ``reason_class`` is one of ``stable_branch_push`` / ``root_checkout_commit``. + The command is stored already-redacted; callers pass the raw line through + :func:`redact_command` (or this builder redacts a raw ``command`` for them). + """ + return { + "kind": CONTAMINATION_KIND, + "reason_class": _clean(reason_class) or "stable_branch_push", + "command_summary": redact_command(command_redacted), + "session_id": _clean(session_id) or None, + "remote": _clean(remote) or None, + "ref": _clean(ref) or None, + "role": _clean(role) or None, + "detail": _clean(detail) or None, + "cleared_by_reconciler": False, + } + + +def assess_contamination_gate( + marker: dict[str, Any] | None, + *, + task: str | None, + actual_role: str | None, +) -> dict[str, Any]: + """Fail closed on gated mutations while a contamination marker is live (#671 AC4). + + * No marker → allowed. + * Reconciler (audit) role → allowed (the sanctioned path to inspect/clear). + * Marker present + ``task`` in :data:`CONTAMINATION_GATED_TASKS` → blocked. + * Marker present + non-gated task (e.g. ``comment_issue``) → allowed, so the + worker can still post the durable audit/handoff comment. + """ + if not marker or marker.get("cleared_by_reconciler"): + return {"block": False, "reasons": [], "task": task} + + role = _clean(actual_role).lower() + if role == "reconciler": + return { + "block": False, + "reasons": [], + "task": task, + "detail": "reconciler audit path is exempt from the contamination gate", + } + + task_name = _clean(task) + if task_name and task_name in CONTAMINATION_GATED_TASKS: + summary = marker.get("command_summary") or marker.get("detail") or "(no summary)" + reason_class = marker.get("reason_class") or "stable_branch_push" + return { + "block": True, + "reasons": [ + f"session is workflow-contaminated ({reason_class}): {summary}. " + f"'{task_name}' is blocked until a reconciler audits and clears " + "the contamination. " + REMEDIATION + ], + "task": task_name, + } + + return {"block": False, "reasons": [], "task": task_name or None} + + +def format_contamination_gate_error(gate: dict[str, Any]) -> str: + """Single RuntimeError message for MCP mutation gates.""" + reasons = "; ".join(gate.get("reasons") or ["session workflow-contaminated"]) + return f"Stable-branch contamination gate (#671): {reasons}" diff --git a/tests/test_stable_branch_contamination_server.py b/tests/test_stable_branch_contamination_server.py new file mode 100644 index 0000000..ea9e65c --- /dev/null +++ b/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") diff --git a/tests/test_stable_branch_push_guard.py b/tests/test_stable_branch_push_guard.py new file mode 100644 index 0000000..79cba0a --- /dev/null +++ b/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 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:tok@gitea.example/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 -- 2.43.7