import os
from typing import Optional
from shieldlabs import ShieldLabs, evaluate_identification, webhooks
client = ShieldLabs(api_key=os.environ["SHIELDLABS_API_KEY"]) # Private API Key, sec_...
used_request_ids: set[str] = set() # use your database or cache in production
def allow_signup(request_id: str) -> bool:
# 1. Read the identification for the request ID the browser sent with the form.
# Scoring is asynchronous, so this waits (up to 10 s by default) for the verdict.
identification = client.identifications.get(request_id)
# 2. Evaluate it: missing, reused, stale, rate-limited, automated or dangerous is refused.
verdict = evaluate_identification(identification, is_replay=lambda rid: rid in used_request_ids)
if identification is not None:
used_request_ids.add(identification.request_id)
return verdict.ok
def on_webhook(raw_body: bytes, signature_header: Optional[str]) -> None:
# 3. Verify and parse a delivery: the raw body bytes and the X-Shield-Signature header.
event = webhooks.construct_event(
raw_body,
signature_header,
os.environ["SHIELDLABS_WEBHOOK_SECRET"], # whsec_...
)
print(event.event_type)