# Copyright 2026 Guildhouse Dev # SPDX-License-Identifier: Apache-2.0 """Tests for shared bearer auth middleware — C-4, C-8, H-6, H-7.""" import os import stat import datetime import pytest from unittest.mock import AsyncMock, patch, MagicMock from jose import jwt as jose_jwt from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.primitives import serialization from gsap_broker.auth.middleware import ( _authenticate_token, _peek_issuer, _verifiers, init_authenticator, ) from gsap_broker.drivers.base import AuthResult # ── Test key setup ─────────────────────────────────────────────── def _generate_rsa_keypair(): pk = rsa.generate_private_key(public_exponent=65537, key_size=2048) return pk, pk.public_key() def _pem(private_key): return private_key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption(), ) def _jwk(public_key, kid="mid-kid"): import base64 nums = public_key.public_numbers() e = nums.e.to_bytes((nums.e.bit_length() + 7) // 8, "big") n = nums.n.to_bytes((nums.n.bit_length() + 7) // 8, "big") return { "kty": "RSA", "kid": kid, "use": "sig", "alg": "RS256", "n": base64.urlsafe_b64encode(n).rstrip(b"=").decode(), "e": base64.urlsafe_b64encode(e).rstrip(b"=").decode(), } PRIV, PUB = _generate_rsa_keypair() KID = "mid-kid" KC_URL = "http://keycloak.test:8080" KC_REALM = "test-realm" KC_ISSUER = f"{KC_URL}/realms/{KC_REALM}" ENTRA_TID = "entra-test-tenant" ENTRA_ISSUER = f"https://login.microsoftonline.com/{ENTRA_TID}/v2.0" JWKS = {"keys": [_jwk(PUB, KID)]} def _make_token(issuer, audience="test-client", claims=None, kid=KID): now = datetime.datetime.now(datetime.UTC) base = { "iss": issuer, "aud": audience, "exp": int((now + datetime.timedelta(hours=1)).timestamp()), "iat": int(now.timestamp()), "nbf": int(now.timestamp()), "sub": "user-1", "oid": "user-1", "preferred_username": "alice", "name": "Alice", "jti": "test-jti", } if claims: base.update(claims) return jose_jwt.encode(base, _pem(PRIV), algorithm="RS256", headers={"kid": kid}) @pytest.fixture def mock_jwks(): with patch("gsap_broker.drivers.jwks.httpx.AsyncClient") as m: resp = MagicMock() resp.json.return_value = JWKS resp.raise_for_status = MagicMock() ctx = AsyncMock() ctx.__aenter__.return_value.get = AsyncMock(return_value=resp) m.return_value = ctx yield m @pytest.fixture def setup_verifiers(mock_jwks): """Register test verifiers for both issuers.""" from gsap_broker.drivers.jwks import JWKSVerifier import gsap_broker.auth.middleware as mw mw._verifiers = { KC_ISSUER: JWKSVerifier( jwks_url=f"{KC_ISSUER}/protocol/openid-connect/certs", audience="test-client", issuer=KC_ISSUER, ), ENTRA_ISSUER: JWKSVerifier( jwks_url=f"https://login.microsoftonline.com/{ENTRA_TID}/discovery/v2.0/keys", audience="test-client", issuer=ENTRA_ISSUER, ), } yield mw._verifiers = {} # ── Issuer peek ────────────────────────────────────────────────── def test_peek_issuer_keycloak(): token = _make_token(KC_ISSUER) assert _peek_issuer(token) == KC_ISSUER def test_peek_issuer_entra(): token = _make_token(ENTRA_ISSUER) assert _peek_issuer(token) == ENTRA_ISSUER def test_peek_issuer_malformed(): with pytest.raises(Exception): _peek_issuer("not-a-jwt") # ── Token authentication ───────────────────────────────────────── @pytest.mark.asyncio async def test_authenticate_keycloak_token(setup_verifiers): """TEST 4: Valid Keycloak token accepted.""" token = _make_token(KC_ISSUER) result = await _authenticate_token(token) assert result.is_authorized assert "alice" in result.principal_did or "user-1" in result.principal_did @pytest.mark.asyncio async def test_authenticate_entra_token(setup_verifiers): """TEST 5: Valid Entra token accepted.""" token = _make_token(ENTRA_ISSUER, claims={"oid": "entra-user-1"}) result = await _authenticate_token(token) assert result.is_authorized assert "entra-user-1" in result.principal_did @pytest.mark.asyncio async def test_auto_detect_driver(setup_verifiers): """TEST 6: Correct driver selected based on issuer.""" kc_token = _make_token(KC_ISSUER) entra_token = _make_token(ENTRA_ISSUER) kc_result = await _authenticate_token(kc_token) entra_result = await _authenticate_token(entra_token) assert kc_result.is_authorized assert entra_result.is_authorized @pytest.mark.asyncio async def test_unknown_issuer_rejected(setup_verifiers): """TEST 7: Unknown issuer → 401.""" token = _make_token("https://evil.example.com/auth") with pytest.raises(Exception) as exc_info: await _authenticate_token(token) assert "401" in str(exc_info.value) or "Unknown" in str(exc_info.value) @pytest.mark.asyncio async def test_invalid_token_rejected(setup_verifiers): """TEST 3: Garbage token → 401.""" with pytest.raises(Exception): await _authenticate_token("not.a.jwt") # ── MCP auth (C-4) ────────────────────────────────────────────── @pytest.mark.asyncio async def test_mcp_unauthenticated_rejected(): """TEST 8: POST /mcp without auth → 401 or 403.""" from httpx import AsyncClient as RealAsyncClient, ASGITransport from gsap_broker.app import app async with RealAsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: resp = await c.post("/mcp", json={"jsonrpc": "2.0", "method": "tools/list", "id": 1}) # HTTPBearer auto_error=True returns 403 when no header assert resp.status_code in (401, 403) # ── Delegation auth (C-8) ──────────────────────────────────────── @pytest.mark.asyncio async def test_delegation_create_unauthenticated_rejected(): """TEST 10: POST /delegations/ without auth → 401 or 403.""" from httpx import AsyncClient as RealAsyncClient, ASGITransport from gsap_broker.app import app async with RealAsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: resp = await c.post("/delegations/", json={ "delegator_ac_id": "test", "agent_type": "claude-code", }) assert resp.status_code in (401, 403) # ── Health stays public ────────────────────────────────────────── @pytest.mark.asyncio async def test_health_unauthenticated_ok(): """TEST 16: /health/ remains accessible without auth.""" from httpx import AsyncClient as RealAsyncClient, ASGITransport from gsap_broker.app import app async with RealAsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: resp = await c.get("/health/") assert resp.status_code == 200 assert resp.json()["status"] == "ok" # ── H-6: SQLite permissions ────────────────────────────────────── @pytest.mark.asyncio async def test_sqlite_permissions(): """TEST 21: Database file permissions are 0o600.""" from gsap_broker.db import _restrict_db_permissions # Create a temp file to test permissions on import tempfile with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: test_path = f.name try: os.chmod(test_path, 0o644) # start with world-readable assert stat.S_IMODE(os.stat(test_path).st_mode) == 0o644 # Simulate the fix os.chmod(test_path, stat.S_IRUSR | stat.S_IWUSR) mode = stat.S_IMODE(os.stat(test_path).st_mode) assert mode == 0o600, f"Expected 0o600, got {oct(mode)}" finally: os.unlink(test_path)