52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
import secrets
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from fastapi_oidc_op.models import MagicLink
|
|
from fastapi_oidc_op.store.protocols import MagicLinkRepository
|
|
|
|
|
|
class MagicLinkService:
|
|
"""Magic link token lifecycle management."""
|
|
|
|
def __init__(self, repo: MagicLinkRepository, ttl: int = 86400) -> None:
|
|
self._repo = repo
|
|
self._ttl = ttl
|
|
|
|
async def create(
|
|
self,
|
|
username: str,
|
|
created_by: str | None = None,
|
|
note: str | None = None,
|
|
) -> MagicLink:
|
|
"""Generate and store a new magic link for the given username."""
|
|
token = secrets.token_urlsafe(32)
|
|
expires_at = datetime.now(UTC) + timedelta(seconds=self._ttl)
|
|
link = MagicLink(
|
|
token=token,
|
|
username=username,
|
|
expires_at=expires_at,
|
|
created_by=created_by,
|
|
note=note,
|
|
)
|
|
return await self._repo.create(link)
|
|
|
|
async def validate(self, token: str) -> MagicLink | None:
|
|
"""Validate a magic link token.
|
|
|
|
Returns the MagicLink if it exists, is not used, and has not expired.
|
|
Returns None otherwise.
|
|
"""
|
|
link = await self._repo.get_by_token(token)
|
|
if link is None or link.used:
|
|
return None
|
|
if link.expires_at < datetime.now(UTC):
|
|
return None
|
|
return link
|
|
|
|
async def mark_used(self, token: str) -> bool:
|
|
"""Mark a magic link as used. Returns True if found and marked."""
|
|
return await self._repo.mark_used(token)
|
|
|
|
async def cleanup_expired(self) -> int:
|
|
"""Delete expired unused links. Returns count deleted."""
|
|
return await self._repo.delete_expired()
|