Webhook listener with FastAPI

Jan 9, 2021 · 3 min read · networking monitoring software ipfabric ·

Share on:

Overview

In the last post I used a webhook to notify a listener to run a remediation script.

In this post I'll document how to configure the listener.

FastAPI

There are many options to setup a webhook listener. After a brief research I decided to use FastAPI instead the more common Flask.

Webhook

What is a webook? In essence, it is an HTTP POST request sent to a webserver. The POST request usually has a payload containing some information useful to the responder.

Pydantinc model

FastAPI has a Pydantic approach, meaning the format of the payload must be defined in advance.

From the IP Fabric documentation we can get the exact format of the payload:

 1{
 2  "type": "snapshot",
 3  "action": "discover" | "clone" | "delete" | "download" | "load" | "unload",
 4  "status": "started" | "completed" | "failed" | "resumed" | "resumed (stopping)" | "stopped",
 5  "reason"?: string,
 6  "requester": "cron" | "user:<id>",
 7  "snapshot"?: {
 8    "id": string,
 9    "name"?: string,
10    "cloneId"?: string,
11    "file"?: string,
12  }
13  "timestamp": number,
14  "test"?: boolean
15}

Translated to a pydantic model:

1class WebhookData(BaseModel):
2  type: str
3  action: str
4  status: str
5  reason : Optional[str]
6  requester : str
7  snapshot: Optional[dict]
8  timestamp : int
9  test : Optional[bool]

Hashing

To ensure that the POST request arrives from a known source the payload is authenticated using a SHA256 HMAC payload hash signature.. The secret is defined when the webhook is created.

On the listener side we define an env var

1export IPF_WEBHOOK_SECRET=0123456789

and retrieve its value in Python with

1import os
2WEBHOOK_SECRET = os.environ.get("IPF_WEBHOOK_SECRET")

Then we need a function to validate the signature. This is the portion of code that validates the signature:

 1logger.info("Webhook received - checking signature")
 2raw_input = await request.body()
 3input_hmac = hmac.new(
 4    key=WEBHOOK_SECRET.encode(),
 5    msg=raw_input,
 6digestmod="sha256" # this is the hash algorithm used by IPF
 7)
 8if not hmac.compare_digest(
 9        input_hmac.hexdigest(),
10        x_ipf_signature
11):
12    response.status_code = 400
13    logger.error(f"Invalid message signature ignature}")
14    logger.error(f"Received signature {x_ipf_signature}")
15    logger.error(f"Computed signature {input_hmac.hexdigest    return {"result": "Invalid message signature"}
16logger.info("Message signature checked ok")

Wrap up

Automated remediation can be a starting point for a larger network automation initiative.

In this demo we fixed the issue in the workflow itself. In some cases, when a more conservative approach is preferable, remediation could be as simple as opening a support ticket to engage a network engineer to apply manually the fix.

Leveraging IP Fabric's automated discovery, data normalization, webhooks, and API we can implement an automation workflow minimizing the necessary code and all the associated additional work and risks.

Don't forget to subscribe to my youtube channel, I will posts more details about the webhook implementation soon.