rfc-app/backend/tests/test_turnstile_vertical.py

"""End-to-end integration tests for the v0.12.0 CloudFlare Turnstile
gate on `/auth/otc/request` (§6.2 / roadmap item #10).

The release gates the OTC request endpoint behind a one-step
browser-side Turnstile challenge before the bcrypt hash + SMTP send.
The tests prove:

  * Happy path: with the secret set, a valid token admits the request
    and the OTC envelope lands.
  * Failure path: with the secret set, a token siteverify rejects
    refuses the request with 400 and produces no envelope.
  * Missing-token: with the secret set, a request without a token
    refuses with 400.
  * Missing-secret-soft: with the secret unset AND
    `TURNSTILE_REQUIRED=false` (the v0.12.0 default), the request
    admits — this is the dev / "operator hasn't wired it yet" path.
  * Missing-secret-hard: with the secret unset AND
    `TURNSTILE_REQUIRED=true`, the request refuses with 500
    "auth misconfigured" — the production fail-closed path once
    the operator has flipped the policy.

The Turnstile siteverify call is mocked at the `httpx.post` boundary
inside `app.turnstile` so no real keys are needed and no real
CloudFlare call is made. The Gitea fakes from `test_propose_vertical`
remain in scope so the rest of the app boots cleanly.
"""
from __future__ import annotations

from types import SimpleNamespace

import pytest

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


def _reset_outbound():
    from app import email as email_mod
    email_mod.reset_sent_envelopes()


def _outbound_otc_envelopes(to_address: str | None = None) -> list[dict]:
    from app import email as email_mod
    out = []
    for env in email_mod.sent_envelopes():
        if env.get("kind") != "otc":
            continue
        if to_address is not None and env["to"] != to_address:
            continue
        out.append(env)
    return out


def _patch_siteverify(monkeypatch, *, success: bool, error_codes: list[str] | None = None):
    """Replace `turnstile._siteverify_post` with an async stub that
    returns the requested success shape. The stub does not touch the
    real CloudFlare endpoint and never sees a real secret.

    I4 (security-audit-0026): the siteverify call is now awaited on an
    `httpx.AsyncClient`, isolated behind the `_siteverify_post` seam.
    Patching that narrow function (rather than the shared
    `httpx.AsyncClient`, which gitea/docs also construct) keeps app boot
    intact.
    """
    captured = {}

    async def fake_post(url, data):
        captured["url"] = url
        captured["data"] = data
        body = {"success": bool(success)}
        if error_codes is not None:
            body["error-codes"] = error_codes
        return SimpleNamespace(json=lambda: body)

    from app import turnstile as turnstile_mod
    monkeypatch.setattr(turnstile_mod, "_siteverify_post", fake_post)
    return captured


# ---------------------------------------------------------------------------
# Happy path: secret set, token valid → admit + OTC envelope lands
# ---------------------------------------------------------------------------


def test_otc_request_admits_when_turnstile_token_is_valid(app_with_fake_gitea, monkeypatch):
    from fastapi.testclient import TestClient

    monkeypatch.setenv("CLOUDFLARE_TURNSTILE_SECRET", "test-secret-not-real")
    monkeypatch.setenv("TURNSTILE_REQUIRED", "true")
    captured = _patch_siteverify(monkeypatch, success=True)

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        r = client.post(
            "/auth/otc/request",
            json={"email": "alice@example.com", "turnstile_token": "fake-token-abc"},
        )
        assert r.status_code == 200, r.text
        # The siteverify call was made with the secret + the token we sent.
        assert captured["data"]["secret"] == "test-secret-not-real"
        assert captured["data"]["response"] == "fake-token-abc"
        # And the OTC dispatch ran — exactly one envelope to the address.
        envs = _outbound_otc_envelopes("alice@example.com")
        assert len(envs) == 1


# ---------------------------------------------------------------------------
# Failure path: secret set, siteverify says success=false → 400 + no envelope
# ---------------------------------------------------------------------------


def test_otc_request_refuses_when_turnstile_siteverify_fails(app_with_fake_gitea, monkeypatch):
    from fastapi.testclient import TestClient

    monkeypatch.setenv("CLOUDFLARE_TURNSTILE_SECRET", "test-secret-not-real")
    monkeypatch.setenv("TURNSTILE_REQUIRED", "true")
    _patch_siteverify(monkeypatch, success=False, error_codes=["invalid-input-response"])

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        r = client.post(
            "/auth/otc/request",
            json={"email": "alice@example.com", "turnstile_token": "fake-bad-token"},
        )
        assert r.status_code == 400, r.text
        # The OTC bcrypt + SMTP path did not run — no envelope was buffered.
        assert _outbound_otc_envelopes("alice@example.com") == []


# ---------------------------------------------------------------------------
# Missing-token: secret set, no token → 400 + no envelope
# ---------------------------------------------------------------------------


def test_otc_request_refuses_when_turnstile_token_is_missing(app_with_fake_gitea, monkeypatch):
    from fastapi.testclient import TestClient

    monkeypatch.setenv("CLOUDFLARE_TURNSTILE_SECRET", "test-secret-not-real")
    monkeypatch.setenv("TURNSTILE_REQUIRED", "true")
    # Even though we patch httpx.post, the missing-token check fires
    # before the siteverify call — so the patch is here only as a
    # safety net in case the implementation regresses to making the
    # network call anyway.
    _patch_siteverify(monkeypatch, success=False)

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        r = client.post(
            "/auth/otc/request",
            json={"email": "alice@example.com"},  # no turnstile_token field at all
        )
        assert r.status_code == 400, r.text
        assert _outbound_otc_envelopes("alice@example.com") == []


# ---------------------------------------------------------------------------
# Missing-secret-soft: no secret, TURNSTILE_REQUIRED=false (default) → admit
# ---------------------------------------------------------------------------


def test_otc_request_admits_when_secret_unset_and_not_required(app_with_fake_gitea, monkeypatch):
    """v0.12.0 default: the operator has not yet wired the Turnstile
    secret and has not enabled `TURNSTILE_REQUIRED`. The gate stays
    open — this is the dev / test / pre-rollout path. Once the
    operator confirms the secret is in place and flips
    `TURNSTILE_REQUIRED=true`, missing-secret becomes fail-closed
    (covered in test_otc_request_refuses_when_required_but_secret_unset).
    """
    from fastapi.testclient import TestClient

    monkeypatch.delenv("CLOUDFLARE_TURNSTILE_SECRET", raising=False)
    monkeypatch.delenv("TURNSTILE_REQUIRED", raising=False)
    # The siteverify call inside turnstile must not be made in this path —
    # patch the seam to a sentinel that explodes if it ever runs
    # (I4: the seam is now `_siteverify_post`, not module-level httpx.post).
    from app import turnstile as turnstile_mod

    async def must_not_be_called(*a, **kw):
        raise AssertionError("siteverify should not run when no secret is configured")

    monkeypatch.setattr(turnstile_mod, "_siteverify_post", must_not_be_called)

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        r = client.post(
            "/auth/otc/request",
            json={"email": "alice@example.com"},
        )
        assert r.status_code == 200, r.text
        # The OTC path ran end-to-end — one envelope to the address.
        assert len(_outbound_otc_envelopes("alice@example.com")) == 1


# ---------------------------------------------------------------------------
# Missing-secret-hard: no secret, TURNSTILE_REQUIRED=true → 500 "misconfigured"
# ---------------------------------------------------------------------------


def test_otc_request_refuses_when_required_but_secret_unset(app_with_fake_gitea, monkeypatch):
    """Once the operator has flipped `TURNSTILE_REQUIRED=true` to lock
    down production, a missing secret stops being a soft-fail and
    becomes a fail-closed 500. This is the regression-detection shape
    the §20.4 upgrade-steps MAY block calls out — flip the flag once
    the secret is wired so a future config drift fails loudly instead
    of silently disabling abuse defense.
    """
    from fastapi.testclient import TestClient

    monkeypatch.delenv("CLOUDFLARE_TURNSTILE_SECRET", raising=False)
    monkeypatch.setenv("TURNSTILE_REQUIRED", "true")

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        r = client.post(
            "/auth/otc/request",
            json={"email": "alice@example.com", "turnstile_token": "doesnt-matter"},
        )
        assert r.status_code == 500, r.text
        assert _outbound_otc_envelopes("alice@example.com") == []


# ---------------------------------------------------------------------------
# I4 (security-audit-0026): verify_token is a coroutine — calling it returns
# an awaitable, not a VerifyOutcome. Locks the async contract so a revert to
# the synchronous event-loop-blocking shape fails here, not just in the
# integration paths.
# ---------------------------------------------------------------------------


def test_verify_token_is_async_and_soft_skips_without_secret(monkeypatch):
    import asyncio

    from app import turnstile as turnstile_mod

    monkeypatch.delenv("CLOUDFLARE_TURNSTILE_SECRET", raising=False)
    monkeypatch.delenv("TURNSTILE_REQUIRED", raising=False)

    coro = turnstile_mod.verify_token("any-token")
    assert asyncio.iscoroutine(coro), "verify_token must be a coroutine (I4)"
    outcome = asyncio.run(coro)
    # No secret + not required → the gate stays open without any network call.
    assert outcome.ok is True
    assert outcome.reason == "skipped"