Step-Up from Anomaly Detection

import time

RISK_THRESHOLD = 75  # Score 0-100; tune for your risk tolerance

async def check_risk_and_step_up(user_id, action, context):
    # Query your fraud engine
    risk_score = await fraud_engine.evaluate(
        user_id=user_id,
        action=action,  # e.g., "change_email", "add_payee", "export_data"
        ip=context["ip"],
        device_fingerprint=context["device_id"],
        geo_location=context["geo"],
        timestamp=int(time.time() * 1000),
    )

    if risk_score < RISK_THRESHOLD:
        # Low risk -- allow the action
        return {"allowed": True}

    # High risk -- require FaceSign step-up
    verification_url = await create_step_up_session(
        user_id, action, risk_score, context
    )

    return {
        "allowed": False,
        "requires_step_up": True,
        "verification_url": verification_url,
    }

import os
import time
import requests

ACTION_DESCRIPTIONS = {
    "change_email": "change your email address",
    "add_payee": "add a new payment recipient",
    "export_data": "export your account data",
    "change_password": "change your password",
    "large_transfer": "make a large transfer",
}

async def create_step_up_session(user_id, action, risk_score, context):
    user = await db.users.find_by_id(user_id)
    action_desc = ACTION_DESCRIPTIONS.get(action, action)

    res = requests.post(
        "https://api.facesign.ai/sessions",
        headers={
            "Authorization": f"Bearer {os.environ['FACESIGN_API_KEY']}",
            "Content-Type": "application/json",
        },
        json={
            "clientReferenceId": f"stepup-{user_id}-{int(time.time() * 1000)}",
            "metadata": {
                "userId": user_id,
                "action": action,
                "riskScore": risk_score,
                "triggerIp": context["ip"],
                "triggerGeo": context.get("geo"),
            },
            "providedData": {
                "email": user["email"],
            },
            "flow": [
                {"id": "start", "type": "start", "outcome": "greeting"},
                {
                    "id": "greeting",
                    "type": "conversation",
                    "prompt": (
                        "Say: Hi! We noticed something unusual on your account "
                        "and want to make sure it's really you. This will only "
                        "take a moment — I'll take a quick look, then ask you "
                        "to confirm."
                    ),
                    "outcomes": [
                        {"id": "next", "targetNodeId": "liveness", "condition": ""}
                    ],
                },
                {
                    "id": "liveness",
                    "type": "liveness_detection",
                    "outcomes": {
                        "livenessDetected": "confirm_action",
                        "deepfakeDetected": "closing",
                        "noFace": "closing",
                    },
                },
                {
                    "id": "confirm_action",
                    "type": "conversation",
                    "prompt": (
                        f"Say: You're attempting to {action_desc} from "
                        f"{context.get('geo', 'an unrecognized location')}. "
                        f"Can you confirm this is you and you intend to take this action?"
                    ),
                    "outcomes": [
                        {
                            "id": "confirmed",
                            "targetNodeId": "email_otp",
                            "condition": "The user explicitly confirmed the action is theirs",
                        },
                        {
                            "id": "denied",
                            "targetNodeId": "closing",
                            "condition": "The user denied the action, disputed details, or expressed concern",
                        },
                        {
                            "id": "stall",
                            "targetNodeId": "closing",
                            "condition": "Conversation reached 4 exchanges with no clear answer",
                        },
                    ],
                },
                {
                    "id": "email_otp",
                    "type": "two_factor_email",
                    "otpLength": 6,
                    "expirySeconds": 300,
                    "outcomes": {
                        "verified": "closing",
                        "delivery_failed": "closing",
                        "failed_unverified": "closing",
                        "cancelled": "closing",
                    },
                },
                {
                    "id": "closing",
                    "type": "conversation",
                    "prompt": (
                        "Thank the user for completing the verification and "
                        "tell them the session is now complete. Keep it to one sentence."
                    ),
                    "doesNotRequireReply": True,
                    "outcomes": [
                        {"id": "done", "targetNodeId": "end", "condition": ""}
                    ],
                },
                {"id": "end", "type": "end"},
            ],
            "videoAIAnalysisEnabled": True,
        },
    )

    data = res.json()
    session = data["session"]
    client_secret = data["clientSecret"]

    # Store the pending step-up
    await db.step_up_requests.create(
        user_id=user_id,
        action=action,
        risk_score=risk_score,
        facesign_session_id=session["id"],
        status="pending",
    )

    return client_secret["url"]

import hmac, hashlib, json, os
from flask import Flask, request, jsonify, abort
import requests

app = Flask(__name__)


def verify_facesign_signature(raw_body: bytes, signature_header: str, secret: str) -> bool:
    """HMAC-SHA256 of ``{timestamp}.{raw_body}`` must match ``v1`` in the header."""
    if not signature_header:
        return False
    try:
        parts = dict(p.split("=", 1) for p in signature_header.split(","))
    except ValueError:
        return False
    if "t" not in parts or "v1" not in parts:
        return False
    expected = hmac.new(
        secret.encode("utf-8"),
        f"{parts['t']}.".encode("utf-8") + raw_body,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(parts["v1"], expected)


@app.route("/webhooks/facesign", methods=["POST"])
def handle_webhook():
    # Read the raw body before parsing — signature is computed over the
    # exact bytes we received.
    raw_body = request.get_data()
    signature = request.headers.get("X-FaceSign-Signature")

    if not verify_facesign_signature(
        raw_body, signature, os.environ["FACESIGN_WEBHOOK_SECRET"]
    ):
        abort(401, description="Invalid signature")

    event = json.loads(raw_body)

    if event["type"] != "session.status":
        return jsonify({"received": True}), 200

    # Fetch the full session. GET /sessions/:id returns
    # {"session": ..., "clientSecret": ...} — unpack the inner session.
    session_res = requests.get(
        f"https://api.facesign.ai/sessions/{event['sessionId']}",
        headers={
            "Authorization": f"Bearer {os.environ['FACESIGN_API_KEY']}",
        },
    )
    session = session_res.json()["session"]

    step_up = db.step_up_requests.find_by(
        facesign_session_id=event["sessionId"]
    )

    if not step_up:
        return jsonify({"received": True}), 200

    # Evaluate results
    node_reports = (session.get("report") or {}).get("nodeReports") or []
    liveness_report = next(
        (r for r in node_reports if r["nodeId"] == "liveness"), None
    )
    otp_report = next(
        (r for r in node_reports if r["nodeId"] == "email_otp"), None
    )

    liveness_pass = (
        liveness_report
        and liveness_report.get("outcome") == "livenessDetected"
    )
    otp_pass = otp_report and otp_report.get("outcome") == "verified"
    verified = session["status"] == "complete" and liveness_pass and otp_pass

    # Feed result back to your fraud engine
    fraud_engine.report_step_up_result(
        user_id=step_up["user_id"],
        action=step_up["action"],
        original_risk_score=step_up["risk_score"],
        facesign_verified=verified,
        facesign_session_id=session["id"],
        liveness_outcome=liveness_report.get("outcome") if liveness_report else None,
    )

    if verified:
        # Allow the original action to proceed
        db.step_up_requests.update(step_up["id"], status="verified")
        action_queue.release(step_up["user_id"], step_up["action"])
    else:
        # Block the action and flag the account
        db.step_up_requests.update(step_up["id"], status="failed")
        fraud_engine.escalate(
            user_id=step_up["user_id"],
            reason="step_up_verification_failed",
            liveness_outcome=liveness_report.get("outcome") if liveness_report else None,
        )

    return jsonify({"received": True}), 200