302 lines
9.7 KiB
Python
302 lines
9.7 KiB
Python
"""Integration tests for passkey (WebAuthn) + magic-link auth endpoints.
|
|
|
|
Run against the Docker Compose infrastructure (`docker compose up -d`).
|
|
|
|
Real WebAuthn ceremonies need a hardware/platform authenticator, so the
|
|
finish endpoints are only exercised on their error paths; the happy paths
|
|
are covered by unit tests on core/passkeys.py plus the webauthn library
|
|
itself. Magic-link verify is tested end-to-end by seeding a token hash
|
|
directly (the raw token never touches the DB).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import uuid
|
|
|
|
import httpx
|
|
import pytest
|
|
from sqlalchemy import text
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
_TG_ID = 915_001
|
|
|
|
|
|
def _hash(token: str) -> str:
|
|
return hashlib.sha256(token.encode("utf-8")).hexdigest()
|
|
|
|
|
|
async def _register_email_user(client: httpx.AsyncClient, email: str) -> str:
|
|
"""Register an email/password user via the API and return the access JWT."""
|
|
r = await client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "name": "Passkey Tester", "password": "supersecret123"},
|
|
)
|
|
assert r.status_code == 201, f"register failed: {r.status_code} {r.text}"
|
|
return r.json()["access_token"]
|
|
|
|
|
|
async def test_passkey_register_start_returns_options(
|
|
client: httpx.AsyncClient,
|
|
infra: dict[str, str],
|
|
) -> None:
|
|
from tests.integration.conftest import user_token
|
|
|
|
token = await user_token(client, infra, telegram_id=_TG_ID)
|
|
|
|
r = await client.post(
|
|
"/api/v1/auth/passkeys/register/start",
|
|
json={"device_name": "YubiKey 5"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["rp"]["id"]
|
|
assert body["rp"]["name"]
|
|
assert body["rp"]["origin"].startswith("http")
|
|
assert body["user"]["id"]
|
|
assert body["timeout"] > 0
|
|
assert body["attestation"] == "none"
|
|
assert body["authenticatorSelection"] == {
|
|
"residentKey": "required",
|
|
"userVerification": "required",
|
|
}
|
|
assert all(p["type"] == "public-key" for p in body["pubKeyCredParams"])
|
|
assert body["excludeCredentials"] == []
|
|
assert len(body["challenge"]) >= 16
|
|
|
|
|
|
async def test_passkey_register_start_requires_user_jwt(
|
|
client: httpx.AsyncClient,
|
|
) -> None:
|
|
r = await client.post("/api/v1/auth/passkeys/register/start", json={})
|
|
assert r.status_code == 401
|
|
|
|
|
|
async def test_passkey_register_finish_rejects_garbage_attestation(
|
|
client: httpx.AsyncClient,
|
|
infra: dict[str, str],
|
|
) -> None:
|
|
from tests.integration.conftest import user_token
|
|
|
|
token = await user_token(client, infra, telegram_id=_TG_ID + 1)
|
|
|
|
start = await client.post(
|
|
"/api/v1/auth/passkeys/register/start",
|
|
json={},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert start.status_code == 200
|
|
|
|
r = await client.post(
|
|
"/api/v1/auth/passkeys/register/finish",
|
|
json={
|
|
"credential": {
|
|
"id": "garbage",
|
|
"rawId": "garbage",
|
|
"type": "public-key",
|
|
"response": {"clientDataJSON": "e30", "attestationObject": "e30"},
|
|
}
|
|
},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert r.status_code == 400
|
|
assert "invalid registration" in r.json()["detail"]
|
|
|
|
|
|
async def test_passkey_register_finish_without_challenge_fails(
|
|
client: httpx.AsyncClient,
|
|
infra: dict[str, str],
|
|
) -> None:
|
|
from tests.integration.conftest import user_token
|
|
|
|
token = await user_token(client, infra, telegram_id=_TG_ID + 2)
|
|
r = await client.post(
|
|
"/api/v1/auth/passkeys/register/finish",
|
|
json={
|
|
"credential": {
|
|
"id": "garbage",
|
|
"rawId": "garbage",
|
|
"type": "public-key",
|
|
"response": {"clientDataJSON": "e30", "attestationObject": "e30"},
|
|
}
|
|
},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert r.status_code == 400
|
|
assert "no pending registration challenge" in r.json()["detail"]
|
|
|
|
|
|
async def test_passkey_authenticate_start_returns_options(
|
|
client: httpx.AsyncClient,
|
|
) -> None:
|
|
r = await client.post("/api/v1/auth/passkeys/authenticate/start")
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["rpId"]
|
|
assert body["allowCredentials"] == []
|
|
assert body["userVerification"] == "required"
|
|
assert len(body["challenge"]) >= 16
|
|
|
|
|
|
async def test_passkey_authenticate_finish_rejects_unknown_challenge(
|
|
client: httpx.AsyncClient,
|
|
) -> None:
|
|
r = await client.post(
|
|
"/api/v1/auth/passkeys/authenticate/finish",
|
|
json={
|
|
"challenge": "dGhpcy1pcy1ub3QtYS1yZWFsLWNoYWxsZW5nZQ",
|
|
"credential": {
|
|
"id": "unknown",
|
|
"rawId": "unknown",
|
|
"type": "public-key",
|
|
"response": {
|
|
"clientDataJSON": "e30",
|
|
"authenticatorData": "e30",
|
|
"signature": "e30",
|
|
},
|
|
},
|
|
},
|
|
)
|
|
assert r.status_code == 400
|
|
assert "unknown or expired challenge" in r.json()["detail"]
|
|
|
|
|
|
async def test_passkey_list_and_delete(
|
|
client: httpx.AsyncClient,
|
|
infra: dict[str, str],
|
|
db_session,
|
|
) -> None:
|
|
from tests.integration.conftest import user_token
|
|
|
|
token = await user_token(client, infra, telegram_id=_TG_ID + 3)
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
# Fetch the user id behind the JWT.
|
|
me = await client.get("/api/v1/auth/me", headers=headers)
|
|
assert me.status_code == 200
|
|
user_id = me.json()["sub"]
|
|
|
|
empty = await client.get("/api/v1/auth/passkeys", headers=headers)
|
|
assert empty.status_code == 200
|
|
assert empty.json() == []
|
|
|
|
# Seed one credential row directly (full attestation needs an authenticator).
|
|
cred_id = "itestcred" + uuid.uuid4().hex
|
|
result = await db_session.execute(
|
|
text(
|
|
"INSERT INTO passkey_credentials "
|
|
"(user_id, credential_id, public_key, sign_count, device_name) "
|
|
"VALUES (:u, :c, :k, 0, 'Integration key') RETURNING id"
|
|
),
|
|
{"u": user_id, "c": cred_id, "k": "itestpubkey"},
|
|
)
|
|
row = result.first()
|
|
assert row is not None
|
|
await db_session.commit()
|
|
passkey_row_id = str(row[0])
|
|
|
|
listed = await client.get("/api/v1/auth/passkeys", headers=headers)
|
|
assert listed.status_code == 200
|
|
items = listed.json()
|
|
assert len(items) == 1
|
|
assert items[0]["credentialID"] == cred_id
|
|
assert items[0]["deviceName"] == "Integration key"
|
|
assert items[0]["counter"] == 0
|
|
assert items[0]["userId"] == user_id
|
|
|
|
deleted = await client.delete(f"/api/v1/auth/passkeys/{passkey_row_id}", headers=headers)
|
|
assert deleted.status_code == 200
|
|
assert deleted.json()["ok"] is True
|
|
|
|
gone = await client.delete(f"/api/v1/auth/passkeys/{passkey_row_id}", headers=headers)
|
|
assert gone.status_code == 404
|
|
|
|
final = await client.get("/api/v1/auth/passkeys", headers=headers)
|
|
assert final.json() == []
|
|
|
|
|
|
async def test_magic_link_request_is_generic_for_unknown_email(
|
|
client: httpx.AsyncClient,
|
|
) -> None:
|
|
r = await client.post(
|
|
"/api/v1/auth/magic-link/request",
|
|
json={"email": "nobody-here@example.com"},
|
|
)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["success"] is True
|
|
assert body["message"]
|
|
assert body["expires_in"] > 0
|
|
|
|
|
|
async def test_magic_link_verify_roundtrip(
|
|
client: httpx.AsyncClient,
|
|
db_session,
|
|
) -> None:
|
|
email = f"magic-{uuid.uuid4().hex[:8]}@example.com"
|
|
token = await _register_email_user(client, email)
|
|
|
|
me = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"})
|
|
user_id = me.json()["sub"]
|
|
|
|
# Seed a magic-link token the way magic-link/request would (raw token
|
|
# never touches the DB — only its SHA-256 hash).
|
|
raw = "itest-magic-token-" + uuid.uuid4().hex
|
|
await db_session.execute(
|
|
text(
|
|
"UPDATE users SET magic_link_token_hash = :h, "
|
|
"magic_link_expires_at = now() + interval '15 minutes' "
|
|
"WHERE id = :u"
|
|
),
|
|
{"h": _hash(raw), "u": user_id},
|
|
)
|
|
await db_session.commit()
|
|
|
|
r = await client.post("/api/v1/auth/magic-link/verify", json={"token": raw})
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["token_type"] == "bearer"
|
|
assert body["expires_in"] > 0
|
|
assert body["user_id"] == user_id
|
|
assert body["email"] == email
|
|
assert body["access_token"]
|
|
|
|
# The issued JWT must introspect as the same user.
|
|
me2 = await client.get(
|
|
"/api/v1/auth/me", headers={"Authorization": f"Bearer {body['access_token']}"}
|
|
)
|
|
assert me2.status_code == 200
|
|
assert me2.json()["sub"] == user_id
|
|
|
|
# The link is one-time: replaying it fails.
|
|
replay = await client.post("/api/v1/auth/magic-link/verify", json={"token": raw})
|
|
assert replay.status_code == 400
|
|
assert "invalid magic-link token" in replay.json()["detail"]
|
|
|
|
|
|
async def test_magic_link_verify_rejects_expired_token(
|
|
client: httpx.AsyncClient,
|
|
db_session,
|
|
) -> None:
|
|
email = f"magic-exp-{uuid.uuid4().hex[:8]}@example.com"
|
|
token = await _register_email_user(client, email)
|
|
|
|
me = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"})
|
|
user_id = me.json()["sub"]
|
|
|
|
raw = "expired-magic-token-" + uuid.uuid4().hex
|
|
await db_session.execute(
|
|
text(
|
|
"UPDATE users SET magic_link_token_hash = :h, "
|
|
"magic_link_expires_at = now() - interval '1 minute' "
|
|
"WHERE id = :u"
|
|
),
|
|
{"h": _hash(raw), "u": user_id},
|
|
)
|
|
await db_session.commit()
|
|
|
|
r = await client.post("/api/v1/auth/magic-link/verify", json={"token": raw})
|
|
assert r.status_code == 400
|
|
assert "expired" in r.json()["detail"]
|