mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-09-30 16:13:15 +00:00
Adam Moussa
ae1f883b4c
Applies the plan's C5 step: git mv every test per the domain-reorg
move-map (movemap-m50.txt) into tests/{agent,analyzer,auth,dashboard,
github,middleware,models,reviewer,sandbox,slack,tools,webhooks}/, plus
the 13 fork-only placements from the scoping report §2c (Atlassian
webhook tests -> tests/webhooks/, test_atlassian_connect.py and
test_auth_error_leak.py -> tests/auth/, jira/confluence util tests ->
tests/tools/, test_repo_binding_isolation.py -> tests/sandbox/,
bot-identity/autofix tests -> tests/github/).
Path-only move: the only content edits are parents[1] -> parents[2]
fixes in test_e2b_integration.py and test_daytona_integration.py,
required because their __file__-relative ROOT path gained one more
directory level in the move.
Monkeypatch retargets for these files were already completed in C4;
none remained outstanding here.
133 lines
5.7 KiB
Python
133 lines
5.7 KiB
Python
"""Shared-secret verification for the Jira Automation webhook (AUTHZ)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from agent.webhooks import common as webhook_common
|
|
|
|
_SECRET = "jira-automation-secret"
|
|
|
|
|
|
def _signed_body(secret: str, *, fresh: bool = True) -> tuple[bytes, str]:
|
|
ts_ms = datetime.now(UTC).timestamp() * 1000
|
|
if not fresh:
|
|
ts_ms -= (webhook_common.JIRA_WEBHOOK_MAX_AGE_SECONDS + 60) * 1000
|
|
body = json.dumps({"issue_key": "PROJ-1", "timestamp": ts_ms}).encode()
|
|
sig = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
|
|
return body, sig
|
|
|
|
|
|
def test_valid_secret_accepted(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_SECRET", _SECRET)
|
|
headers = {"X-Automation-Webhook-Token": _SECRET}
|
|
assert webhook_common.verify_jira_secret(headers) is True
|
|
|
|
|
|
def test_wrong_secret_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_SECRET", _SECRET)
|
|
headers = {"X-Automation-Webhook-Token": "wrong-token"}
|
|
assert webhook_common.verify_jira_secret(headers) is False
|
|
|
|
|
|
def test_missing_header_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_SECRET", _SECRET)
|
|
assert webhook_common.verify_jira_secret({}) is False
|
|
|
|
|
|
def test_empty_header_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_SECRET", _SECRET)
|
|
assert webhook_common.verify_jira_secret({"X-Automation-Webhook-Token": ""}) is False
|
|
|
|
|
|
def test_unset_env_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_SECRET", "")
|
|
headers = {"X-Automation-Webhook-Token": _SECRET}
|
|
assert webhook_common.verify_jira_secret(headers) is False
|
|
|
|
|
|
# --- Opt-in HMAC body signature + timestamp (JIRA_WEBHOOK_REQUIRE_SIGNATURE) ---
|
|
|
|
|
|
def test_signature_check_is_noop_when_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_REQUIRE_SIGNATURE", False)
|
|
assert webhook_common.verify_jira_signature(b"{}", {}) is True
|
|
|
|
|
|
def test_valid_signature_and_fresh_timestamp_accepted(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_REQUIRE_SIGNATURE", True)
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_SECRET", _SECRET)
|
|
body, sig = _signed_body(_SECRET)
|
|
assert webhook_common.verify_jira_signature(body, {"X-Openswe-Signature": sig}) is True
|
|
|
|
|
|
def test_missing_signature_rejected_when_required(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_REQUIRE_SIGNATURE", True)
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_SECRET", _SECRET)
|
|
body, _sig = _signed_body(_SECRET)
|
|
assert webhook_common.verify_jira_signature(body, {}) is False
|
|
|
|
|
|
def test_wrong_signature_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_REQUIRE_SIGNATURE", True)
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_SECRET", _SECRET)
|
|
body, _sig = _signed_body(_SECRET)
|
|
assert webhook_common.verify_jira_signature(body, {"X-Openswe-Signature": "deadbeef"}) is False
|
|
|
|
|
|
def test_stale_timestamp_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_REQUIRE_SIGNATURE", True)
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_SECRET", _SECRET)
|
|
body, sig = _signed_body(_SECRET, fresh=False)
|
|
assert webhook_common.verify_jira_signature(body, {"X-Openswe-Signature": sig}) is False
|
|
|
|
|
|
# --- Opt-in source-IP allowlist (JIRA_WEBHOOK_IP_ALLOWLIST) ---
|
|
|
|
|
|
def _req(host: str | None) -> object:
|
|
client = None if host is None else SimpleNamespace(host=host)
|
|
return SimpleNamespace(client=client)
|
|
|
|
|
|
def test_ip_check_is_noop_when_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_IP_ALLOWLIST", ())
|
|
assert webhook_common.verify_jira_source_ip(_req("9.9.9.9")) is True
|
|
|
|
|
|
def test_ip_in_allowlist_accepted(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_IP_ALLOWLIST", ("10.0.0.0/24",))
|
|
assert webhook_common.verify_jira_source_ip(_req("10.0.0.5")) is True
|
|
|
|
|
|
def test_ip_not_in_allowlist_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_IP_ALLOWLIST", ("10.0.0.0/24",))
|
|
assert webhook_common.verify_jira_source_ip(_req("192.168.1.1")) is False
|
|
|
|
|
|
def test_ip_missing_client_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "JIRA_WEBHOOK_IP_ALLOWLIST", ("10.0.0.0/24",))
|
|
assert webhook_common.verify_jira_source_ip(_req(None)) is False
|
|
|
|
|
|
# --- Fail-closed repo allowlist (REQUIRE_REPO_ALLOWLIST) ---
|
|
|
|
|
|
def test_empty_allowlist_allows_all_by_default(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "ALLOWED_GITHUB_ORGS", frozenset())
|
|
monkeypatch.setattr(webhook_common, "ALLOWED_GITHUB_REPOS", frozenset())
|
|
monkeypatch.setattr(webhook_common, "REQUIRE_REPO_ALLOWLIST", False)
|
|
assert webhook_common._is_repo_allowed({"owner": "anyone", "name": "anything"}) is True
|
|
|
|
|
|
def test_empty_allowlist_fails_closed_when_required(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(webhook_common, "ALLOWED_GITHUB_ORGS", frozenset())
|
|
monkeypatch.setattr(webhook_common, "ALLOWED_GITHUB_REPOS", frozenset())
|
|
monkeypatch.setattr(webhook_common, "REQUIRE_REPO_ALLOWLIST", True)
|
|
assert webhook_common._is_repo_allowed({"owner": "anyone", "name": "anything"}) is False
|