"""Help by DeltaBreen for your server: sign the token that says who is logged in to your site, and answer the calls Help makes to your URL (the side panel, buttons, webhooks), checked against the signature first. No dependencies: Python 3.10 and later, with any framework. The contract is docs/integrations.md and docs/embed.md; docs/signature-vectors.json and docs/identity-vectors.json hold this file to it, as they hold Help's own server. """ from __future__ import annotations import asyncio import base64 import hashlib import hmac import inspect import json import os import re import secrets as _random import threading import time from collections import OrderedDict from dataclasses import dataclass, field from typing import Any, Awaitable, Callable, Mapping, Optional, Protocol, Sequence, Union __version__ = "0.1.0" __all__ = [ "SIGNATURE_HEADER", "HelpError", "sign_identity", "identity_answer", "verify_signature", "handshake_answer", "Answer", "Call", "Store", "MemoryStore", "Handler", ] SIGNATURE_HEADER = "X-DeltaBreen-Signature" _REPLAY_WINDOW_SECS = 300 _MAX_BODY_BYTES = 1024 * 1024 _MAX_ANSWER_BYTES = 256 * 1024 _ACTION_KEPT_SECS = 24 * 3600 _EVENT_KEPT_SECS = 48 * 3600 _BUSY_SECS = 60 _MAX_LIFETIME_SECS = 24 * 3600 # Unicode's control characters (C0, DEL and C1), as Help's server refuses them, and the two line separators. _CONTROL = re.compile("[\x00-\x1f\x7f-\x9f\u2028\u2029]") _FACT_NAME = re.compile(r"^[a-z0-9_]{1,40}$") _SECRET = re.compile(r"^hsk_(live|test)_.") _IDEMPOTENCY_KEY = re.compile(r"^[A-Za-z0-9_-]{16,64}$") class HelpError(ValueError): """Something we refused before anything was signed or answered, by a code that says which.""" def __init__(self, code: str, message: str): super().__init__(message) self.code = code def _b64url(raw: bytes) -> str: return base64.urlsafe_b64encode(raw).rstrip(b"=").decode() def _json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, separators=(",", ":")) def _mac(secret: str, *parts: bytes) -> bytes: return hmac.new(secret.encode(), b"".join(parts), hashlib.sha256).digest() def _secrets_of(given: Union[str, Sequence[Optional[str]], None]) -> list[str]: items = [given] if isinstance(given, str) or given is None else list(given) return [s for s in items if isinstance(s, str) and s] def sign_identity( user_id: Union[str, int], *, email: Optional[str] = None, facts: Optional[Mapping[str, Union[str, int, float, bool]]] = None, secret: Optional[str] = None, kid: Optional[str] = None, expires_in: int = 600, once: bool = True, now: Optional[int] = None, jti: Optional[str] = None, ) -> str: """Sign the token that tells Help who is logged in, on your server and never in the page. The secret is the site's for logged-in users (HELP_IDENTITY_SECRET by default). A token lasts ten minutes and works once; pass once=False for an app's web view, which hands one token over before the page loads. """ secret = secret if secret is not None else os.environ.get("HELP_IDENTITY_SECRET", "") if not isinstance(secret, str) or not _SECRET.match(secret): raise HelpError("bad_secret", "The secret is the one your site shows under Recognise logged-in customers, starting hsk_live_ or hsk_test_.") if isinstance(user_id, bool) or not isinstance(user_id, (str, int)): raise HelpError("bad_user", "The user's id is your own id for them: text or a whole number, 1 to 128 characters.") sub = user_id.strip() if isinstance(user_id, str) else str(user_id) if not sub or len(sub) > 128 or _CONTROL.search(sub): raise HelpError("bad_user", "The user's id is your own id for them: text or a whole number, 1 to 128 characters.") if email is not None and not isinstance(email, str): raise HelpError("bad_email", "The user's email is an address, or nothing.") email = (email or "").strip() if email and (len(email) > 254 or not re.fullmatch(r"[^\s@]+@[^\s@]+", email)): raise HelpError("bad_email", "The user's email is an address, or nothing.") facts = dict(facts or {}) if len(facts) > 10: raise HelpError("bad_facts", "At most ten facts.") for name, value in facts.items(): ok = isinstance(value, bool) or (isinstance(value, (int, float)) and value == value and abs(value) != float("inf")) or (isinstance(value, str) and value.strip() and len(value) <= 100 and not _CONTROL.search(value)) if not isinstance(name, str) or not _FACT_NAME.match(name) or not ok: raise HelpError("bad_facts", f'The fact "{name}" is a name of lowercase letters, digits and _ (up to 40), with text up to 100 characters, a number, or true or false.') if isinstance(expires_in, bool) or not isinstance(expires_in, int) or not 1 <= expires_in <= _MAX_LIFETIME_SECS: raise HelpError("bad_expiry", "A token lasts from a second to a day; ten minutes is plenty.") now = int(time.time()) if now is None else now jti = None if not once else (jti or _b64url(_random.token_bytes(16))) header = '{"alg":"HS256","typ":"JWT"' + (',"kid":' + _json(kid) if kid else "") + "}" claims = ['"sub":' + _json(sub)] if email: claims.append('"email":' + _json(email)) claims += ['"iat":%d' % now, '"exp":%d' % (now + expires_in)] if jti: claims.append('"jti":' + _json(jti)) if facts: claims.append('"user":{' + ",".join(_json(n) + ":" + _json(facts[n]) for n in sorted(facts)) + "}") signing_input = _b64url(header.encode()) + "." + _b64url(("{" + ",".join(claims) + "}").encode()) return signing_input + "." + _b64url(_mac(secret, signing_input.encode())) @dataclass(frozen=True) class Answer: """What to answer: a status, headers and the body's bytes.""" status: int headers: dict[str, str] body: bytes def _answer(status: int, value: Any) -> Answer: return Answer(status, {"Content-Type": "application/json"}, _json(value).encode()) def identity_answer(user_id: Union[str, int, None], **options: Any) -> Answer: """What your identity URL answers: 200 and {"token"} for a logged-in user, 204 for anyone else, never cached.""" if user_id is None: return Answer(204, {"Cache-Control": "no-store"}, b"") token = sign_identity(user_id, **options) return Answer(200, {"Content-Type": "application/json", "Cache-Control": "no-store"}, _json({"token": token}).encode()) def verify_signature(header: Optional[str], raw_body: bytes, secrets: Union[str, Sequence[str]], *, now: Optional[int] = None) -> Optional[str]: """The secret a call to your URL was signed with, of those given, or None: check this before reading the body.""" if not isinstance(header, str): return None now = int(time.time()) if now is None else now parts = [p.strip() for p in header.split(",")] t = next((p[2:] for p in parts if p.startswith("t=")), "") if not t or not t.isascii() or not t.isdigit() or abs(now - int(t)) > _REPLAY_WINDOW_SECS: return None sent = [p[3:] for p in parts if p.startswith("v1=")] body = raw_body.encode() if isinstance(raw_body, str) else bytes(raw_body) for secret in _secrets_of(secrets): want = _mac(secret, t.encode(), b".", body).hex() if any(hmac.compare_digest(v.encode(), want.encode()) for v in sent): return secret return None def handshake_answer(secret: str, challenge: str) -> str: """The answer to Help's connection test: the challenge's HMAC under your secret, never the challenge itself.""" return _mac(secret, challenge.encode()).hex() class Store(Protocol): """Where presses and events are remembered; give one shared by every copy of your server (Redis, a table).""" def claim(self, key: str, ttl: int) -> tuple[str, Optional[str]]: ... # ("new", None) | ("busy", None) | ("done", result) def finish(self, key: str, result: str, ttl: int) -> None: ... def release(self, key: str) -> None: ... class MemoryStore: """Remembers presses and events in this process, which is enough for one worker. With several, pass a shared Store.""" def __init__(self, max_entries: int = 10_000): self._max = max_entries self._held: OrderedDict[str, tuple[float, Optional[str]]] = OrderedDict() self._lock = threading.Lock() def claim(self, key: str, ttl: int) -> tuple[str, Optional[str]]: now = time.monotonic() with self._lock: had = self._held.get(key) if had and had[0] > now: return ("busy", None) if had[1] is None else ("done", had[1]) while len(self._held) >= self._max: self._held.popitem(last=False) self._held[key] = (now + min(ttl, _BUSY_SECS), None) return ("new", None) def finish(self, key: str, result: str, ttl: int) -> None: with self._lock: self._held[key] = (time.monotonic() + ttl, result) def release(self, key: str) -> None: with self._lock: self._held.pop(key, None) @dataclass(frozen=True) class Call: """One call Help makes to your URL about a logged-in user of your site.""" id: str type: str mode: str # "live" or "test": the key the ticket came through product: str sent_at: str user_id: str # the id your own server signed for them ticket: Optional[dict] agent_id: Optional[str] key: Optional[str] = None # a button's key, for an action params: dict = field(default_factory=dict) idempotency_key: Optional[str] = None def _call_of(e: dict) -> Call: action = e.get("action") or {} return Call( id=str(e.get("id", "")), type=str(e.get("type", "")), mode=str(e.get("mode", "")), product=str(e.get("product", "")), sent_at=str(e.get("sent_at", "")), user_id=str((e.get("user") or {}).get("id", "")), ticket=e.get("ticket"), agent_id=(e.get("agent") or {}).get("id"), key=action.get("key"), params=action.get("params") or {}, idempotency_key=action.get("idempotency_key"), ) def _result_of(r: Any) -> dict: get = r.get if isinstance(r, dict) else (lambda k, d=None: getattr(r, k, d)) out: dict = {"ok": get("ok") is True} for k in ("message", "reference"): v = get(k) if isinstance(v, str) and v: out[k] = v return out def _header(headers: Mapping[str, str], name: str) -> Optional[str]: want = name.lower() for k, v in headers.items(): if k.lower() == want or k.upper() == "HTTP_" + want.upper().replace("-", "_"): return v return None class Handler: """Answers Help at your URL: checks the signature, answers the connection test, and calls hydrate for the side panel, action for a button (once per press), and event for a webhook's event (once per event). Secrets default to HELP_CONNECTOR_SECRET and HELP_WEBHOOK_SECRET. mode="live" or "test" answers only calls about tickets that came through that key. hydrate, action and event may be plain functions or coroutines. """ def __init__( self, *, secrets: Union[str, Sequence[str], None] = None, hydrate: Optional[Callable[[Call], Any]] = None, action: Optional[Callable[[Call], Any]] = None, event: Optional[Callable[[dict], Any]] = None, mode: Optional[str] = None, store: Optional[Store] = None, now: Optional[Callable[[], int]] = None, log: Optional[Callable[[str], None]] = None, ): given = secrets if secrets is not None else [os.environ.get("HELP_CONNECTOR_SECRET"), os.environ.get("HELP_WEBHOOK_SECRET")] self._secrets = _secrets_of(given) if not self._secrets: raise HelpError("no_secret", "Give the integration's secret (hcs_…) or the webhook's (whs_…), or set HELP_CONNECTOR_SECRET.") self._hydrate, self._action, self._event, self._mode = hydrate, action, event, mode self._store = store or MemoryStore() self._now = now or (lambda: int(time.time())) self._log = log or (lambda line: print(line)) def handle(self, raw_body: bytes, headers: Mapping[str, str]) -> Answer: """For Flask, Django and any framework that is not async: the raw body and the request's headers.""" return _run(self.handle_async(raw_body, headers)) async def handle_async(self, raw_body: bytes, headers: Mapping[str, str]) -> Answer: """For FastAPI, Starlette and any async framework: the raw body and the request's headers.""" body = raw_body.encode() if isinstance(raw_body, str) else bytes(raw_body) if len(body) > _MAX_BODY_BYTES: return _answer(413, {"error": "too_large"}) secret = verify_signature(_header(headers, SIGNATURE_HEADER), body, self._secrets, now=self._now()) if not secret: return _answer(401, {"error": "bad_signature"}) try: e = json.loads(body) except ValueError: return _answer(400, {"error": "bad_json"}) if not isinstance(e, dict): return _answer(400, {"error": "bad_json"}) kind = e.get("type") other_mode = self._mode is not None and e.get("mode") != self._mode if kind == "handshake": if not isinstance(e.get("challenge"), str): return _answer(400, {"error": "no_challenge"}) return _answer(200, {"challenge_response": handshake_answer(secret, e["challenge"])}) if kind == "hydrate": if not self._hydrate: return _answer(400, {"error": "not_handled"}) if other_mode: return _answer(404, {"error": "not_found"}) shown = await _maybe(self._hydrate(_call_of(e))) if shown is None: return _answer(404, {"error": "not_found"}) if not isinstance(shown, dict): return _answer(500, {"error": "answer_not_an_object"}) return self._fit(200, shown) if kind == "action": if not self._action: return _answer(400, {"error": "not_handled"}) key = (e.get("action") or {}).get("idempotency_key") if not isinstance(key, str) or not _IDEMPOTENCY_KEY.match(key): return _answer(400, {"error": "bad_idempotency_key"}) if other_mode: return _answer(200, {"ok": False, "message": f"This is a {e.get('mode')} ticket; this URL acts for {self._mode} ones only."}) async def press() -> Answer: return self._fit(200, _result_of(await _maybe(self._action(_call_of(e))))) return await self._once("action:" + key, _ACTION_KEPT_SECS, press) if kind == "ping": return _answer(200, {"ok": True}) if isinstance(kind, str) and kind.startswith("ticket."): if not self._event: return _answer(400, {"error": "not_handled"}) if not isinstance(e.get("id"), str) or not e["id"]: return _answer(400, {"error": "no_id"}) async def happen() -> Answer: await _maybe(self._event(e)) return _answer(200, {"ok": True}) return await self._once("event:" + e["id"], _EVENT_KEPT_SECS, happen) return _answer(400, {"error": "unknown_type"}) def _fit(self, status: int, value: dict) -> Answer: a = _answer(status, value) if len(a.body) <= _MAX_ANSWER_BYTES: return a self._log("[help] an answer was over 256 KB, which Help does not take; send less in the side panel.") return _answer(500, {"error": "answer_too_large"}) async def _once(self, key: str, ttl: int, run: Callable[[], Awaitable[Answer]]) -> Answer: state, result = self._store.claim(key, ttl) if state == "done" and result is not None: return Answer(200, {"Content-Type": "application/json"}, result.encode()) if state == "busy": return _answer(409, {"error": "in_progress"}) try: a = await run() except Exception as e: # what threw is not kept, so a press can be tried again self._store.release(key) self._log(f"[help] {key.split(':')[0]} failed: {type(e).__name__}") return _answer(500, {"error": "failed"}) self._store.finish(key, a.body.decode(), ttl) return a async def _maybe(value: Any) -> Any: return await value if inspect.isawaitable(value) else value def _run(coroutine: Awaitable[Answer]) -> Answer: try: asyncio.get_running_loop() except RuntimeError: return asyncio.run(coroutine) # type: ignore[arg-type] raise RuntimeError("Inside an event loop, await handler.handle_async(...) instead.")