Rolf
8601584099
- Playing handicap: computed at game creation (WHS formula), stored on game_players alongside per-hole strokes allocation; displayed on game summary with HC card and / markers on scorecard (only when SI defined); live game shows +N per player card when SI exists, HC total otherwise; new game confirmation previews computed HC per player - Login: 6-digit code sent alongside magic link in every auth email; check_email page has auto-advancing digit inputs with paste support; new POST /auth/verify-code route handles code verification - Game view: Score/Card toggle fixed with position:fixed to always show above nav bar on mobile; swipe on score view changes holes, swipe on scorecard switches tabs; dirty tracking prevents saving unmodified holes Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
116 lines
3.2 KiB
Python
116 lines
3.2 KiB
Python
import os
|
|
import secrets
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import aiosqlite
|
|
|
|
from app.email import send_magic_link
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SITE_URL = os.getenv("SITE_URL", "http://localhost:8000")
|
|
ENV = os.getenv("ENV", "development")
|
|
TOKEN_TTL_MINUTES = 15
|
|
SESSION_TTL_DAYS = 30
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def _in(minutes: int = 0, days: int = 0) -> str:
|
|
delta = timedelta(minutes=minutes, days=days)
|
|
return (datetime.now(timezone.utc) + delta).isoformat()
|
|
|
|
|
|
def _expired(iso: str) -> bool:
|
|
return datetime.now(timezone.utc) > datetime.fromisoformat(iso)
|
|
|
|
|
|
async def create_magic_token(email: str, db: aiosqlite.Connection) -> str:
|
|
# Clean up old unused tokens for this email
|
|
await db.execute(
|
|
"DELETE FROM magic_link_tokens WHERE email = ? AND used_at IS NULL",
|
|
(email,),
|
|
)
|
|
|
|
token_id = secrets.token_urlsafe(16)
|
|
token = secrets.token_urlsafe(32)
|
|
code = f"{secrets.randbelow(1000000):06d}"
|
|
expires_at = _in(minutes=TOKEN_TTL_MINUTES)
|
|
|
|
await db.execute(
|
|
"INSERT INTO magic_link_tokens (id, email, token, code, expires_at) VALUES (?, ?, ?, ?, ?)",
|
|
(token_id, email, token, code, expires_at),
|
|
)
|
|
await db.commit()
|
|
|
|
link = f"{SITE_URL}/auth/verify?token={token}"
|
|
await send_magic_link(email, link, code)
|
|
return token
|
|
|
|
|
|
async def verify_magic_code(email: str, code: str, db: aiosqlite.Connection):
|
|
"""Returns the email if code matches a valid unused token, else None."""
|
|
async with db.execute(
|
|
"SELECT expires_at, used_at FROM magic_link_tokens WHERE email = ? AND code = ? AND used_at IS NULL",
|
|
(email, code),
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
|
|
if row is None:
|
|
return None
|
|
if _expired(row["expires_at"]):
|
|
return None
|
|
|
|
await db.execute(
|
|
"UPDATE magic_link_tokens SET used_at = ? WHERE email = ? AND code = ?",
|
|
(_now(), email, code),
|
|
)
|
|
await db.commit()
|
|
return email
|
|
|
|
|
|
async def verify_magic_token(token: str, db: aiosqlite.Connection):
|
|
"""Returns the email if token is valid and unused, else None."""
|
|
async with db.execute(
|
|
"SELECT email, expires_at, used_at FROM magic_link_tokens WHERE token = ?",
|
|
(token,),
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
|
|
if row is None:
|
|
return None
|
|
if row["used_at"] is not None:
|
|
return None
|
|
if _expired(row["expires_at"]):
|
|
return None
|
|
|
|
# Mark token as used
|
|
await db.execute(
|
|
"UPDATE magic_link_tokens SET used_at = ? WHERE token = ?",
|
|
(_now(), token),
|
|
)
|
|
await db.commit()
|
|
|
|
return row["email"]
|
|
|
|
|
|
async def create_session(user_id: str, db: aiosqlite.Connection) -> str:
|
|
session_id = secrets.token_urlsafe(16)
|
|
token = secrets.token_urlsafe(32)
|
|
expires_at = _in(days=SESSION_TTL_DAYS)
|
|
|
|
await db.execute(
|
|
"INSERT INTO sessions (id, user_id, token, expires_at, created_at) VALUES (?, ?, ?, ?, ?)",
|
|
(session_id, user_id, token, expires_at, _now()),
|
|
)
|
|
await db.commit()
|
|
return token
|
|
|
|
|
|
async def delete_session(token: str, db: aiosqlite.Connection):
|
|
await db.execute("DELETE FROM sessions WHERE token = ?", (token,))
|
|
await db.commit()
|