83 lines
2.3 KiB
Python
83 lines
2.3 KiB
Python
import secrets
|
|
import time
|
|
from dataclasses import dataclass
|
|
from datetime import timedelta
|
|
from enum import Enum
|
|
from typing import Callable
|
|
|
|
TTL_SECONDS = 120.0
|
|
MAX_PENDING = 100
|
|
TOKEN_BYTES = 6
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DowntimeRequest:
|
|
backend: str
|
|
sender_id: str
|
|
host: str
|
|
service: str
|
|
duration: timedelta
|
|
comment: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Pending:
|
|
token: str
|
|
request: DowntimeRequest
|
|
expires_at: float
|
|
|
|
|
|
class Outcome(Enum):
|
|
CONFIRMED = "confirmed"
|
|
CANCELLED = "cancelled"
|
|
EXPIRED = "expired"
|
|
UNKNOWN = "unknown"
|
|
WRONG_SENDER = "wrong_sender"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Resolution:
|
|
outcome: Outcome
|
|
request: DowntimeRequest | None = None
|
|
|
|
|
|
class PendingStore:
|
|
def __init__(self, clock: Callable[[], float] = time.monotonic, ttl: float = TTL_SECONDS,
|
|
max_pending: int = MAX_PENDING):
|
|
self._clock = clock
|
|
self._ttl = ttl
|
|
self._max = max_pending
|
|
self._items: dict[str, Pending] = {}
|
|
|
|
@property
|
|
def ttl(self) -> float:
|
|
return self._ttl
|
|
|
|
def create(self, request: DowntimeRequest) -> Pending:
|
|
while len(self._items) >= self._max:
|
|
self._items.pop(next(iter(self._items)))
|
|
token = secrets.token_urlsafe(TOKEN_BYTES)
|
|
while token in self._items:
|
|
token = secrets.token_urlsafe(TOKEN_BYTES)
|
|
pending = Pending(token, request, self._clock() + self._ttl)
|
|
self._items[token] = pending
|
|
return pending
|
|
|
|
def resolve(self, token: str, backend: str, sender_id: str, approve: bool) -> Resolution:
|
|
pending = self._items.get(token)
|
|
if pending is None:
|
|
return Resolution(Outcome.UNKNOWN)
|
|
request = pending.request
|
|
if pending.expires_at <= self._clock():
|
|
del self._items[token]
|
|
return Resolution(Outcome.EXPIRED, request)
|
|
if (request.backend, request.sender_id) != (backend, sender_id):
|
|
return Resolution(Outcome.WRONG_SENDER, request)
|
|
del self._items[token]
|
|
return Resolution(Outcome.CONFIRMED if approve else Outcome.CANCELLED, request)
|
|
|
|
def expire(self) -> list[DowntimeRequest]:
|
|
now = self._clock()
|
|
gone = [token for token, item in self._items.items() if item.expires_at <= now]
|
|
return [self._items.pop(token).request for token in gone]
|