adam/open-swe
1
0
Fork 0
mirror of https://github.com/Sea-Haven-Industries/open-swe.git synced 2026-10-07 16:19:09 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(workflow-guard): fail closed on ambiguous wrappers and multi/rewritten push URLs

Browse source 
Follow-up hardening from the security re-verification of the prior fix:

- H6: the parser's wrapper skip-loop skipped a wrapper's option flag but not its
  space-separated value, so `nice -n 10 git push`, `sudo -u ci git push`,
  `timeout 5 git push` (and wrappers outside the set) returned None and ran
  unguarded. Rewrite `_parse_git_tokens` to locate the `git` executable and fail
  closed on any unrecognized leading token before a git push, instead of a silent
  pass-through.
- H1: `git remote get-url --push` returns only the first of multiple pushurl
  entries while `git push` writes to ALL of them, and insteadOf/pushInsteadOf can
  rewrite the destination. Use `get-url --push --all` and fail closed unless there
  is exactly one push URL and no URL rewrite is configured.

Adds tests for valued-wrapper-option pushes, multiple push URLs, and URL rewrites.

Claude-Session: https://claude.ai/code/session_01GxSndB7VoGQyeS196eUr5E
This commit is contained in:
Adam Moussa 2026-07-01 18:59:59 -04:00
parent 7ec08d1870
commit abbb10dd00
No known key found for this signature in database
2 changed files with 161 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

72
agent/middleware/workflow_push_guard.py
View file

@ -181,38 +181,33 @@ def _parse_git_push(command: str) -> ParsedGitPush | _BlockedGitPush | None:
return _parse_git_tokens(tokens, repo_dir=None)
def _git_invocation_args(tokens: list[str]) -> list[str] | None:
"""Return the args following the `git` executable if these tokens invoke git, else None.
Recognizes bareword `git`, path invocations like `/usr/bin/git`, and simple command
wrappers (`command`/`env`/`nice`/`sudo`/...), including `env NAME=VALUE ...`. This keeps
a wrapped or path-qualified `git push` from being mistaken for a non-git command and
passed through the guard unchecked.
"""
idx = 0
while idx < len(tokens):
tok = tokens[idx]
if tok.rsplit("/", 1)[-1] == "git":
return tokens[idx + 1 :]
if tok in _GIT_WRAPPERS:
idx += 1
# Skip wrapper options and `NAME=VALUE` assignments (e.g. `env FOO=bar git ...`).
while idx < len(tokens) and (tokens[idx].startswith("-") or "=" in tokens[idx]):
idx += 1
continue
return None
return None
def _parse_git_tokens(
tokens: list[str], *, repo_dir: str | None
) -> ParsedGitPush | _BlockedGitPush | None:
git_args = _git_invocation_args(tokens)
if git_args is None:
"""Parse tokens into a supported git push, a block sentinel, or None.
Recognizes bareword `git`, path invocations (`/usr/bin/git`), and wrapper-prefixed
forms (`command`/`env`/`nice`/... git). Anything git-push-shaped that cannot be reduced
to `git [-C <dir>] push [-u] origin <refspec>` fails closed to a block rather than a
silent pass-through. Non-push git commands and non-git commands return None.
"""
git_idx = next((idx for idx, tok in enumerate(tokens) if tok.rsplit("/", 1)[-1] == "git"), None)
if git_idx is None:
return None
git_args = tokens[git_idx + 1 :]
if "push" not in git_args:
# A non-push git command (e.g. `git status`) is not our concern.
# A non-push git command (e.g. `git status`, `git pull`).
return None
# Everything before the `git` executable must be a benign wrapper prefix (a known
# wrapper, an option flag, or a NAME=VALUE assignment). An unrecognized leading token
# (a value-taking wrapper option like the `10` in `nice -n 10`, an unknown wrapper, or
# `git` used as an argument to another program) is ambiguous, so fail closed.
if any(
not (tok in _GIT_WRAPPERS or tok.startswith("-") or "=" in tok) for tok in tokens[:git_idx]
):
return _BlockedGitPush(
"unrecognized wrapper before `git push`; use `git push origin <branch>`"
)
i = 0
while i < len(git_args) and git_args[i] != "push":
if git_args[i] == "-C" and i + 1 < len(git_args):
@ -376,12 +371,31 @@ def _fetch_remote_base(backend: Any, repo_dir: str | None, remote: str, branch:
repo. Fetching the base from the push URL keeps the base and the push destination the
same authenticated repo, so a split cannot hide a workflow change. Uses `FETCH_HEAD`
(freshly fetched), never a local `refs/remotes/origin/*` ref the sandbox could rewrite.
Returns None (which makes the caller require approval) if the push destination is
ambiguous or rewritten: multiple `pushurl` entries (git pushes to ALL of them, so a
single base cannot represent the destination) or any `insteadOf`/`pushInsteadOf` URL
rewrite the sandbox could use to make the inspected URL differ from the push target.
"""
push_url_res = _run_git(backend, repo_dir, f"remote get-url --push {shlex.quote(remote)}")
push_url = _first_line(push_url_res.output) if push_url_res.ok else ""
if not push_url:
rewrites = _run_git(
backend, repo_dir, "config --get-regexp " + shlex.quote(r"url\..*\.(push)?insteadof")
)
if rewrites.ok and rewrites.output.strip():
# A URL rewrite means the inspected push URL may not be the real destination.
return None
push_url_res = _run_git(backend, repo_dir, f"remote get-url --push --all {shlex.quote(remote)}")
push_urls = (
[line.strip() for line in push_url_res.output.splitlines() if line.strip()]
if push_url_res.ok
else []
)
# `git push` sends to every configured push URL; if there is not exactly one we cannot
# represent the destination with a single base, so fail closed.
if len(push_urls) != 1:
return None
push_url = push_urls[0]
fetch = _run_git(backend, repo_dir, f"fetch {shlex.quote(push_url)} {shlex.quote(branch)}")
if fetch.ok:
return "FETCH_HEAD"

122
tests/test_workflow_push_guard.py
View file

@ -29,6 +29,8 @@ class _Backend:
remote_branches: set[str] | None = None,
default_branch: str = "main",
push_url: str = "https://github.com/langchain-ai/open-swe.git",
push_urls: list[str] | None = None,
url_rewrites: str = "",
ls_tree_head_ok: bool = True,
) -> None:
self.workflow_files = workflow_files
@ -38,6 +40,9 @@ class _Backend:
self.remote_branches = remote_branches or {"feature"}
self.default_branch = default_branch
self.push_url = push_url
# Effective push URLs (`git remote get-url --push --all`); defaults to [push_url].
self.push_urls = push_urls if push_urls is not None else ([push_url] if push_url else [])
self.url_rewrites = url_rewrites
self.ls_tree_head_ok = ls_tree_head_ok
self.commands: list[str] = []
self.head = "a" * 40
@ -76,8 +81,12 @@ class _Backend:
self.commands.append(command)
if "rev-parse --show-toplevel" in command:
return _Response("/repo\n")
if "config --get-regexp" in command:
return _Response(f"{self.url_rewrites}\n") if self.url_rewrites else _Response("", 1)
if "remote get-url --push" in command:
return _Response("", 1) if not self.push_url else _Response(f"{self.push_url}\n")
if not self.push_urls:
return _Response("", 1)
return _Response("".join(f"{url}\n" for url in self.push_urls))
if "ls-tree -r" in command:
if "FETCH_HEAD" in command:
return _Response(self._base_tree())
@ -711,14 +720,21 @@ def test_parse_git_push_guards_wrapped_and_optioned_forms() -> None:
assert isinstance(parsed, guard.ParsedGitPush), command
assert parsed.remote_ref == "feature"
# A push carrying unsupported global options cannot be normalized, so it is blocked
# (fail closed) rather than run unguarded.
# A push carrying unsupported global options, an unknown wrapper, or a value-taking
# wrapper option cannot be normalized, so it is blocked (fail closed) rather than run
# unguarded. None of these may return None.
for command in (
"git -c protocol.version=2 push origin feature",
"git --git-dir=.git push origin feature",
"git --no-pager push origin feature",
"nice -n 10 git push origin feature",
"sudo -u ci git push origin feature",
"timeout 5 git push origin feature",
"ionice -c 2 git push origin feature",
):
assert isinstance(guard._parse_git_push(command), guard._BlockedGitPush), command
result = guard._parse_git_push(command)
assert result is not None, command
assert isinstance(result, guard._BlockedGitPush), command
# A non-push git command is still ignored.
assert guard._parse_git_push("/usr/bin/git status") is None
@ -775,6 +791,104 @@ async def test_base_fetched_from_push_url_not_origin_remote(
assert not any(" fetch origin " in cmd for cmd in backend.commands)
async def test_multiple_push_urls_fail_closed(monkeypatch: pytest.MonkeyPatch) -> None:
# `git push` sends to ALL configured push URLs, but `get-url --push` shows only the
# first. If the head workflow tree matches the base fetched from the first (benign)
# URL, the guard must still require approval rather than skip, because a second push
# URL could land the change on the real repo. Head == base here, but multiple push
# URLs force fail-closed (approval required).
backend = _Backend(
tree_entries={".github/workflows/ci.yml": "same-sha"},
base_tree_entries={".github/workflows/ci.yml": "same-sha"},
push_urls=[
"https://github.com/attacker/local.git",
"https://github.com/langchain-ai/open-swe.git",
],
)
guard.SANDBOX_BACKENDS["thread-1"] = backend
async def fake_approved(thread_id: str, fingerprint: str) -> bool:
return False
async def fake_rejected(thread_id: str, fingerprint: str) -> bool:
return False
async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None:
return None
async def fake_pending(thread_id: str, **kwargs: Any) -> tuple[dict[str, Any], bool]:
return {"fingerprint": kwargs["fingerprint"], "status": "pending", "notified": False}, True
async def fake_post(*args: Any, **kwargs: Any) -> tuple[str, None]:
return "1700000000.000300", None
async def fake_notified(*args: Any, **kwargs: Any) -> None:
return None
monkeypatch.setattr(guard, "workflow_push_approved", fake_approved)
monkeypatch.setattr(guard, "workflow_push_rejected", fake_rejected)
monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval)
monkeypatch.setattr(guard, "ensure_workflow_push_pending", fake_pending)
monkeypatch.setattr(guard, "post_slack_thread_reply_with_ts", fake_post)
monkeypatch.setattr(guard, "mark_workflow_push_notified", fake_notified)
async def handler(_request: Any) -> ToolMessage:
raise AssertionError("push must not run without approval when push URLs are ambiguous")
result = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
assert isinstance(result, ToolMessage)
assert result.status == "error"
payload = json.loads(str(result.content))
assert payload["workflow_approval_status"] == "approval_required"
async def test_url_rewrite_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None:
# An insteadOf/pushInsteadOf rewrite means the inspected push URL may not be the real
# destination, so the guard must require approval (never auto-skip).
backend = _Backend(
tree_entries={".github/workflows/ci.yml": "same-sha"},
base_tree_entries={".github/workflows/ci.yml": "same-sha"},
url_rewrites="url.https://evil.example/.insteadof https://github.com/",
)
guard.SANDBOX_BACKENDS["thread-1"] = backend
async def fake_approved(thread_id: str, fingerprint: str) -> bool:
return False
async def fake_rejected(thread_id: str, fingerprint: str) -> bool:
return False
async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None:
return None
async def fake_pending(thread_id: str, **kwargs: Any) -> tuple[dict[str, Any], bool]:
return {"fingerprint": kwargs["fingerprint"], "status": "pending", "notified": False}, True
async def fake_post(*args: Any, **kwargs: Any) -> tuple[str, None]:
return "1700000000.000300", None
async def fake_notified(*args: Any, **kwargs: Any) -> None:
return None
monkeypatch.setattr(guard, "workflow_push_approved", fake_approved)
monkeypatch.setattr(guard, "workflow_push_rejected", fake_rejected)
monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval)
monkeypatch.setattr(guard, "ensure_workflow_push_pending", fake_pending)
monkeypatch.setattr(guard, "post_slack_thread_reply_with_ts", fake_post)
monkeypatch.setattr(guard, "mark_workflow_push_notified", fake_notified)
async def handler(_request: Any) -> ToolMessage:
raise AssertionError("push must not run without approval when a URL rewrite exists")
result = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
assert isinstance(result, ToolMessage)
assert result.status == "error"
payload = json.loads(str(result.content))
assert payload["workflow_approval_status"] == "approval_required"
async def test_unreadable_head_tree_blocks(monkeypatch: pytest.MonkeyPatch) -> None:
# If `ls-tree` on the head cannot be read, the guard must fail closed (block), not
# skip the guard and run the original push.