WL: Using Webhooks

Overview

This documentation describes the webhook functionality for tracking user progress in the self-employment registration form. Webhooks are sent at key steps during the form completion process, allowing partners to track user progress and trigger follow-up actions.

Configuration

Setting Up Webhooks

Contact our team with your webhook processing URLs for dev and production environments to get access keys for separate environments.
Our team validates the request and activates the URLs in our system and will provide the secret keys for different environments
That's it. You can start implementing the webhook processing on your side and test it. See the Implementation Guide and Example Webhook Implementation for faster implementation.

Security

Each webhook request is signed using HMAC-SHA256
The signature is sent in the X-Webhook-Signature header
Partners should verify the signature using their secret key
The payload is encrypted using Fernet (symmetric encryption) derived from partner's secret key
The X-Webhook-Encrypted header indicates that the payload is encrypted
NOTE: The signature is generated from the encrypted payload, not the original payload. Always verify the signature before processing any payload data

Webhook Events

Event Types

Step Completion Events
- step_1_completed: User completed step 1
- step_2_completed: User completed step 2
- step_3_completed: User completed step 3
- step_4_completed: User completed step 4
- step_5_completed: User completed step 5
- step_6_completed: User completed step 6
Form Completion Event
- form_completed: User completed the entire form

Note: We don't save or send a webhook for a step until the user completes it and moves to the next one. For example, step 1 is sent only when the user reaches step 2. The form_completed event is triggered when the full form is submitted. Make sure to track intent and redirect users to step 1 in your app.

Webhook Headers

Content-Type: application/json
X-Webhook-Signature: <hmac-sha256-signature>
X-Webhook-Event: <event-type>
X-Webhook-Timestamp: <iso-8601-timestamp>
X-Webhook-Encrypted: true

Webhook Payload

The payload is encrypted using Fernet symmetric encryption. After decryption, the payload will have the following structure:

{
    "form_id": "258bfa8f-9751-4193-8b6a-ef0d8391d2dc",
    "current_step": 1,
    "created_at": "2024-04-05T16:02:28Z",
    "modified_at": "2024-04-05T16:13:09Z",
    "session_key": "958bfa8f-9751-4193-8b6a-ef0d8391d2dc",
    "external_user_id": "8ba52356-22ec-46e6-8fa6-2be1e6f4ee03", // Unique identifier for the user {utm_user_id
    "sent_at": "", // Date when the form was sent to the Tax Office; Only included in form_completed event
    "locale": "en",
    "event_type": "step_1_completed",
    "report_url": "", // Only included in form_completed event
    "form_data": {
        // Step-specific form data
        "civil_status" : "001",
        "person_a_gender" : "2",
        "person_a_last_name" : "Dore",
        "person_a_first_name" : "John",
        "person_a_birth_name" : "John",
        "person_a_current_profession" : "Sofware Dev",
        "person_a_dob" : "1995-12-12",
        "person_a_street" : "Hauptstraße",
        "person_a_house_number" : "28",
        "person_a_apartment_number" : "6",
        "person_a_address_ext" : "",
        "person_a_city" : "Berlin",
        "person_a_post_code" : "10115",
        "uses_post_office_box" : false,
        "person_a_separate_post_box_number" : "",
        "person_a_separate_post_box_postcode" : "",
        "person_a_separate_post_box_city" : "",
        "person_a_idnr" : "72951682030",
        "person_a_religion" : "11",
        "person_b_gender" : "1",
        "person_b_last_name" : "",
        "person_b_first_name" : "",
        "person_b_birth_name" : "",
        "person_b_current_profession" : "",
        "person_b_same_address" : false,
        "person_b_street" : "",
        "person_b_house_number" : "",
        "person_b_apartment_number" : "",
        "person_b_address_ext" : "",
        "person_b_city" : "",
        "person_b_post_code" : "",
        "person_b_idnr" : "",
        "person_b_religion" : "11",
        "person_a_phone_code" : "",
        "person_a_phone_area_code" : "1231",
        "person_a_phone_number" : "123123123",
        "person_a_email" : "[email protected]",
        "person_a_website" : "",
        "moved_from_other_german_city" : false,
        "person_a_other_city_moving_street" : "",
        "person_a_other_city_moving_house_number" : "",
        "person_a_other_city_moving_apartment_number" : "",
        "person_a_other_city_moving_address_ext" : "",
        "person_a_other_city_moving_city" : "",
        "person_a_other_city_moving_post_code" : ""
    }
}

Step-Specific Form Data

Each step's webhook includes only the relevant fields for that step:

Step 1: Personal Information
- Civil status
- Personal details (name, gender, DOB)
- Contact information
- Address details
Step 2: Business Information
- Profession description
- Business founding details
- Office information
- Previous business details (if applicable)
Step 3: Tax Information
- Previous tax number
- Profit determination method
- Revenue estimates
- VAT-related information
Step 4: Financial Projections
- Expected profits from various sources
- Income projections
- Special expenses
- Tax deductions
Step 5: Bank Account Information
- Business bank account details
- Private bank account details
Step 6: Tax Office Selection
- Selected tax office
Final Step:form_completed event
- All previously mentioned fields
- Date when the form was sent to the tax office
- Report URL (for form completion)

Check the Retrieve a tax registration API endpoint for more details on the form data structure, specific field options, and their possible values, such as Civil Status.

Implementation Guide

Verifying Webhook Signatures and Decrypting Payloads

import hmac
import hashlib
import base64
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC

def verify_webhook_signature(payload: str, signature: str, secret_key: str) -> bool:
    expected_signature = hmac.new(
        secret_key.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected_signature, signature)

def generate_encryption_key(secret_key: str) -> bytes:
    """Generate a Fernet key from the secret key using PBKDF2"""
    # Use PBKDF2 to derive a key suitable for Fernet
    kdf = PBKDF2HMAC(
        algorithm=hashes.SHA256(),
        length=32,
        salt=b"norman_webhook_salt",  # Fixed salt for consistency
        iterations=100000,
    )
    key = base64.urlsafe_b64encode(kdf.derive(secret_key.encode()))
    return key

def decrypt_payload(encrypted_payload: str, secret_key: str) -> str:
    """Decrypt the payload using Fernet (symmetric encryption)"""
    key = generate_encryption_key(secret_key)
    f = Fernet(key)
    decrypted_data = f.decrypt(base64.b64decode(encrypted_payload))
    return decrypted_data.decode('utf-8')

Handling Webhooks

Verify the signature using your secret key (the signature is generated from the encrypted payload)
Check the timestamp to prevent replay attacks
Decrypt the payload if the X-Webhook-Encrypted header is present
Process the event based on the event type
Store the form data as needed
Trigger any follow-up actions

Testing

We have a separate sandbox environment to test the webhooks before using them in the production environment. You need to send us different webhook processing URLs on your side, both for production and dev env. We will generate separate keys and activate these URLs according to the environment for processing the webhooks

Support

For any questions or issues regarding webhook implementation, please contact the Norman Finance support team.

Example Webhook Implementation

from flask import Flask, request, jsonify
import hmac
import hashlib
import base64
import json
from datetime import datetime, timezone
import logging
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

WEBHOOK_SECRET = "your-secret-key"

@app.route('/webhook', methods=['POST'])
def webhook():
    # Get headers first
    signature = request.headers.get('X-Webhook-Signature')
    event_type = request.headers.get('X-Webhook-Event')
    timestamp = request.headers.get('X-Webhook-Timestamp')
    
    # Get raw payload for signature verification
    raw_payload = request.get_data().decode('utf-8')
    
    # Verify signature before processing any data
    if not verify_webhook_signature(raw_payload, signature, WEBHOOK_SECRET):
        return jsonify({'error': 'Invalid signature'}), 401
    
    # Decrypt the payload
      try:
          decrypted_payload = decrypt_payload(raw_payload, WEBHOOK_SECRET)
          data = json.loads(decrypted_payload)
      except Exception as e:
          return jsonify({'error': 'Invalid encrypted payload'}), 400
    
    # Process webhook
    try:
        process_webhook(event_type, data)
        return jsonify({'status': 'success'}), 200
    except Exception as e:
        return jsonify({'error': 'Internal server error'}), 500

def verify_webhook_signature(payload: str, signature: str, secret_key: str) -> bool:
    if not signature or not secret_key:
        return False
        
    expected_signature = hmac.new(
        secret_key.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected_signature, signature)

def generate_encryption_key(secret_key: str) -> bytes:
    """Generate a Fernet key from the secret key using PBKDF2"""
    # Use PBKDF2 to derive a key suitable for Fernet
    kdf = PBKDF2HMAC(
        algorithm=hashes.SHA256(),
        length=32,
        salt=b"norman_webhook_salt",  # Fixed salt for consistency
        iterations=100000,
    )
    key = base64.urlsafe_b64encode(kdf.derive(secret_key.encode()))
    return key

def decrypt_payload(encrypted_payload: str, secret_key: str) -> str:
    """Decrypt the payload using Fernet (symmetric encryption)"""
    key = generate_encryption_key(secret_key)
    f = Fernet(key)
    decrypted_data = f.decrypt(base64.b64decode(encrypted_payload))
    return decrypted_data.decode('utf-8')

def process_webhook(event_type: str, data: dict):
    # Handle different event types
    if event_type == 'form_completed':
        handle_form_completion(data)
    elif event_type.startswith('step_'):
        handle_step_completion(event_type, data)

def handle_form_completion(data: dict):
    # Process completed form
    form_id = data['form_id']
    report_url = data['report_url']
    # ... implementation

def handle_step_completion(event_type: str, data: dict):
    # Process step completion
    step = data['current_step']
    form_data = data['form_data']
    # ... implementation