Senad Uka
e855650f09
- FastAPI skeleton, SQLAlchemy models (§13), Alembic initial migration - SchedulingProvider interface with google_calendar (free/busy read-only), partner_api (Appendix B client) and mock implementations - Proposal engine: create → provider-routed delivery → owner actions (resolve/confirm+SMS/reject) → expiry + reminders (§9) - Signed single-use action links, .ics METHOD:REQUEST attachment - Partner outcome webhook with HMAC verification + polling fallback - SmsProvider (console) with Bosnian templates (§5.5), EmailProvider (console/SMTP) - Fake partner API server in tests/ — Appendix B reference implementation - 43 tests: slot math, proposal lifecycle, action links, partner contract Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
45 lines
1.3 KiB
Python
45 lines
1.3 KiB
Python
"""Single-use signed action tokens for proposal email links (§9.1).
|
|
|
|
Tokens are random, stored in DB (single-use enforced there) and the URL carries
|
|
an itsdangerous signature so guessing/forging is infeasible even if the DB row
|
|
leaked. Action links must be idempotent (§15): a used 'resolve' link re-visited
|
|
shows "already resolved", it never errors or double-fires.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
import uuid
|
|
|
|
from itsdangerous import BadSignature, URLSafeSerializer
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from gogo.config import get_settings
|
|
from gogo.models import ActionToken
|
|
|
|
|
|
def _serializer() -> URLSafeSerializer:
|
|
return URLSafeSerializer(get_settings().secret_key, salt="proposal-action")
|
|
|
|
|
|
async def create_action_token(
|
|
session: AsyncSession, request_id: uuid.UUID, action: str, slot_index: int | None = None
|
|
) -> str:
|
|
"""Create a DB-backed token and return the signed URL-safe token string."""
|
|
raw = secrets.token_urlsafe(24)[:40]
|
|
session.add(
|
|
ActionToken(token=raw, request_id=request_id, action=action, slot_index=slot_index)
|
|
)
|
|
return _serializer().dumps(raw)
|
|
|
|
|
|
def action_url(signed: str) -> str:
|
|
return f"{get_settings().base_url}/a/{signed}"
|
|
|
|
|
|
def unsign(signed: str) -> str | None:
|
|
try:
|
|
return _serializer().loads(signed)
|
|
except BadSignature:
|
|
return None
|