Account Recovery

async function initiateRecovery(recoveryRequest) {
  const res = await fetch('https://api.facesign.ai/sessions', {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${process.env.FACESIGN_API_KEY}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      clientReferenceId: `recovery-${recoveryRequest.id}`,
      metadata: {
        recoveryType: recoveryRequest.type,
        requestedAt: new Date().toISOString(),
      },
      flow: recoveryFlow, // The flow defined above
      videoAIAnalysisEnabled: true,
    }),
  })

  const { session, clientSecret } = await res.json()

  // Track the recovery attempt
  await db.recoveryAttempts.create({
    id: recoveryRequest.id,
    facesignSessionId: session.id,
    status: 'pending',
    createdAt: new Date(),
  })

  return clientSecret.url
}

import os
import requests
from datetime import datetime

def initiate_recovery(recovery_request):
    res = requests.post(
        "https://api.facesign.ai/sessions",
        headers={
            "Authorization": f"Bearer {os.environ['FACESIGN_API_KEY']}",
            "Content-Type": "application/json",
        },
        json={
            "clientReferenceId": f"recovery-{recovery_request['id']}",
            "metadata": {
                "recoveryType": recovery_request["type"],
                "requestedAt": datetime.now().isoformat(),
            },
            "flow": recovery_flow,  # The flow defined above
            "videoAIAnalysisEnabled": True,
        },
    )

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

    # Track the recovery attempt
    db.recovery_attempts.create(
        id=recovery_request["id"],
        facesign_session_id=session["id"],
        status="pending",
        created_at=datetime.now(),
    )

    return client_secret["url"]

import crypto from 'node:crypto'

// Verify an HMAC-SHA256 signature of the form `t=<timestamp>,v1=<hex>`
// over `${timestamp}.${rawBody}`, using the webhook secret from the
// FaceSign dashboard. Reject any request whose signature doesn't match.
function verifyFacesignSignature(rawBody, signatureHeader, secret) {
  if (!signatureHeader) return false
  const parts = Object.fromEntries(
    signatureHeader.split(',').map((p) => p.split('='))
  )
  if (!parts.t || !parts.v1) return false
  const expected = crypto
    .createHmac('sha256', secret)
    .update(`${parts.t}.${rawBody}`, 'utf8')
    .digest('hex')
  try {
    return crypto.timingSafeEqual(
      Buffer.from(parts.v1, 'hex'),
      Buffer.from(expected, 'hex')
    )
  } catch {
    return false
  }
}

export async function POST(req) {
  // Read the raw body before parsing — signature is computed over the
  // exact bytes we received.
  const rawBody = await req.text()
  const signature = req.headers.get('x-facesign-signature')

  if (!verifyFacesignSignature(rawBody, signature, process.env.FACESIGN_WEBHOOK_SECRET)) {
    return new Response('Invalid signature', { status: 401 })
  }

  const event = JSON.parse(rawBody)

  if (event.type !== 'session.status') {
    return Response.json({ received: true })
  }

  // Fetch the full session. GET /sessions/:id returns
  // `{ session, clientSecret }` — unpack the inner session.
  const sessionRes = await fetch(
    `https://api.facesign.ai/sessions/${event.sessionId}`,
    {
      headers: {
        Authorization: `Bearer ${process.env.FACESIGN_API_KEY}`,
      },
    }
  )
  const { session } = await sessionRes.json()

  // Find the recovery attempt
  const recovery = await db.recoveryAttempts.findBy({
    facesignSessionId: event.sessionId,
  })

  if (!recovery) {
    return Response.json({ received: true })
  }

  // Check recognition + liveness outcomes
  const recognitionReport = session.report?.nodeReports
    ?.find((r) => r.nodeId === 'recognition')
  const livenessReport = session.report?.nodeReports
    ?.find((r) => r.nodeId === 'liveness_known')

  const recognized = recognitionReport?.outcome === 'recognized'
  const livenessPass = livenessReport?.outcome === 'livenessDetected'

  if (session.status === 'complete' && recognized && livenessPass) {
    // Face matched an enrolled user. The recognition report exposes the
    // enrollment profile fields (name, email, phone, company, jobTitle) —
    // not an internal FaceSign user ID. Look the customer up in your own
    // database by the enrolled email.
    const matchedEmail = recognitionReport.report?.email
    const user = matchedEmail
      ? await db.users.findByEmail(matchedEmail)
      : null

    if (user) {
      await db.recoveryAttempts.update(recovery.id, {
        status: 'approved',
        matchedUserId: user.id,
      })

      // Generate a password reset token
      const resetToken = await auth.createPasswordResetToken(user.id)
      await notifications.sendResetLink(user.id, resetToken)
    } else {
      await db.recoveryAttempts.update(recovery.id, {
        status: 'denied',
        reason: 'matched_face_has_no_local_user',
      })
    }
  } else {
    await db.recoveryAttempts.update(recovery.id, {
      status: 'denied',
      reason: !recognized ? 'face_not_recognized' : 'liveness_failed',
    })
  }

  return Response.json({ received: true })
}

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"]

    # Find the recovery attempt
    recovery = db.recovery_attempts.find_by(
        facesign_session_id=event["sessionId"]
    )

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

    # Check recognition + liveness outcomes
    node_reports = (session.get("report") or {}).get("nodeReports") or []
    recognition_report = next(
        (r for r in node_reports if r["nodeId"] == "recognition"), None
    )
    liveness_report = next(
        (r for r in node_reports if r["nodeId"] == "liveness_known"), None
    )

    recognized = (
        recognition_report
        and recognition_report.get("outcome") == "recognized"
    )
    liveness_pass = (
        liveness_report
        and liveness_report.get("outcome") == "livenessDetected"
    )

    if session["status"] == "complete" and recognized and liveness_pass:
        # Face matched an enrolled user. The recognition report exposes
        # the enrollment profile fields (name, email, phone, company,
        # jobTitle) — not an internal FaceSign user ID. Look the customer
        # up in your own database by the enrolled email.
        matched_email = (recognition_report.get("report") or {}).get("email")
        user = (
            db.users.find_by_email(matched_email) if matched_email else None
        )

        if user:
            db.recovery_attempts.update(
                recovery["id"],
                status="approved",
                matched_user_id=user["id"],
            )

            reset_token = auth.create_password_reset_token(user["id"])
            notifications.send_reset_link(user["id"], reset_token)
        else:
            db.recovery_attempts.update(
                recovery["id"],
                status="denied",
                reason="matched_face_has_no_local_user",
            )
    else:
        db.recovery_attempts.update(
            recovery["id"],
            status="denied",
            reason="face_not_recognized" if not recognized else "liveness_failed",
        )

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