auth: G3.2 hub JWKS verifier — humans can sign in to the plane (SSO #14)
The human path refused every token in production (503 human_signin_not_ready) because no verifier existed. api/app/hub_jwt.py verifies hub access tokens against account.windyword.ai's JWKS: - RS256 only (closes alg:none and HS256-with-public-key confusion) - iss must be "windy-identity" — what hub ACCESS tokens carry (observed live); id_tokens (discovery-URL issuer) are not accepted as bearers - aud optional today, must name Windy Git when present; hub_require_aud flips it mandatory once the hub emits it. PyJWT's own aud check is off on purpose: it rejects ANY aud-bearing token when no audience is given. - type must be human; identity = windy_identity_id, never sub (per-row id) - production verifies even if require_verified_jwt is off 11 behavioral tests sign real RS256 tokens with a local key. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
156
api/tests/test_hub_jwt.py
Normal file
156
api/tests/test_hub_jwt.py
Normal file
@@ -0,0 +1,156 @@
|
||||
"""G3.2 / I-8 — human tokens are verified, never read (SSO #14, 2026-09-23).
|
||||
|
||||
Behavioral: every case signs a real RS256 token with a locally generated key
|
||||
and drives `get_caller`, so a green run means the gate refuses what it must —
|
||||
not that some string appears in auth.py.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import time
|
||||
|
||||
import jwt
|
||||
import pytest
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
|
||||
from api.app import hub_jwt
|
||||
from api.app.config import Settings
|
||||
from api.app.errors import RepairPointer
|
||||
|
||||
KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||||
OTHER = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||||
IDENTITY = "5e1b9569-7f01-489d-bf14-6fe5a367fa3f"
|
||||
|
||||
|
||||
def _claims(**over):
|
||||
now = int(time.time())
|
||||
c = {
|
||||
"iss": "windy-identity",
|
||||
"type": "human",
|
||||
"sub": "row-id-not-identity",
|
||||
"windy_identity_id": IDENTITY,
|
||||
"email": "grant@example.com",
|
||||
"iat": now,
|
||||
"exp": now + 900,
|
||||
}
|
||||
c.update(over)
|
||||
return {k: v for k, v in c.items() if v is not None}
|
||||
|
||||
|
||||
def _sign(claims, key=KEY, alg="RS256"):
|
||||
return jwt.encode(claims, key, algorithm=alg, headers={"kid": "test"})
|
||||
|
||||
|
||||
class _Req:
|
||||
def __init__(self, settings):
|
||||
self.app = type("A", (), {"state": type("S", (), {"settings": settings})()})()
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _local_jwks(monkeypatch):
|
||||
"""The hub's JWKS, served from KEY's public half — no network."""
|
||||
|
||||
class _Key:
|
||||
key = KEY.public_key()
|
||||
|
||||
class _Client:
|
||||
def get_signing_key_from_jwt(self, token):
|
||||
return _Key()
|
||||
|
||||
monkeypatch.setattr(hub_jwt, "_client", lambda base_url: _Client())
|
||||
|
||||
|
||||
async def _caller(token, **settings):
|
||||
from api.app.auth import get_caller
|
||||
|
||||
s = Settings(environment="production", **settings)
|
||||
return await get_caller(_Req(s), authorization=f"Bearer {token}", x_service_token=None)
|
||||
|
||||
|
||||
async def _refused(token, **settings):
|
||||
with pytest.raises(RepairPointer) as exc:
|
||||
await _caller(token, **settings)
|
||||
assert exc.value.status_code == 401 and exc.value.code == "token_invalid"
|
||||
return exc.value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_genuine_hub_token_is_a_human_named_by_windy_identity_id():
|
||||
c = await _caller(_sign(_claims()))
|
||||
assert c.actor_type == "human"
|
||||
assert c.identity_id == IDENTITY # NOT `sub`, which is the per-row user id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_forged_signature_is_refused():
|
||||
await _refused(_sign(_claims(), key=OTHER))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_expired_token_is_refused():
|
||||
await _refused(_sign(_claims(iat=int(time.time()) - 2000, exp=int(time.time()) - 60)))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_wrong_issuer_and_id_tokens_are_refused():
|
||||
await _refused(_sign(_claims(iss="https://evil.example")))
|
||||
# An id_token (discovery-URL issuer, a relying party's aud) is not a bearer.
|
||||
await _refused(_sign(_claims(iss="https://account.windyword.ai", aud="some-other-client")))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_hs256_confusion_is_refused():
|
||||
# The classic forgery: HMAC the token with the PUBLIC key as the secret.
|
||||
pub = KEY.public_key().public_bytes(
|
||||
serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo
|
||||
)
|
||||
header = base64.urlsafe_b64encode(json.dumps({"alg": "HS256", "typ": "JWT"}).encode()).rstrip(
|
||||
b"="
|
||||
)
|
||||
body = base64.urlsafe_b64encode(json.dumps(_claims()).encode()).rstrip(b"=")
|
||||
sig = base64.urlsafe_b64encode(
|
||||
hmac.new(pub, header + b"." + body, hashlib.sha256).digest()
|
||||
).rstrip(b"=")
|
||||
await _refused((header + b"." + body + b"." + sig).decode())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_non_human_type_is_refused():
|
||||
await _refused(_sign(_claims(type="agent")))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_missing_windy_identity_is_refused_not_read_from_sub():
|
||||
await _refused(_sign(_claims(windy_identity_id=None)))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_aud_is_tolerated_when_it_names_windy_git_and_refused_otherwise():
|
||||
"""PyJWT rejects ANY aud-bearing token when no audience is configured — the
|
||||
trap that would break the day the hub starts emitting aud."""
|
||||
c = await _caller(_sign(_claims(aud=["windy-chat", "windy-git"])))
|
||||
assert c.identity_id == IDENTITY
|
||||
await _refused(_sign(_claims(aud="windy-chat")))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_require_aud_refuses_tokens_without_it():
|
||||
await _refused(_sign(_claims()), hub_require_aud=True)
|
||||
c = await _caller(_sign(_claims(aud="windy-git")), hub_require_aud=True)
|
||||
assert c.identity_id == IDENTITY
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_production_verifies_even_if_the_flag_is_off():
|
||||
"""require_verified_jwt=False is a local-dev convenience; production must
|
||||
never take the unverified path."""
|
||||
await _refused(_sign(_claims(), key=OTHER), require_verified_jwt=False)
|
||||
|
||||
|
||||
def test_algorithm_list_is_exactly_rs256():
|
||||
assert hub_jwt.ALGORITHMS == ["RS256"]
|
||||
@@ -308,12 +308,13 @@ def test_g36_trust_client_never_soft_allows():
|
||||
|
||||
def test_g36_unverified_human_jwt_is_refused_in_production():
|
||||
"""I-8 applied to ourselves: an unverified JWT is an authentication bypass,
|
||||
not a shortcut. Until G3.2's JWKS verifier exists, production refuses."""
|
||||
not a shortcut. G3.2's verifier now exists; behavioral proof that forged,
|
||||
expired, mis-issued and mis-audienced tokens are refused lives in
|
||||
test_hub_jwt.py. Here: the gate defaults closed."""
|
||||
from api.app.config import Settings
|
||||
|
||||
assert Settings().require_verified_jwt is True
|
||||
src = (ROOT / "api" / "app" / "auth.py").read_text()
|
||||
assert "human_signin_not_ready" in src
|
||||
assert Settings().hub_issuers == ["windy-identity"]
|
||||
|
||||
|
||||
def test_no_auth_bypass_env_var_anywhere():
|
||||
|
||||
Reference in New Issue
Block a user