74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
"""Per-user (non-singleton) delete confirmation token store."""
|
|
from __future__ import annotations
|
|
import secrets
|
|
import threading
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import Optional
|
|
from ..config import settings
|
|
|
|
|
|
@dataclass
|
|
class PendingDelete:
|
|
token: str
|
|
instance: str
|
|
model: str
|
|
record_id: int
|
|
record_name: str
|
|
expires_at: datetime
|
|
|
|
|
|
class ConfirmationManager:
|
|
"""In-memory store for delete confirmation tokens. Single-use, TTL-limited.
|
|
One instance per user — no shared state between tenants.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._lock = threading.Lock()
|
|
self._pending: dict[str, PendingDelete] = {}
|
|
|
|
def _purge_expired(self):
|
|
now = datetime.now(timezone.utc)
|
|
expired = [t for t, p in self._pending.items() if p.expires_at <= now]
|
|
for t in expired:
|
|
del self._pending[t]
|
|
|
|
def create_token(self, instance: str, model: str, record_id: int, record_name: str) -> PendingDelete:
|
|
token = f"del_{secrets.token_hex(8)}"
|
|
ttl = settings.confirm_token_ttl
|
|
pending = PendingDelete(
|
|
token=token, instance=instance, model=model,
|
|
record_id=record_id, record_name=record_name,
|
|
expires_at=datetime.now(timezone.utc) + timedelta(seconds=ttl),
|
|
)
|
|
with self._lock:
|
|
self._purge_expired()
|
|
self._pending[token] = pending
|
|
return pending
|
|
|
|
def consume_token(self, token: str, model: str, record_id: int) -> tuple[bool, str]:
|
|
with self._lock:
|
|
self._purge_expired()
|
|
pending = self._pending.get(token)
|
|
if not pending:
|
|
return False, "Invalid or expired confirmation token. Call request_delete_confirmation again."
|
|
if pending.model != model or pending.record_id != record_id:
|
|
return False, "Token does not match the requested model/record."
|
|
del self._pending[token]
|
|
return True, ""
|
|
|
|
def list_pending(self) -> list[dict]:
|
|
with self._lock:
|
|
self._purge_expired()
|
|
now = datetime.now(timezone.utc)
|
|
return [
|
|
{
|
|
"token": p.token,
|
|
"model": p.model,
|
|
"record_id": p.record_id,
|
|
"record_name": p.record_name,
|
|
"expires_in_seconds": int((p.expires_at - now).total_seconds()),
|
|
}
|
|
for p in self._pending.values()
|
|
]
|