79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
"""Single-use invitation tokens for employee onboarding."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models import Account, MemberInvitation
|
|
|
|
INVITE_EXPIRY_HOURS = 48
|
|
INVITE_FROM_LOCAL = "invites"
|
|
|
|
|
|
def hash_invite_token(raw_token: str) -> str:
|
|
return hashlib.sha256(raw_token.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def generate_invite_token() -> str:
|
|
return secrets.token_urlsafe(32)
|
|
|
|
|
|
async def invalidate_pending_invitations(
|
|
session: AsyncSession,
|
|
account_id,
|
|
) -> None:
|
|
invitations = await session.scalars(
|
|
select(MemberInvitation).where(
|
|
MemberInvitation.account_id == account_id,
|
|
MemberInvitation.used_at.is_(None),
|
|
)
|
|
)
|
|
now = datetime.now(timezone.utc)
|
|
for invitation in invitations.all():
|
|
invitation.used_at = now
|
|
|
|
|
|
async def create_member_invitation(
|
|
session: AsyncSession,
|
|
member: Account,
|
|
) -> str:
|
|
await invalidate_pending_invitations(session, member.id)
|
|
raw_token = generate_invite_token()
|
|
invitation = MemberInvitation(
|
|
account_id=member.id,
|
|
token_hash=hash_invite_token(raw_token),
|
|
expires_at=datetime.now(timezone.utc)
|
|
+ timedelta(hours=INVITE_EXPIRY_HOURS),
|
|
)
|
|
session.add(invitation)
|
|
await session.flush()
|
|
return raw_token
|
|
|
|
|
|
async def get_valid_invitation(
|
|
session: AsyncSession,
|
|
raw_token: str,
|
|
) -> tuple[MemberInvitation, Account]:
|
|
token_hash = hash_invite_token(raw_token.strip())
|
|
invitation = await session.scalar(
|
|
select(MemberInvitation)
|
|
.options(selectinload(MemberInvitation.account))
|
|
.where(MemberInvitation.token_hash == token_hash)
|
|
)
|
|
if invitation is None:
|
|
raise ValueError("This invitation link is invalid.")
|
|
if invitation.used_at is not None:
|
|
raise ValueError("This invitation has already been used.")
|
|
now = datetime.now(timezone.utc)
|
|
if invitation.expires_at <= now:
|
|
raise ValueError("This invitation has expired.")
|
|
account = invitation.account
|
|
if account is None or account.member_status != "invited":
|
|
raise ValueError("This invitation is no longer valid.")
|
|
return invitation, account
|