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:

{
  "type": "snapshot",
  "action": "discover" | "clone" | "delete" | "download" | "load" | "unload",
  "status": "started" | "completed" | "failed" | "resumed" | "resumed (stopping)" | "stopped",
  "reason"?: string,
  "requester": "cron" | "user:<id>",
  "snapshot"?: {
    "id": string,
    "name"?: string,
    "cloneId"?: string,
    "file"?: string,
  }
  "timestamp": number,
  "test"?: boolean
}

Translated to a pydantic model:

class WebhookData(BaseModel):
  type: str
  action: str
  status: str
  reason : Optional[str]
  requester : str
  snapshot: Optional[dict]
  timestamp : int
  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

export IPF_WEBHOOK_SECRET=0123456789

and retrieve its value in Python with

import os
WEBHOOK_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:

logger.info("Webhook received - checking signature")
raw_input = await request.body()
input_hmac = hmac.new(
    key=WEBHOOK_SECRET.encode(),
    msg=raw_input,
digestmod="sha256" # this is the hash algorithm used by IPF
)
if not hmac.compare_digest(
        input_hmac.hexdigest(),
        x_ipf_signature
):
    response.status_code = 400
    logger.error(f"Invalid message signature ignature}")
    logger.error(f"Received signature {x_ipf_signature}")
    logger.error(f"Computed signature {input_hmac.hexdigest    return {"result": "Invalid message signature"}
logger.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.