Session Webhooks

{
  "status": "success",
  "data": {
    "notification_config": {
      "webhook_url": "https://your-server.com/webhooks/kyc",
      "webhook_secret": "****************************a1b2",
      "notification_emails": [],
      "slack_webhook_url": null,
      "batch_notifications_enabled": true
    }
  }
}

curl -X PATCH https://kyc.legaltalent.ai/kyc/tenants/me/notification-channels \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "webhook_secret": null
  }'

POST /webhooks/kyc HTTP/1.1
Host: your-server.com
Content-Type: application/json
User-Agent: KYC-Webhooks/1.0
X-Webhook-ID: evt_7f8a9b2c3d4e5f6a7b8c9d0e
X-Webhook-Timestamp: 1735228800
X-Webhook-Signature: sha256=a1b2c3d4e5f6789012345678901234567890abcdef...

{
  "id": "evt_7f8a9b2c3d4e5f6a7b8c9d0e",
  "type": "kyc.session.processed",
  "created": 1735228800,
  "data": {
    // Event-specific payload
  }
}

message = "{timestamp}.{payload}"
signature = HMAC-SHA256(webhook_secret, message)

import time

MAX_TIMESTAMP_AGE = 300  # 5 minutes

def verify_timestamp(timestamp: str) -> bool:
    try:
        ts = int(timestamp)
        now = int(time.time())
        return abs(now - ts) < MAX_TIMESTAMP_AGE
    except ValueError:
        return False

{
  "id": "evt_abc123def456",
  "type": "kyc.session.created",
  "created": 1735228800,
  "data": {
    "event_type": "kyc.session.created",
    "tenant_id": "tenant_abc",
    "session_id": "sess_123456",
    "workflow_id": "wf_789012",
    "access_link": "https://kyc.legaltalent.ai/public/sessions/abc123token",
    "expires_at": 1735833600,
    "timestamp": "2024-12-26T12:00:00Z"
  }
}

{
  "id": "evt_def456ghi789",
  "type": "kyc.session.completed",
  "created": 1735230600,
  "data": {
    "event_type": "kyc.session.completed",
    "tenant_id": "tenant_abc",
    "session_id": "sess_123456",
    "workflow_id": "wf_789012",
    "current_step_index": 4,
    "total_steps": 5,
    "completed_at": "2024-12-26T12:30:00Z",
    "timestamp": "2024-12-26T12:30:00Z",
    "form_data": {
      "full_name": {
        "field_id": "field_abc123",
        "name": "Full Name",
        "type": "name",
        "value": "John Doe"
      },
      "date_of_birth": {
        "field_id": "field_xyz789",
        "name": "Date of Birth",
        "type": "date",
        "value": "1990-01-15"
      },
      "is_pep": {
        "field_id": "field_pep456",
        "name": "Are you a PEP?",
        "type": "boolean",
        "value": false
      }
    }
  }
}

{
  "<field_key>": {
    "field_id": "field_abc123",
    "name": "Human-readable label",
    "type": "field_type",
    "value": "user_entered_value"
  }
}

{
  "rule_type": "list_match",
  "action": "auto_deny",
  "severity": "critical",
  "message": "Found 2 match(es) in sanction lists",
  "rule_details": {
    "list_type": "ofac",
    "match_level": "high_confidence",
    "matching_entries": 2,
    "lists_matched": ["OFAC SDN", "UN Consolidated"]
  }
}

{
  "rule_type": "country",
  "action": "auto_deny",
  "severity": "critical",
  "message": "Country KP detected in nationality. Reason: Sanctioned country",
  "rule_details": {
    "country_code": "KP",
    "applies_to": ["all"],
    "matched_fields": ["nationality"]
  }
}

{
  "rule_type": "face_match",
  "action": "auto_deny",
  "severity": "critical",
  "message": "Face match failed (confidence: 45.00%, threshold: 80.00%)",
  "rule_details": {
    "confidence": 0.45,
    "threshold": 0.8,
    "is_match": false
  }
}

{
  "rule_type": "face_dedup",
  "action": "manual_review",
  "severity": "warning",
  "message": "Face detected in 3 previous session(s) (2 approved, 1 rejected)",
  "rule_details": {
    "min_sessions_required": 1,
    "sessions_found": 3,
    "approved": 2,
    "rejected": 1,
    "pending": 0,
    "matched_session_ids": ["sess_abc", "sess_def", "sess_ghi"]
  }
}

{
  "rule_type": "document",
  "action": "auto_deny",
  "severity": "critical",
  "message": "Document id_document expired 45 days ago",
  "rule_details": {
    "document_type": "id_document",
    "doc_id": "doc_abc123",
    "check_type": "expired",
    "expiry_date": "2024-11-01",
    "days_expired": 45
  }
}

{
  "rule_type": "document",
  "action": "manual_review",
  "severity": "info",
  "message": "Document proof_of_address issued 120 days ago (max: 90)",
  "rule_details": {
    "document_type": "proof_of_address",
    "doc_id": "doc_def456",
    "check_type": "too_old",
    "issue_date": "2024-09-01",
    "days_since_issue": 120,
    "max_allowed": 90
  }
}

{
  "rule_type": "document_validity",
  "action": "auto_deny",
  "severity": "critical",
  "message": "Invalid document: expected id_document, got receipt. Document type validation failed",
  "rule_details": {
    "doc_id": "doc_abc123",
    "expected_type": "id_document",
    "detected_type": "receipt",
    "validation_reason": "Document type validation failed"
  }
}

{
  "rule_type": "email",
  "action": "auto_deny",
  "severity": "critical",
  "message": "Email test@tempmail.com is from a disposable/temporary domain",
  "rule_details": {
    "email": "test@tempmail.com",
    "check_type": "disposable"
  }
}

{
  "rule_type": "volume",
  "action": "manual_review",
  "severity": "warning",
  "message": "Transaction volume: volume $150,000.00/month",
  "rule_details": {
    "monthly_volume": 150000,
    "monthly_transactions": 500,
    "rule_min_volume": 100000,
    "rule_max_volume": null,
    "rule_min_transactions": null,
    "rule_max_transactions": null
  }
}

{
  "rule_type": "time_to_complete",
  "action": "manual_review",
  "severity": "warning",
  "message": "Session completed in 15s (threshold: 60s)",
  "rule_details": {
    "completion_seconds": 15,
    "min_seconds": null,
    "max_seconds": 60
  }
}

{
  "rule_type": "form_field",
  "action": "manual_review",
  "severity": "warning",
  "message": "User declared PEP status (field 'pep_status' = 'Former PEP')",
  "rule_details": {
    "field_id": "pep_status",
    "field_value": "Former PEP",
    "operator": "in",
    "rule_value": ["PEP", "Former PEP", "Related to PEP"],
    "reason": "User declared PEP status"
  }
}

{
  "rule_type": "web_ssl",
  "action": "auto_deny",
  "severity": "critical",
  "message": "Website example.com has no SSL certificate",
  "rule_details": { "url": "example.com", "has_ssl": false }
}

{
  "rule_type": "web_industry",
  "action": "auto_deny",
  "severity": "critical",
  "message": "Website example.com industry 'gambling' is blocked",
  "rule_details": { "url": "example.com", "industry": "gambling", "allowed": false }
}

{
  "rule_type": "web_adverse_media",
  "action": "auto_deny",
  "severity": "critical",
  "message": "Website example.com has HIGH_RISK adverse media (score: 85)",
  "rule_details": { "url": "example.com", "decision": "HIGH_RISK", "risk_score": 85 }
}

{
  "rule_type": "web_sanction",
  "action": "auto_deny",
  "severity": "critical",
  "message": "Website example.com found on sanction lists",
  "rule_details": { "url": "example.com", "matches": ["OFAC"] }
}

{
  "rule_type": "extraction_validation",
  "action": "manual_review",
  "severity": "critical",
  "message": "Extracted full_name 'John Doe' matched in sanctions lists",
  "rule_details": {
    "validation_type": "sanctions_lists",
    "extracted_field": "full_name",
    "extracted_value": "John Doe",
    "doc_id": "doc_abc",
    "doc_type": "id_document"
  }
}

{
  "rule_type": "extraction_validation",
  "action": "flag",
  "severity": "warning",
  "message": "Extracted full_name 'JOHN D. DOE' differs from form value 'John Doe' (similarity: 85%)",
  "rule_details": {
    "validation_type": "cross_validate_form",
    "extracted_field": "full_name",
    "extracted_value": "JOHN D. DOE",
    "form_field": "full_name",
    "form_value": "John Doe",
    "similarity": 0.85,
    "threshold": 0.8
  }
}

function getCustomerFacingMessage(triggeredRules) {
  // Get the most critical rule
  const criticalRule = triggeredRules.find(r => r.severity === 'critical');
  
  if (criticalRule) {
    // Map rule types to user-friendly messages
    const messages = {
      'list_match': 'Your information could not be verified against our security databases.',
      'country': 'Unfortunately, we cannot accept applications from your country at this time.',
      'face_match': 'The photo on your ID could not be matched with your selfie. Please try again.',
      'document': 'Your document appears to be expired. Please upload a valid document.',
      'document_validity': 'The uploaded document type could not be verified. Please upload a valid ID.',
      'email': 'The email address provided could not be verified.',
    };
    
    return messages[criticalRule.rule_type] || criticalRule.message;
  }
  
  return 'Your application requires additional review.';
}

def log_session_decision(webhook_data):
    automation = webhook_data['data'].get('automation_result', {})
    
    for rule in automation.get('triggered_rules', []):
        logger.info({
            'event': 'automation_rule_triggered',
            'session_id': webhook_data['data']['session_id'],
            'rule_type': rule['rule_type'],
            'action': rule['action'],
            'severity': rule['severity'],
            'message': rule['message'],
            'details': rule['rule_details']
        })

{
  "id": "evt_jkl012mno345",
  "type": "kyc.session.approved",
  "created": 1735238400,
  "data": {
    "event_type": "kyc.session.approved",
    "tenant_id": "tenant_abc",
    "session_id": "sess_123456",
    "status": "approved",
    "notes": "Verified via phone call with customer. Documentation confirmed.",
    "updated_by": "compliance@yourcompany.com",
    "timestamp": "2024-12-26T14:00:00Z"
  }
}

{
  "id": "evt_mno345pqr678",
  "type": "kyc.session.rejected",
  "created": 1735238400,
  "data": {
    "event_type": "kyc.session.rejected",
    "tenant_id": "tenant_abc",
    "session_id": "sess_123456",
    "status": "rejected",
    "notes": "Document appears to be fraudulent. Inconsistent information provided.",
    "updated_by": "compliance@yourcompany.com",
    "timestamp": "2024-12-26T14:00:00Z"
  }
}

{
  "id": "evt_pqr678stu901",
  "type": "kyc.session.manual_review",
  "created": 1735238400,
  "data": {
    "event_type": "kyc.session.manual_review",
    "tenant_id": "tenant_abc",
    "session_id": "sess_123456",
    "status": "manual_review",
    "notes": "Additional documentation required",
    "updated_by": "system",
    "timestamp": "2024-12-26T14:00:00Z"
  }
}

from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
import hmac
import hashlib
import time
import json
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

WEBHOOK_SECRET = "whsec_your_secret"
MAX_TIMESTAMP_AGE = 300  # 5 minutes

async def process_session_created(data: dict):
    """Handle new session - maybe send link to user."""
    session_id = data['session_id']
    access_link = data['access_link']
    logger.info(f"New session created: {session_id}")
    # Send email to user with access_link
    
async def process_session_completed(data: dict):
    """Handle completed session - user finished their part."""
    session_id = data['session_id']
    logger.info(f"Session completed by user: {session_id}")
    # Notify internal team that documents are ready

async def process_session_processed(data: dict):
    """Handle processed session - main decision webhook."""
    session_id = data['session_id']
    decision = data['final_decision']
    automation = data.get('automation_result', {})
    risk_score = automation.get('risk_score', 0)
    flags = automation.get('flags', [])
    
    logger.info(f"Session {session_id} processed: {decision} (risk: {risk_score})")
    
    if decision == 'approved':
        # Activate user account immediately
        await activate_user(session_id)
    elif decision == 'flagged':
        # Activate user but add to monitoring queue
        await activate_user(session_id)
        await add_to_monitoring_queue(session_id, flags, risk_score)
    elif decision == 'rejected':
        # Send rejection notification
        await notify_rejection(session_id, automation.get('recommendation'))
    else:  # manual_review
        # Queue for compliance team
        await queue_for_review(session_id, automation)

async def process_status_update(data: dict):
    """Handle manual status updates."""
    session_id = data['session_id']
    status = data['status']
    notes = data.get('notes', '')
    
    logger.info(f"Session {session_id} status changed to: {status}")
    
    if status == 'approved':
        await activate_user(session_id)
    elif status == 'rejected':
        await notify_rejection(session_id, notes)

EVENT_HANDLERS = {
    'kyc.session.created': process_session_created,
    'kyc.session.completed': process_session_completed,
    'kyc.session.processed': process_session_processed,
    'kyc.session.approved': process_status_update,
    'kyc.session.rejected': process_status_update,
    'kyc.session.manual_review': process_status_update,
}

@app.post("/webhooks/kyc")
async def handle_kyc_webhook(request: Request, background_tasks: BackgroundTasks):
    # Get raw body for signature verification
    body = await request.body()
    
    # Get headers
    timestamp = request.headers.get('X-Webhook-Timestamp')
    signature = request.headers.get('X-Webhook-Signature')
    event_id = request.headers.get('X-Webhook-ID')
    
    # Validate headers
    if not timestamp or not signature:
        raise HTTPException(400, "Missing signature headers")
    
    # Verify timestamp freshness
    try:
        ts = int(timestamp)
        if abs(time.time() - ts) > MAX_TIMESTAMP_AGE:
            raise HTTPException(401, "Timestamp too old")
    except ValueError:
        raise HTTPException(400, "Invalid timestamp")
    
    # Verify signature
    sig = signature.replace('sha256=', '')
    message = f"{timestamp}.{body.decode('utf-8')}"
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        message.encode(),
        hashlib.sha256
    ).hexdigest()
    
    if not hmac.compare_digest(expected, sig):
        raise HTTPException(401, "Invalid signature")
    
    # Parse payload
    payload = json.loads(body)
    event_type = payload.get('type')
    data = payload.get('data', {})
    
    # Check idempotency (optional but recommended)
    if await is_already_processed(event_id):
        return {"received": True, "status": "already_processed"}
    
    # Find and queue handler
    handler = EVENT_HANDLERS.get(event_type)
    if handler:
        background_tasks.add_task(handler, data)
        await mark_as_processed(event_id)
    else:
        logger.warning(f"Unknown event type: {event_type}")
    
    return {"received": True, "event_id": event_id}

# Start your local server
python -m uvicorn main:app --port 8000

# In another terminal, expose it
ngrok http 8000

# Use the ngrok URL as your webhook_url
# https://abc123.ngrok.io/webhooks/kyc

import hmac
import hashlib
import time
import json

def generate_test_webhook(secret: str, event_type: str, data: dict) -> dict:
    """Generate a test webhook payload with valid signature."""
    timestamp = int(time.time())
    
    envelope = {
        "id": f"evt_test_{timestamp}",
        "type": event_type,
        "created": timestamp,
        "data": data
    }
    
    payload = json.dumps(envelope)
    message = f"{timestamp}.{payload}"
    signature = hmac.new(secret.encode(), message.encode(), hashlib.sha256).hexdigest()
    
    return {
        "headers": {
            "X-Webhook-ID": envelope["id"],
            "X-Webhook-Timestamp": str(timestamp),
            "X-Webhook-Signature": f"sha256={signature}",
            "Content-Type": "application/json"
        },
        "body": envelope
    }

# Generate test webhook
test = generate_test_webhook(
    secret="whsec_test123",
    event_type="kyc.session.processed",
    data={
        "event_type": "kyc.session.processed",
        "tenant_id": "test_tenant",
        "session_id": "test_session",
        "status": "approved",
        "final_decision": "approved"
    }
)
print(json.dumps(test, indent=2))