Python

python -m venv .venv
source .venv/bin/activate
pip install shieldlabs==1.0.0

import json
import math
import os
import threading
from datetime import datetime, timezone
from typing import Any, Optional, cast
from urllib.request import Request, urlopen
from uuid import UUID

from shieldlabs import webhooks

DOMAIN_ID = str(UUID(os.environ["SHIELDLABS_DOMAIN_ID"]))
API_KEY = os.environ["SHIELDLABS_API_KEY"]
used: set[str] = set()  # demonstration only; use a shared durable store in production
used_lock = threading.Lock()


def read_identification(identification_id: str) -> dict[str, Any]:
    record_id = str(UUID(identification_id))
    request = Request(
        f"https://account.shieldlabs.ai/api/v1/domains/{DOMAIN_ID}/history/identifications/{record_id}",
        headers={"Authorization": "Bearer " + API_KEY, "Accept": "application/json"},
    )
    with urlopen(request, timeout=5) as response:
        card = cast(dict[str, Any], json.load(response)["data"])
    if (card.get("id"), card.get("domain_id"), card.get("type")) != (record_id, DOMAIN_ID, "identification"):
        raise ValueError("History identity mismatch")
    return card


def allow_signup(identification_id: str, expected_user_hid: Optional[str] = None) -> bool:
    card = read_identification(identification_id)
    risk, signals = card.get("risk", {}), card.get("signals", {})
    if not isinstance(risk, dict) or not isinstance(signals, dict):
        return False
    score = risk.get("score")
    if risk.get("state") != "ok" or signals.get("state") != "ok":
        return False
    if isinstance(score, bool) or not isinstance(score, (int, float)) or not math.isfinite(score) or not 0 <= score <= 100:
        return False
    band = "trusted" if score < 30 else "suspicious" if score < 60 else "dangerous"
    items = signals.get("items")
    if risk.get("band") != band or not isinstance(items, list) or not all(isinstance(x, str) for x in items):
        return False
    # Example policy: refuse Dangerous, Banned IP and browser automation.
    if score >= 60 or {"banned_ip", "browser_automation"}.intersection(items):
        return False
    observed = datetime.fromisoformat(str(card.get("created_at", "")).replace("Z", "+00:00"))
    if observed.tzinfo is None or not 0 <= (datetime.now(timezone.utc) - observed).total_seconds() <= 300:
        return False
    if expected_user_hid is not None and card.get("user_hid") != expected_user_hid:
        return False
    key = DOMAIN_ID + ":" + str(card["id"])
    with used_lock:
        if key in used:
            return False
        used.add(key)
    return True


def verify_webhook(raw_body: bytes, signature_header: Optional[str]) -> dict[str, Any]:
    secret = os.environ["SHIELDLABS_WEBHOOK_SECRET"]
    if not secret:
        raise RuntimeError("Webhook secret is not configured")
    if not webhooks.verify_signature(raw_body, signature_header, secret):
        raise ValueError("Invalid webhook signature")
    envelope = cast(dict[str, Any], json.loads(raw_body))
    if not isinstance(envelope.get("event_id"), str) or not envelope["event_id"]:
        raise ValueError("Missing event_id")
    return envelope  # retain the verified envelope, including data.identification_id