docs: add Claude AI assistant rules and configuration
Add comprehensive rule documentation for AI-assisted development covering authentication surfaces, outbound-call safety invariants, and other project conventions to guide Claude's understanding of critical system behaviors.
This commit is contained in:
239
tests/test_auth.py
Normal file
239
tests/test_auth.py
Normal file
@@ -0,0 +1,239 @@
|
||||
"""
|
||||
Auth tests — Casdoor JWT + PAT resolution, owner gating, dev-owner mode.
|
||||
|
||||
No live Casdoor: we generate an RSA keypair, stub the JWKS client so
|
||||
`_decode_casdoor_jwt` trusts our public key, and mint RS256 JWTs locally.
|
||||
The app is exercised without its lifespan, so a 503 ("Gateway not
|
||||
initialized") proves auth was accepted and the request reached a handler.
|
||||
|
||||
DB access (resolve_bearer → users/PATs) hits an in-memory SQLite database
|
||||
wired in via the `mem_db` fixture, mirroring tests/test_data_layer.py.
|
||||
"""
|
||||
|
||||
import time
|
||||
import uuid
|
||||
|
||||
import httpx
|
||||
import jwt
|
||||
import pytest
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
import auth as authmod
|
||||
import db.database as dbmod
|
||||
import main
|
||||
from config import get_settings
|
||||
from db.database import Base, PersonalAccessToken, User
|
||||
|
||||
ENDPOINT = "https://id.example.test"
|
||||
OWNER = "owner@example.test"
|
||||
|
||||
|
||||
# ── RSA keypair + JWKS stub ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def keypair():
|
||||
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||||
private_pem = key.private_bytes(
|
||||
serialization.Encoding.PEM,
|
||||
serialization.PrivateFormat.PKCS8,
|
||||
serialization.NoEncryption(),
|
||||
)
|
||||
return private_pem, key.public_key()
|
||||
|
||||
|
||||
def _mint(private_pem, *, sub, name, email=None, exp_delta=3600):
|
||||
claims = {
|
||||
"iss": ENDPOINT,
|
||||
"sub": sub,
|
||||
"name": name,
|
||||
"displayName": name,
|
||||
"exp": int(time.time()) + exp_delta,
|
||||
"iat": int(time.time()),
|
||||
}
|
||||
if email:
|
||||
claims["email"] = email
|
||||
return jwt.encode(claims, private_pem, algorithm="RS256")
|
||||
|
||||
|
||||
class _StubJWKS:
|
||||
"""Stands in for jwt.PyJWKClient — returns our fixed public key."""
|
||||
|
||||
def __init__(self, public_key):
|
||||
self._key = public_key
|
||||
|
||||
def get_signing_key_from_jwt(self, token):
|
||||
class _K:
|
||||
key = self._key
|
||||
|
||||
return _K()
|
||||
|
||||
def fetch_data(self):
|
||||
pass
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sso_enabled(monkeypatch, keypair):
|
||||
"""Enable Casdoor SSO with a known owner and a stubbed JWKS client."""
|
||||
_, public_key = keypair
|
||||
settings = get_settings()
|
||||
monkeypatch.setattr(settings.casdoor, "enabled", True)
|
||||
monkeypatch.setattr(settings.casdoor, "endpoint", ENDPOINT)
|
||||
monkeypatch.setattr(settings, "owner_name", OWNER)
|
||||
monkeypatch.setattr(authmod, "_jwks_client", _StubJWKS(public_key))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def mem_db(monkeypatch):
|
||||
engine = create_async_engine(
|
||||
"sqlite+aiosqlite:///:memory:",
|
||||
poolclass=StaticPool,
|
||||
connect_args={"check_same_thread": False},
|
||||
)
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
factory = async_sessionmaker(engine, expire_on_commit=False)
|
||||
monkeypatch.setattr(dbmod, "_engine", engine)
|
||||
monkeypatch.setattr(dbmod, "_session_factory", factory)
|
||||
yield factory
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client():
|
||||
transport = httpx.ASGITransport(app=main.app)
|
||||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
||||
yield c
|
||||
|
||||
|
||||
async def _seed_user(factory, *, name, casdoor_sub=None, email=None) -> str:
|
||||
uid = uuid.uuid4().hex
|
||||
async with factory() as session:
|
||||
session.add(
|
||||
User(id=uid, name=name, display_name=name, email=email, casdoor_sub=casdoor_sub)
|
||||
)
|
||||
await session.commit()
|
||||
return uid
|
||||
|
||||
|
||||
async def _seed_pat(factory, user_id, *, revoked=False, expires_at=None) -> str:
|
||||
from datetime import UTC, datetime
|
||||
|
||||
plaintext = authmod.PAT_PREFIX + uuid.uuid4().hex
|
||||
async with factory() as session:
|
||||
pat = PersonalAccessToken(
|
||||
id=uuid.uuid4().hex,
|
||||
user_id=user_id,
|
||||
name="test",
|
||||
token_hash=authmod.hash_token(plaintext),
|
||||
token_prefix=plaintext[: len(authmod.PAT_PREFIX) + 4],
|
||||
revoked_at=(datetime.now(UTC) if revoked else None),
|
||||
expires_at=expires_at,
|
||||
)
|
||||
session.add(pat)
|
||||
await session.commit()
|
||||
return plaintext
|
||||
|
||||
|
||||
PROTECTED = "/api/v1/calls/active"
|
||||
|
||||
|
||||
# ── JWT paths ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestCasdoorJWT:
|
||||
async def test_owner_jwt_reaches_handler(self, sso_enabled, mem_db, client, keypair):
|
||||
private_pem, _ = keypair
|
||||
token = _mint(private_pem, sub="s-owner", name=OWNER, email=OWNER)
|
||||
resp = await client.get(PROTECTED, headers={"Authorization": f"Bearer {token}"})
|
||||
assert resp.status_code == 503 # auth accepted; no lifespan → handler 503s
|
||||
|
||||
async def test_non_owner_jwt_forbidden(self, sso_enabled, mem_db, client, keypair):
|
||||
private_pem, _ = keypair
|
||||
token = _mint(private_pem, sub="s-guest", name="guest@example.test")
|
||||
resp = await client.get(PROTECTED, headers={"Authorization": f"Bearer {token}"})
|
||||
assert resp.status_code == 403
|
||||
|
||||
async def test_expired_jwt_unauthenticated(self, sso_enabled, mem_db, client, keypair):
|
||||
private_pem, _ = keypair
|
||||
token = _mint(private_pem, sub="s-owner", name=OWNER, exp_delta=-10)
|
||||
resp = await client.get(PROTECTED, headers={"Authorization": f"Bearer {token}"})
|
||||
assert resp.status_code == 401
|
||||
|
||||
async def test_garbage_token_unauthenticated(self, sso_enabled, mem_db, client):
|
||||
resp = await client.get(PROTECTED, headers={"Authorization": "Bearer not.a.jwt"})
|
||||
assert resp.status_code == 401
|
||||
|
||||
async def test_no_credentials_unauthenticated(self, sso_enabled, mem_db, client):
|
||||
resp = await client.get(PROTECTED)
|
||||
assert resp.status_code == 401
|
||||
assert resp.headers["www-authenticate"] == "Bearer"
|
||||
|
||||
|
||||
# ── PAT paths ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestPAT:
|
||||
async def test_owner_pat_reaches_handler(self, sso_enabled, mem_db, client):
|
||||
uid = await _seed_user(mem_db, name=OWNER, casdoor_sub="s-owner", email=OWNER)
|
||||
pat = await _seed_pat(mem_db, uid)
|
||||
resp = await client.get(PROTECTED, headers={"Authorization": f"Bearer {pat}"})
|
||||
assert resp.status_code == 503
|
||||
|
||||
async def test_non_owner_pat_forbidden(self, sso_enabled, mem_db, client):
|
||||
uid = await _seed_user(mem_db, name="guest@example.test", casdoor_sub="s-guest")
|
||||
pat = await _seed_pat(mem_db, uid)
|
||||
resp = await client.get(PROTECTED, headers={"Authorization": f"Bearer {pat}"})
|
||||
assert resp.status_code == 403
|
||||
|
||||
async def test_revoked_pat_unauthenticated(self, sso_enabled, mem_db, client):
|
||||
uid = await _seed_user(mem_db, name=OWNER, casdoor_sub="s-owner", email=OWNER)
|
||||
pat = await _seed_pat(mem_db, uid, revoked=True)
|
||||
resp = await client.get(PROTECTED, headers={"Authorization": f"Bearer {pat}"})
|
||||
assert resp.status_code == 401
|
||||
|
||||
async def test_expired_pat_unauthenticated(self, sso_enabled, mem_db, client):
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
uid = await _seed_user(mem_db, name=OWNER, casdoor_sub="s-owner", email=OWNER)
|
||||
pat = await _seed_pat(mem_db, uid, expires_at=datetime.now(UTC) - timedelta(minutes=1))
|
||||
resp = await client.get(PROTECTED, headers={"Authorization": f"Bearer {pat}"})
|
||||
assert resp.status_code == 401
|
||||
|
||||
async def test_unknown_pat_unauthenticated(self, sso_enabled, mem_db, client):
|
||||
resp = await client.get(
|
||||
PROTECTED, headers={"Authorization": f"Bearer {authmod.PAT_PREFIX}nope"}
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
# ── Dev-owner mode (SSO disabled) ────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestDevOwnerMode:
|
||||
async def test_tokenless_request_is_owner(self, monkeypatch, mem_db, client):
|
||||
monkeypatch.setattr(get_settings().casdoor, "enabled", False)
|
||||
resp = await client.get(PROTECTED)
|
||||
assert resp.status_code == 503 # dev-owner resolved; handler 503s (no lifespan)
|
||||
|
||||
async def test_auth_me_reports_owner(self, monkeypatch, mem_db, client):
|
||||
monkeypatch.setattr(get_settings().casdoor, "enabled", False)
|
||||
resp = await client.get("/auth/me")
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["is_owner"] is True
|
||||
|
||||
|
||||
# ── /auth/me for a non-owner (200 + is_owner:false, not a hard 401) ──────────
|
||||
|
||||
|
||||
class TestAuthMe:
|
||||
async def test_non_owner_gets_200_not_owner(self, sso_enabled, mem_db, client, keypair):
|
||||
private_pem, _ = keypair
|
||||
token = _mint(private_pem, sub="s-guest", name="guest@example.test")
|
||||
resp = await client.get("/auth/me", headers={"Authorization": f"Bearer {token}"})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["is_owner"] is False
|
||||
Reference in New Issue
Block a user