rfc-app/backend/tests/test_webhooks_vertical.py

"""End-to-end integration tests for the Gitea webhook receiver
(v0.18.0 Slice 3 — webhook tightening per the email + webhook
hygiene proposal).

The release changes the receiver from "verifies the signature only
when a secret is configured; silently accepts unsigned POSTs
otherwise" to "requires the secret unless `RFC_APP_INSECURE_WEBHOOKS=1`
is set as an explicit dev-bypass." The startup-time check lives in
`config.load_config()`; the request-time check lives in
`webhooks.receive`.

These tests prove:

  * The framework refuses to start when `GITEA_WEBHOOK_SECRET` is
    empty and the dev-bypass is not set.
  * The dev-bypass (`RFC_APP_INSECURE_WEBHOOKS=1`) lets the
    framework boot with an empty secret AND lets webhook POSTs
    land without signature verification (a loud-warning log line
    surfaces, but the request is accepted).
  * Default path (secret bound): a POST with a valid signature
    lands; a POST with an invalid signature gets 401; a POST with
    no signature gets 401.
  * Unknown-repo POSTs surface in the log (the "stale Gitea hook"
    case the proposal targets).
"""
from __future__ import annotations

import hashlib
import hmac
import json
import logging

import pytest

from test_propose_vertical import (  # noqa: F401
    FakeGitea,
    app_with_fake_gitea,
    tmp_env,
)


# ---------------------------------------------------------------------------
# Startup-time secret check (config.load_config)
# ---------------------------------------------------------------------------


def test_config_refuses_to_load_with_empty_secret_and_no_bypass(monkeypatch, tmp_path):
    """The framework MUST refuse to start when `GITEA_WEBHOOK_SECRET`
    is empty unless `RFC_APP_INSECURE_WEBHOOKS=1` is set. This is
    the v0.18.0 startup-loud-failure shape — silent acceptance was
    the bug."""
    monkeypatch.setenv("GITEA_URL", "http://gitea.test")
    monkeypatch.setenv("GITEA_BOT_USER", "rfc-bot")
    monkeypatch.setenv("GITEA_BOT_TOKEN", "bot-token")
    monkeypatch.setenv("GITEA_ORG", "wiggleverse")
    monkeypatch.setenv("OAUTH_CLIENT_ID", "cid")
    monkeypatch.setenv("OAUTH_CLIENT_SECRET", "csec")
    monkeypatch.setenv("SECRET_KEY", "test-secret-key")
    monkeypatch.setenv("DATABASE_PATH", str(tmp_path / "test.db"))
    monkeypatch.setenv("GITEA_WEBHOOK_SECRET", "")
    monkeypatch.delenv("RFC_APP_INSECURE_WEBHOOKS", raising=False)

    from app.config import load_config
    with pytest.raises(RuntimeError, match="GITEA_WEBHOOK_SECRET"):
        load_config()


def test_config_loads_with_empty_secret_when_bypass_is_set(monkeypatch, tmp_path):
    """The explicit `RFC_APP_INSECURE_WEBHOOKS=1` opt-in lets the
    framework boot with an empty webhook secret. This is the
    local-dev escape hatch."""
    monkeypatch.setenv("GITEA_URL", "http://gitea.test")
    monkeypatch.setenv("GITEA_BOT_USER", "rfc-bot")
    monkeypatch.setenv("GITEA_BOT_TOKEN", "bot-token")
    monkeypatch.setenv("GITEA_ORG", "wiggleverse")
    monkeypatch.setenv("REGISTRY_REPO", "registry")
    monkeypatch.setenv("OAUTH_CLIENT_ID", "cid")
    monkeypatch.setenv("OAUTH_CLIENT_SECRET", "csec")
    monkeypatch.setenv("SECRET_KEY", "test-secret-key")
    monkeypatch.setenv("DATABASE_PATH", str(tmp_path / "test.db"))
    monkeypatch.setenv("GITEA_WEBHOOK_SECRET", "")
    monkeypatch.setenv("RFC_APP_INSECURE_WEBHOOKS", "1")

    from app.config import load_config
    cfg = load_config()  # MUST NOT raise
    assert cfg.webhook_secret == ""


def test_config_loads_with_secret_set(monkeypatch, tmp_path):
    """Sanity: the happy path (secret bound, bypass not set) loads
    cleanly."""
    monkeypatch.setenv("GITEA_URL", "http://gitea.test")
    monkeypatch.setenv("GITEA_BOT_USER", "rfc-bot")
    monkeypatch.setenv("GITEA_BOT_TOKEN", "bot-token")
    monkeypatch.setenv("GITEA_ORG", "wiggleverse")
    monkeypatch.setenv("REGISTRY_REPO", "registry")
    monkeypatch.setenv("OAUTH_CLIENT_ID", "cid")
    monkeypatch.setenv("OAUTH_CLIENT_SECRET", "csec")
    monkeypatch.setenv("SECRET_KEY", "test-secret-key")
    monkeypatch.setenv("DATABASE_PATH", str(tmp_path / "test.db"))
    monkeypatch.setenv("GITEA_WEBHOOK_SECRET", "my-real-secret")
    monkeypatch.delenv("RFC_APP_INSECURE_WEBHOOKS", raising=False)

    from app.config import load_config
    cfg = load_config()
    assert cfg.webhook_secret == "my-real-secret"


# ---------------------------------------------------------------------------
# Request-time signature verification (webhooks.receive)
#
# The default `app_with_fake_gitea` fixture binds
# `GITEA_WEBHOOK_SECRET=test-webhook-secret-for-signature-verification`,
# so these tests exercise the production path.
# ---------------------------------------------------------------------------


_SECRET = "test-webhook-secret-for-signature-verification"


def _sign(body: bytes) -> str:
    return hmac.new(_SECRET.encode("utf-8"), body, hashlib.sha256).hexdigest()


def test_webhook_post_with_valid_signature_accepted(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        body = json.dumps({"repository": {"full_name": "wiggleverse/meta"}}).encode()
        sig = _sign(body)
        r = client.post(
            "/api/webhooks/gitea",
            content=body,
            headers={
                "X-Gitea-Event": "push",
                "X-Gitea-Signature": sig,
                "Content-Type": "application/json",
            },
        )
        assert r.status_code == 200, r.text


def test_webhook_post_with_invalid_signature_refused_401(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        body = json.dumps({"repository": {"full_name": "wiggleverse/meta"}}).encode()
        r = client.post(
            "/api/webhooks/gitea",
            content=body,
            headers={
                "X-Gitea-Event": "push",
                "X-Gitea-Signature": "0" * 64,  # wrong signature
                "Content-Type": "application/json",
            },
        )
        assert r.status_code == 401


def test_webhook_post_with_missing_signature_refused_401(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        body = json.dumps({"repository": {"full_name": "wiggleverse/meta"}}).encode()
        r = client.post(
            "/api/webhooks/gitea",
            content=body,
            headers={
                "X-Gitea-Event": "push",
                "Content-Type": "application/json",
            },
        )
        assert r.status_code == 401


# ---------------------------------------------------------------------------
# Unknown-repo logging (the "stale hook on a fork" surface)
# ---------------------------------------------------------------------------


def test_webhook_unknown_repo_logs_at_info(app_with_fake_gitea, caplog):
    """Per the proposal: a hook on a fork or a stale Gitea binding
    used to silently 200-OK. v0.18.0 surfaces it as an INFO log."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        body = json.dumps({"repository": {"full_name": "someone-else/unrelated"}}).encode()
        sig = _sign(body)
        with caplog.at_level(logging.INFO, logger="app.webhooks"):
            r = client.post(
                "/api/webhooks/gitea",
                content=body,
                headers={
                    "X-Gitea-Event": "push",
                    "X-Gitea-Signature": sig,
                    "Content-Type": "application/json",
                },
            )
        assert r.status_code == 200  # the handler still 200s; surface is the log line
        assert any(
            "unknown repo" in rec.message and "someone-else/unrelated" in rec.message
            for rec in caplog.records
        ), f"expected unknown-repo log line; got: {[r.message for r in caplog.records]}"