IP Fabric Webhook Listener

In version 7.5.1 of the ipfabric Python SDK a new ipf_webhook_listener CLI tool was added. After installing the package it can be run from the command line.

This tool spins up a FastAPI web server that listens for incoming webhook events from IP Fabric. You can configure the tool to process snapshot and intent verification events by passing your custom Python functions into the CLI arguments.

Requirements

Requires installing the ipfabric Python SDK with CLI dependencies: pip install ipfabric[webhook].

Dependencies include: - fastapi - uvicorn - sqlmodel

Documentation

IPF Webhook Listener Help

>ipf_webhook_listener --help

usage: ipf_webhook_listener.cmd [-h] [--uvicorn_ip str] [--uvicorn_port int] [--uvicorn_reload bool] [--ipf_secret {str,null}] [--database_url str] [--title str] [--webhook_path str]
                                [--custom_settings {Any,null}] [--snapshot_event bool] [--snapshot_method {str,null}] [--snapshot_bg bool] [--snapshot_scheduled_only bool] [--intent_event bool]
                                [--intent_method {str,null}] [--intent_bg bool] [--intent_scheduled_only bool]

options:
  -h, --help            show this help message and exit

  --uvicorn_ip str      IP address to bind the FastAPI server. (default: 0.0.0.0)
  --uvicorn_port int    Port to bind the FastAPI server. (default: 8000)
  --uvicorn_reload bool
                        Enable auto-reload for the FastAPI server. (default: False)

  --ipf_secret {str,null}
                        Secret token to validate incoming webhook requests. Validation will be skipped if not set. (default: null)
  --database_url str    Database URL for storing webhook events. (default: sqlite:///database.db)
  --title str           Title of the FastAPI application. (default: IP Fabric Webhook Receiver)
  --webhook_path str    Path where the webhook endpoint will be available, default '/ipfabric'. (default: /ipfabric)
  --custom_settings {Any,null}
                        Settings to pass to your custom functions. You functions must handle this parameter. (default: null)

  --snapshot_method {str,null}
                        Path to the Python file containing the snapshot processing method in 'path:function' format. (default: null)
  --snapshot_scheduled_only bool
                        If set to True, only process webhooks from scheduled snapshots (default: True)

  --intent_method {str,null}
                        Path to the Python file containing the intent processing method in 'path:function' format. (default: null)
  --intent_scheduled_only bool
                        If set to True, only process webhooks from scheduled snapshots' intent verifications. Requires 'snapshot_event' to be configured and received for intent calculations. (default:
                        True)

Usage

Create a Custom Processing Function

First we create a Python file(s) containing our custom processing functions for snapshot and intent events.

The functions will receive the following parameters via kwargs:

• event: An instance of ipfabric.models.WebhookEvent containing the event data.
• settings: The WebhookConfig[custom_settings] you provided in the command line via --custom_settings.

custom_webhook_handlers.py

"""custom_webhook_handlers.py"""

from typing import Any
from ipfabric.models import WebhookEvent


def snapshot_event(event: WebhookEvent, settings: Any, **kwargs):
    """
    The following information is passed to the function via kwargs:
        - event: The WebhookEvent instance containing the event data.
        - settings: The WebhookConfig[custom_settings] you provided in the command line via --custom_settings.

    Please ensure your function contains '**kwargs' to accept any parameters passed as they may change in future versions.
    """
    if event.test:
        # Testing Event is always passed to your function so it is your responsibility to handle it.
        print("This is a test event.")
        return
    print("Not a test event.")
    if event.status == "completed":
        # If a completed Snapshot event, then do some processing
        print("Completed Snapshot Event Received; do some processing.")


def intent_event(event: WebhookEvent, **kwargs):
    if event.test:
        # Testing Event is always passed to your function so it is your responsibility to handle it.
        print("This is a test event.")
    elif event.status == "completed":
        # If a completed Intent Verification event, then do some processing
        print("Completed Snapshot Event Received; do some processing.")

• The functions you create must accept **kwargs to handle any future changes to the parameters passed.
• These methods will be run in the background and will not block the webhook listener from processing new incoming requests.
  • Since IP Fabric is the source of the webhook events:
    • Any response being returned is sent to the IPF VM.
    • The IPF VM does not process the response.
  • Because of this your method does not need to return any value because it will not be used.
• If you need to do some validation, reporting, or logging you can implement that in your custom function:
  • Log to a file or external logging service.
  • Send an email or message to a monitoring system or chat application.
  • Create a database that tracks processing results.

Webhook Event Model

The WebhookEvent model contains the following attributes:

Configuring the Webhook in IP Fabric

To configure the webhook in IP Fabric, navigate to Settings > Integrations > Webhooks and create a new webhook with the following settings:

• Name: A descriptive name for the webhook (e.g., “IPF Webhook Listener”).
• URL: The URL where the webhook listener is running (e.g., http://<listener_ip>:<listener_port>/ipfabric).

After creating the webhook it will auto-generate a Webhook Secret which will be passed into the CLI. Please save this secret as it will not be shown again.

Webhook Event Types

Select the event types you want to send to the webhook listener:

• Snapshot Events
  • Best for sending messages when snapshots are created or updated.
• Intent Verification Events
  • Best for all other post processing of data.

Most commonly automation should be triggered on completed Intent Verification events.

• After a snapshot is completed, IP Fabric will automatically run the Assurance Engine.
• Once the Assurance Engine completes, an Intent Verification event will be sent to the webhook listener.

If you process the event from a Snapshot event instead of an Intent Verification event you may run into the following issues:

• You will not have access to the Intent Verification results or possibly other data.
• If manipulating the snapshot data (e.g., updating sites, attributes, etc.), you could run into a job conflict.

If you want to process only Scheduled Intent Verifications, both events need selected.

1. --intent_scheduled_only flag defaults to True and will only process Intent Verification events from scheduled snapshots.
2. Snapshot events will have a requester of cron.
3. Once discovery completes, the Intent Verification event will be sent with a requester of snapshot:discover.
4. The automation looks at the event database to ensure the Intent Verification is only processed if a Scheduled Snapshot Event was received.
   • Record exists in DB matching the following: - Snapshot ID == Intent Verification Snapshot ID - Requester == cron - Status == completed.
5. It will also check to ensure that it runs once only after a completed scheduled snapshot and will skip any manual modifications (rediscoveries, adding/removing devices, etc.).
   • No records exist matching the following: - Snapshot ID == Intent Verification Snapshot ID - Requester != cron - Status == completed.

Running the Webhook Listener

By default the listener will do the following:

• Listen on all IP addresses 0.0.0.0 and port 8000.
• Use an SQLite database file named database.db in the current working directory.
• The webhook endpoint will be available at /ipfabric.
• When an event is received, it will be stored in the database.

Run IPF Webhook Listener with Custom Handlers

> ipf_webhook_listener --uvicorn_reload=1 \
--uvicorn_ip=127.0.0.1 \
--snapshot_method="custom_webhook_handlers.py:snapshot_event" \
--snapshot_scheduled_only=0 \
--intent_method="custom_webhook_handlers.py:intent_event" \
--ipf_secret=5b71eecda8fcf68c3f997ce9e963ea23

These flags can be used in combination with an .env file. By default it will look for a file named .env in the current working directory and then your home directory. CLI arguments will override any settings in the .env file.

.env File Example

uvicorn_ip=127.0.0.1
ipf_secret=5b71eecda8fcf68c3f997ce9e963ea23

This command will start the webhook listener with the following configurations:

• Auto-reload the server for development testing when changes are found.
• Bind to http://127.0.0.1:8000
• Use the snapshot_event function from custom_webhook_handlers.py for processing snapshot events.
  • --snapshot_scheduled_only is disabled it will process all snapshot events.
• Use the intent_event function from custom_webhook_handlers.py for processing intent verification events.
  • --intent_scheduled_only is enabled (default) it will only process intent verification events from scheduled snapshots.
• The webhook secret is set to validate incoming requests.
  • This is recommended for security to ensure requests are from your IP Fabric instance but can be omitted for testing.

API Documentation

By default FastAPI provides interactive API documentation at /docs and /redoc endpoints.

A healthy root endpoint is available at /healthcheck:

Healthcheck Endpoint

> curl http://127.0.0.1:8000/healthcheck
"Ok"

Also included endpoints:

• GET /events/: Retrieve all stored webhook events.
• GET /events/{dbId}: Retrieve a specific webhook event by its ID.
• GET /snapshots/: Retrieve all snapshot webhook events.
• GET /intents/: Retrieve all intent webhook events.

These endpoints have the following query parameters for filtering and paginating results:

• snapshotId: Filter events based on snapshotId.
• limit: Default 100, maximum 1000.
• offset: Default 0.