115 lines
4.2 KiB
Python
115 lines
4.2 KiB
Python
import asyncio
|
|
from types import SimpleNamespace
|
|
from urllib.parse import quote, urlsplit
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from pwdlib import PasswordHash
|
|
from sqlalchemy import delete
|
|
|
|
from ai_platform.bootstrap.app import create_app
|
|
from ai_platform.bootstrap.wiring import build_app_credential_use_cases
|
|
from ai_platform.infrastructure.config.settings import Settings
|
|
from ai_platform.infrastructure.persistence.database import make_session_factory
|
|
from ai_platform.infrastructure.persistence.models import (
|
|
AdminSession,
|
|
AdminUser,
|
|
AppCredential,
|
|
)
|
|
from ai_platform.platform.security import digest
|
|
|
|
|
|
@pytest.fixture
|
|
def env():
|
|
settings = Settings.load()
|
|
if urlsplit(settings.database_url).path.rstrip("/").rsplit("/", 1)[-1] != "ai_platform_test":
|
|
raise RuntimeError("App API tests require /ai_platform_test")
|
|
sessions = make_session_factory(settings.database_url)
|
|
app_id = f"omni/{uuid4().hex}"
|
|
email, password = f"app-admin-{uuid4().hex}@example.invalid", "test-only-password"
|
|
|
|
async def run(action):
|
|
async with sessions() as db:
|
|
result = await action(db)
|
|
await sessions.kw["bind"].dispose()
|
|
return result
|
|
|
|
async def setup(db):
|
|
db.add(AdminUser(email=email, password_hash=PasswordHash.recommended().hash(password)))
|
|
await db.commit()
|
|
return await build_app_credential_use_cases(db).create(app_id)
|
|
|
|
async def cleanup(db):
|
|
await db.execute(delete(AppCredential).where(AppCredential.app_id == app_id))
|
|
await db.execute(delete(AdminSession).where(AdminSession.admin_email == email))
|
|
await db.execute(delete(AdminUser).where(AdminUser.email == email))
|
|
await db.commit()
|
|
|
|
async def rotate(db):
|
|
return await build_app_credential_use_cases(db).rotate(app_id)
|
|
|
|
async def stored_verifier(db):
|
|
row = await db.get(AppCredential, app_id)
|
|
return row.verifier if row and row.active else None
|
|
|
|
initial_key = asyncio.run(run(setup))
|
|
try:
|
|
with TestClient(create_app(settings)) as client:
|
|
yield SimpleNamespace(
|
|
client=client,
|
|
email=email,
|
|
password=password,
|
|
path=f"/v1/admin/apps/{quote(app_id, safe='')}",
|
|
app_id=app_id,
|
|
initial_key=initial_key,
|
|
rotate=lambda: asyncio.run(run(rotate)),
|
|
verifier=lambda: asyncio.run(run(stored_verifier)),
|
|
)
|
|
finally:
|
|
asyncio.run(run(cleanup))
|
|
|
|
|
|
def _login(env) -> dict:
|
|
login = env.client.post("/v1/admin/login", json={"email": env.email, "password": env.password})
|
|
assert login.status_code == 200
|
|
return {"x-csrf-token": login.json()["csrf_token"]}
|
|
|
|
|
|
def _probe(env, key: str) -> int:
|
|
# An empty body is 422 once authenticated, 401 before.
|
|
headers = {"Authorization": f"Bearer {key}"}
|
|
return env.client.post("/v1/lead-scoring/score", headers=headers, json={}).status_code
|
|
|
|
|
|
def test_admin_apps_require_a_session_and_never_expose_keys(env):
|
|
client = env.client
|
|
assert client.get("/v1/admin/apps").status_code == 401
|
|
assert client.delete(env.path).status_code == 401
|
|
assert client.post("/v1/admin/apps", json={"app_id": "new"}).status_code == 405
|
|
_login(env)
|
|
|
|
listed = client.get("/v1/admin/apps")
|
|
item = next(x for x in listed.json()["apps"] if x["app_id"] == env.app_id)
|
|
assert set(item) == {"app_id", "active", "created_at", "updated_at"} and item["active"]
|
|
assert env.initial_key not in listed.text and "verifier" not in listed.text
|
|
assert client.delete(env.path).status_code == 403
|
|
|
|
|
|
def test_rotate_and_revoke_change_which_key_authenticates(env):
|
|
csrf = _login(env)
|
|
assert _probe(env, env.initial_key) == 422
|
|
assert env.verifier() == digest(env.initial_key) != env.initial_key
|
|
|
|
rotated = env.rotate()
|
|
assert (_probe(env, env.initial_key), _probe(env, rotated)) == (401, 422)
|
|
|
|
revoked = env.client.delete(env.path, headers=csrf)
|
|
assert revoked.status_code == 200 and revoked.json()["active"] is False
|
|
assert _probe(env, rotated) == 401
|
|
|
|
active = env.rotate()
|
|
assert _probe(env, active) == 422
|
|
assert env.verifier() == digest(active)
|
|
assert active not in env.client.get("/v1/admin/apps").text
|