91a2277664
Magic-link tokens were persisted in plaintext, so a database read disclosed usable login/invite tokens. The service now hashes tokens (HMAC-SHA256 when a pepper is configured, else SHA-256 of the high-entropy token) and persists only the hash; the raw token is exposed solely in the registration URL and is re-attached to objects returned to callers. Refs: porchlight-42h Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
125 lines
4.4 KiB
Python
125 lines
4.4 KiB
Python
from collections.abc import AsyncIterator
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import aiosqlite
|
|
import pytest
|
|
|
|
from porchlight.invite.service import MagicLinkService
|
|
from porchlight.store.sqlite.migrations import run_migrations
|
|
from porchlight.store.sqlite.repositories import SQLiteMagicLinkRepository
|
|
|
|
MIGRATIONS_DIR = (
|
|
Path(__file__).resolve().parent.parent.parent / "src" / "porchlight" / "store" / "sqlite" / "migrations"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
async def db() -> AsyncIterator[aiosqlite.Connection]:
|
|
conn = await aiosqlite.connect(":memory:")
|
|
conn.row_factory = aiosqlite.Row
|
|
await conn.execute("PRAGMA foreign_keys=ON")
|
|
await run_migrations(conn, MIGRATIONS_DIR)
|
|
yield conn
|
|
await conn.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def repo(db: aiosqlite.Connection) -> SQLiteMagicLinkRepository:
|
|
return SQLiteMagicLinkRepository(db)
|
|
|
|
|
|
@pytest.fixture
|
|
def service(repo: SQLiteMagicLinkRepository) -> MagicLinkService:
|
|
return MagicLinkService(repo=repo, ttl=3600)
|
|
|
|
|
|
async def test_create_returns_magic_link(service: MagicLinkService) -> None:
|
|
link = await service.create(username="alice")
|
|
assert link.username == "alice"
|
|
assert link.token # non-empty
|
|
assert link.used is False
|
|
assert link.expires_at > datetime.now(UTC)
|
|
|
|
|
|
async def test_token_stored_hashed_not_plaintext(service: MagicLinkService, repo: SQLiteMagicLinkRepository) -> None:
|
|
# The raw token handed to the user must never be the value persisted in
|
|
# the store; a DB read must not yield a usable token.
|
|
link = await service.create(username="alice")
|
|
raw = link.token
|
|
assert await repo.get_by_token(raw) is None
|
|
# ...but the raw token must still validate through the service.
|
|
validated = await service.validate(raw)
|
|
assert validated is not None
|
|
assert validated.username == "alice"
|
|
|
|
|
|
async def test_create_generates_unique_tokens(service: MagicLinkService) -> None:
|
|
link1 = await service.create(username="alice")
|
|
link2 = await service.create(username="bob")
|
|
assert link1.token != link2.token
|
|
|
|
|
|
async def test_create_with_optional_fields(service: MagicLinkService) -> None:
|
|
link = await service.create(username="alice", created_by="admin-id", note="Welcome")
|
|
assert link.created_by == "admin-id"
|
|
assert link.note == "Welcome"
|
|
|
|
|
|
async def test_create_respects_ttl(service: MagicLinkService) -> None:
|
|
link = await service.create(username="alice")
|
|
expected_min = datetime.now(UTC) + timedelta(seconds=3500)
|
|
expected_max = datetime.now(UTC) + timedelta(seconds=3700)
|
|
assert expected_min < link.expires_at < expected_max
|
|
|
|
|
|
async def test_validate_valid_token(service: MagicLinkService) -> None:
|
|
link = await service.create(username="alice")
|
|
result = await service.validate(link.token)
|
|
assert result is not None
|
|
assert result.token == link.token
|
|
assert result.username == "alice"
|
|
|
|
|
|
async def test_validate_nonexistent_token(service: MagicLinkService) -> None:
|
|
result = await service.validate("nonexistent-token")
|
|
assert result is None
|
|
|
|
|
|
async def test_validate_used_token(service: MagicLinkService) -> None:
|
|
link = await service.create(username="alice")
|
|
await service.mark_used(link.token)
|
|
result = await service.validate(link.token)
|
|
assert result is None
|
|
|
|
|
|
async def test_validate_expired_token(service: MagicLinkService, repo: SQLiteMagicLinkRepository) -> None:
|
|
# Create a link that's already expired by using a negative TTL
|
|
expired_service = MagicLinkService(repo=repo, ttl=-1)
|
|
link = await expired_service.create(username="alice")
|
|
# The link expires_at is essentially now, so by the time we validate it should be expired
|
|
result = await service.validate(link.token)
|
|
assert result is None
|
|
|
|
|
|
async def test_mark_used_returns_true(service: MagicLinkService) -> None:
|
|
link = await service.create(username="alice")
|
|
result = await service.mark_used(link.token)
|
|
assert result is True
|
|
|
|
|
|
async def test_mark_used_nonexistent_returns_false(service: MagicLinkService) -> None:
|
|
result = await service.mark_used("nonexistent")
|
|
assert result is False
|
|
|
|
|
|
async def test_cleanup_expired(service: MagicLinkService, repo: SQLiteMagicLinkRepository) -> None:
|
|
# Create an expired link
|
|
expired_service = MagicLinkService(repo=repo, ttl=-3600)
|
|
await expired_service.create(username="expired-user")
|
|
|
|
# Create a valid link
|
|
await service.create(username="valid-user")
|
|
|
|
count = await service.cleanup_expired()
|
|
assert count == 1
|