open-swe/tests/test_jira_webhook_corroboration.py

"""Route-level corroboration + input validation for /webhooks/jira.

These cover the hardening from the Phase 2 security review: the unsigned webhook
body is only a pointer (issue_key + comment_id), and the triggering comment's
author and text are re-fetched from Jira server-side. A payload-claimed author
must never be trusted, a malformed issue_key must be rejected, and a comment
that can't be corroborated must be rejected.
"""

from __future__ import annotations

import asyncio
import json
from contextlib import ExitStack
from typing import Any
from unittest.mock import AsyncMock, patch

from agent import webapp


class _FakeRequest:
    def __init__(self, body: bytes, headers: dict[str, str] | None = None) -> None:
        self.headers = headers or {}
        self._body = body

    async def body(self) -> bytes:
        return self._body


class _FakeBackgroundTasks:
    def __init__(self) -> None:
        self.tasks: list[tuple[Any, tuple, dict]] = []

    def add_task(self, func: Any, *args: Any, **kwargs: Any) -> None:
        self.tasks.append((func, args, kwargs))


def _call(
    payload: dict[str, Any],
    *,
    server_comment: dict[str, Any] | None,
    email: str | None = "real@example.com",
) -> tuple[dict[str, str], _FakeBackgroundTasks, AsyncMock]:
    req = _FakeRequest(json.dumps(payload).encode())
    bg = _FakeBackgroundTasks()
    get_email = AsyncMock(return_value=email)
    with ExitStack() as stack:
        stack.enter_context(patch.object(webapp, "verify_jira_secret", return_value=True))
        stack.enter_context(
            patch.object(webapp, "fetch_jira_comment", new=AsyncMock(return_value=server_comment))
        )
        stack.enter_context(patch.object(webapp, "get_jira_user_email", new=get_email))
        stack.enter_context(
            patch.object(webapp, "resolve_login_from_email_async", new=AsyncMock(return_value=None))
        )
        stack.enter_context(
            patch.object(webapp, "get_profile_default_repo", new=AsyncMock(return_value=None))
        )
        stack.enter_context(
            patch.object(
                webapp,
                "get_repo_config_from_jira_mapping",
                return_value={"owner": "langchain-ai", "name": "open-swe"},
            )
        )
        stack.enter_context(patch.object(webapp, "_is_repo_allowed", return_value=True))
        result = asyncio.run(webapp.jira_webhook(req, bg))
    return result, bg, get_email


def _server_comment(*, account_id: str, name: str, body: str) -> dict[str, Any]:
    return {"id": "10050", "body": body, "author": {"account_id": account_id, "name": name}}


def test_malformed_issue_key_rejected() -> None:
    result, bg, _ = _call(
        {"issue_key": "../../../../rest/api/2/myself", "comment_id": "1"},
        server_comment=None,
    )
    assert result["status"] == "ignored"
    assert "issue key" in result["reason"].lower()
    assert bg.tasks == []


def test_missing_comment_id_rejected() -> None:
    result, bg, _ = _call({"issue_key": "PROJ-42"}, server_comment=None)
    assert result["status"] == "ignored"
    assert bg.tasks == []


def test_uncorroborated_comment_rejected() -> None:
    # fetch_jira_comment returns None (nonexistent / forged) -> hard reject.
    result, bg, _ = _call(
        {"issue_key": "PROJ-42", "comment_id": "10050", "comment_body": "@openswe do it"},
        server_comment=None,
    )
    assert result["status"] == "ignored"
    assert bg.tasks == []


def test_identity_and_body_come_from_server_not_payload() -> None:
    # Payload claims a victim's account + benign body; the REAL comment (server)
    # has a different author and the actual trigger text. The scheduled task must
    # carry the server author, and email lookup must use the server account id.
    payload = {
        "issue_key": "PROJ-42",
        "comment_id": "10050",
        "comment_author_account_id": "victim-account-id",
        "comment_author_display_name": "Victim",
        "comment_body": "totally benign",
    }
    server = _server_comment(
        account_id="real-author-id", name="Real Author", body="@openswe fix the bug"
    )
    result, bg, get_email = _call(payload, server_comment=server)

    assert result["status"] == "accepted"
    assert len(bg.tasks) == 1
    _func, (issue_data, _repo), _kw = bg.tasks[0]
    # Server author wins; payload's victim account is never used.
    assert issue_data["comment_author"]["account_id"] == "real-author-id"
    assert issue_data["comment_author"]["name"] == "Real Author"
    assert issue_data["triggering_comment"] == "@openswe fix the bug"
    get_email.assert_awaited_once_with("real-author-id")


def test_project_key_derived_from_issue_key() -> None:
    payload = {"issue_key": "OSPROJ-7", "comment_id": "10050", "project_key": "ATTACKER-INJECTED"}
    server = _server_comment(account_id="a", name="A", body="@openswe go")
    result, bg, _ = _call(payload, server_comment=server)
    assert result["status"] == "accepted"
    _func, (issue_data, _repo), _kw = bg.tasks[0]
    assert issue_data["project_key"] == "OSPROJ"


def test_server_comment_without_mention_ignored() -> None:
    # The @openswe check runs on the authoritative server body, not the payload.
    payload = {"issue_key": "PROJ-42", "comment_id": "10050", "comment_body": "@openswe do it"}
    server = _server_comment(account_id="a", name="A", body="just a normal comment")
    result, bg, _ = _call(payload, server_comment=server)
    assert result["status"] == "ignored"
    assert bg.tasks == []