Spaces:
Sleeping
Sleeping
File size: 3,423 Bytes
50094a0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | """Signed, expiring tokens for the web dashboard.
The Space is public, so the dashboard cannot simply trust a user id in a query
string -- anyone could type someone else's. Instead the bot issues a token in
Telegram (where identity is already established) and the page verifies it.
Signed with ``TELEGRAM_WEBHOOK_SECRET``, so there is no additional secret to
manage or leak. Tokens are short-lived because a URL leaks easily: browser
history, referrer headers, a screenshot in a chat.
Nothing here can mutate the ledger. A stolen token, within its window, exposes
one person's spending summary and coach chat -- bad, but bounded, and the
window is minutes.
"""
from __future__ import annotations
import base64
import hashlib
import hmac
import time
from typing import NamedTuple, Optional
#: Long enough to open the link and read it, short enough that a leaked URL in
#: browser history is stale by the time anyone finds it.
DEFAULT_TTL_SECONDS = 30 * 60
_SIG_BYTES = 16
class TokenError(ValueError):
"""The token was malformed, tampered with, or expired."""
class Claims(NamedTuple):
user_id: int
expires_at: int
@property
def seconds_left(self) -> int:
return max(0, self.expires_at - int(time.time()))
def _b64encode(raw: bytes) -> str:
return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
def _b64decode(text: str) -> bytes:
padding = "=" * (-len(text) % 4)
return base64.urlsafe_b64decode(text + padding)
def _sign(payload: bytes, secret: str) -> bytes:
return hmac.new(secret.encode("utf-8"), payload, hashlib.sha256).digest()[
:_SIG_BYTES
]
def issue(
user_id: int,
secret: str,
ttl_seconds: int = DEFAULT_TTL_SECONDS,
now: Optional[int] = None,
) -> str:
"""Mint a dashboard token for ``user_id``."""
if not secret:
raise TokenError("cannot issue tokens without a signing secret")
now = now if now is not None else int(time.time())
payload = f"{int(user_id)}.{now + int(ttl_seconds)}".encode("utf-8")
return f"{_b64encode(payload)}.{_b64encode(_sign(payload, secret))}"
def verify(token: str, secret: str, now: Optional[int] = None) -> Claims:
"""Validate a token and return its claims, or raise :class:`TokenError`."""
if not secret:
raise TokenError("no signing secret configured")
if not token or "." not in token:
raise TokenError("malformed token")
encoded_payload, _, encoded_sig = token.rpartition(".")
try:
payload = _b64decode(encoded_payload)
signature = _b64decode(encoded_sig)
except Exception as exc:
raise TokenError("malformed token") from exc
# Constant-time: a timing-variable comparison would leak the signature
# byte by byte to anyone willing to measure.
if not hmac.compare_digest(signature, _sign(payload, secret)):
raise TokenError("bad signature")
try:
user_part, _, expiry_part = payload.decode("utf-8").partition(".")
claims = Claims(user_id=int(user_part), expires_at=int(expiry_part))
except Exception as exc:
raise TokenError("malformed payload") from exc
now = now if now is not None else int(time.time())
if claims.expires_at < now:
raise TokenError("token expired")
return claims
def dashboard_url(base_url: str, token: str) -> str:
base = (base_url or "").rstrip("/")
return f"{base}/?t={token}"
|