Spaces:
Sleeping
Sleeping
File size: 8,451 Bytes
b81a86b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 | """
Supabase JWT verification and endpoint gating.
The point of these tests is that the login actually protects something. Each
rejection case below is a way an attacker would try to get past it:
forged signature, no signature at all ("alg": "none"), expired token, wrong
audience, another project's secret, and simply omitting the header.
"""
from __future__ import annotations
import os
import sys
from dataclasses import replace
from datetime import datetime, timedelta, timezone
import jwt
import pytest
from fastapi.testclient import TestClient
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
SECRET = "unit-test-jwt-secret"
OTHER_SECRET = "some-other-projects-secret"
SUB = "11111111-2222-3333-4444-555555555555"
def make_token(
*, secret=SECRET, sub=SUB, aud="authenticated", expires_in=3600,
algorithm="HS256", email="dinas@example.go.id",
):
now = datetime.now(timezone.utc)
payload = {
"sub": sub, "email": email, "role": "authenticated",
"iat": now, "exp": now + timedelta(seconds=expires_in),
}
if aud is not None:
payload["aud"] = aud
return jwt.encode(payload, secret, algorithm=algorithm)
def bearer(token):
return {"Authorization": f"Bearer {token}"}
@pytest.fixture
def client(monkeypatch):
monkeypatch.setenv("SUPABASE_JWT_SECRET", SECRET)
monkeypatch.delenv("SUPABASE_URL", raising=False)
monkeypatch.setenv("PHONE_HASH_SALT", "t")
from whatsapp_bot import server
with TestClient(server.app) as c:
yield c
@pytest.fixture
def strict_client(monkeypatch):
"""A client with REQUIRE_AUTH on — the production posture."""
monkeypatch.setenv("SUPABASE_JWT_SECRET", SECRET)
monkeypatch.delenv("SUPABASE_URL", raising=False)
monkeypatch.setenv("PHONE_HASH_SALT", "t")
monkeypatch.setenv("REQUIRE_AUTH", "true")
from whatsapp_bot import server
with TestClient(server.app) as c:
yield c
# =============================================================================
# A. TOKEN VERIFICATION
# =============================================================================
class TestA_Verification:
def test_valid_token_is_accepted(self, client):
r = client.get("/billing/status", params={"phone": "+628111222333"},
headers=bearer(make_token()))
assert r.status_code == 200
def test_no_header_is_rejected(self, client):
r = client.get("/billing/status", params={"phone": "+628111222333"})
assert r.status_code == 401
def test_forged_signature_is_rejected(self, client):
r = client.get("/billing/status", params={"phone": "+628111222333"},
headers=bearer(make_token(secret=OTHER_SECRET)))
assert r.status_code == 401
def test_alg_none_token_is_rejected(self, client):
# The classic JWT bypass: strip the signature and claim it isn't needed.
now = datetime.now(timezone.utc)
unsigned = jwt.encode(
{"sub": SUB, "aud": "authenticated", "exp": now + timedelta(hours=1)},
key="", algorithm="none",
)
r = client.get("/billing/status", params={"phone": "+628111222333"},
headers=bearer(unsigned))
assert r.status_code == 401
def test_expired_token_is_rejected(self, client):
r = client.get("/billing/status", params={"phone": "+628111222333"},
headers=bearer(make_token(expires_in=-60)))
assert r.status_code == 401
def test_wrong_audience_is_rejected(self, client):
# A token minted for another Supabase surface must not open this door.
r = client.get("/billing/status", params={"phone": "+628111222333"},
headers=bearer(make_token(aud="some-other-service")))
assert r.status_code == 401
def test_token_without_subject_is_rejected(self, client):
r = client.get("/billing/status", params={"phone": "+628111222333"},
headers=bearer(make_token(sub=None)))
assert r.status_code == 401
@pytest.mark.parametrize("header", [
{"Authorization": "Bearer"},
{"Authorization": "Bearer "},
{"Authorization": "Basic abc123"},
{"Authorization": make_token()}, # missing the Bearer scheme
{"Authorization": "Bearer not.a.token"},
])
def test_malformed_headers_are_rejected(self, client, header):
r = client.get("/billing/status", params={"phone": "+628111222333"},
headers=header)
assert r.status_code == 401
def test_rejection_does_not_leak_the_reason(self, client):
# Expired vs forged must look identical, or a prober learns which
# tokens are real.
expired = client.get("/billing/status", params={"phone": "+628111222333"},
headers=bearer(make_token(expires_in=-60)))
forged = client.get("/billing/status", params={"phone": "+628111222333"},
headers=bearer(make_token(secret=OTHER_SECRET)))
assert expired.json() == forged.json()
# =============================================================================
# B. ENDPOINT GATING
# =============================================================================
class TestB_Gating:
PREMIUM = [
("/api/v1/matches", {}),
("/api/v1/forecast", {"commodity": "cabai_rawit", "city": "3578"}),
("/api/v1/anomalies", {}),
]
PUBLIC = ["/api/v1/commodities", "/api/v1/kabupaten"]
@pytest.mark.parametrize("path,params", PREMIUM)
def test_premium_open_when_require_auth_off(self, client, path, params):
# Demo posture: judges can browse without an account.
assert client.get(path, params=params).status_code != 401
@pytest.mark.parametrize("path,params", PREMIUM)
def test_premium_closed_when_require_auth_on(self, strict_client, path, params):
assert strict_client.get(path, params=params).status_code == 401
@pytest.mark.parametrize("path,params", PREMIUM)
def test_premium_open_with_token_when_require_auth_on(self, strict_client, path, params):
r = strict_client.get(path, params=params, headers=bearer(make_token()))
assert r.status_code != 401
@pytest.mark.parametrize("path", PUBLIC)
def test_reference_data_stays_public(self, strict_client, path):
# Commodity and kabupaten lists are public government reference data;
# gating them would break the map for anonymous visitors.
assert strict_client.get(path).status_code == 200
def test_billing_status_ignores_require_auth_flag(self, client):
# Per-person data is gated even in demo posture, because the endpoint
# is an enumeration oracle for phone numbers.
assert client.get("/billing/status",
params={"phone": "+628111222333"}).status_code == 401
def test_health_is_always_public(self, strict_client):
assert strict_client.get("/health").status_code == 200
def test_health_reports_auth_posture(self, strict_client):
body = strict_client.get("/health").json()
assert body["auth_configured"] is True
assert body["require_auth"] is True
# =============================================================================
# C. MISCONFIGURATION
# =============================================================================
class TestC_Misconfiguration:
def test_unconfigured_auth_still_rejects_protected_routes(self, monkeypatch):
# Fail closed: no secret and no URL must mean "nobody gets in",
# never "everybody gets in".
monkeypatch.delenv("SUPABASE_JWT_SECRET", raising=False)
monkeypatch.delenv("SUPABASE_URL", raising=False)
monkeypatch.setenv("PHONE_HASH_SALT", "t")
from whatsapp_bot import server
with TestClient(server.app) as c:
r = c.get("/billing/status", params={"phone": "+628111222333"},
headers=bearer(make_token()))
assert r.status_code == 401
def test_health_flags_unconfigured_auth(self, monkeypatch):
monkeypatch.delenv("SUPABASE_JWT_SECRET", raising=False)
monkeypatch.delenv("SUPABASE_URL", raising=False)
from whatsapp_bot import server
with TestClient(server.app) as c:
assert c.get("/health").json()["auth_configured"] is False
|