Creating a Webhook Server

Learn how to set up a server to handle results of file scans and alerts sent based on policy alert configurations.

Webhook Challenges

Nightfall will send a POST request with a JSON payload with a single field challenge containing randomly-generated bytes when it sends a message to a user-provided webhook address. This is to ensure that the caller owns the server.

{"challenge": "z78woE1uDFu7tPrPvEBV"}

In order to authenticate your webhook server to Nightfall, you must reply with (1) a 200 HTTP Status Code, and (2) a plaintext request body containing only the value of the challenge key.

If Nightfall receives the expected value back, then the file scan operation will proceed; otherwise it will be aborted.

When a server responds successfully to a challenge request, the validity of that URL will be cached for up to 24 hours, after which it will need to be validated again.

If the webhook cannot be reached, you will receive an error with the code "40012" and the description "Webhook URL validation failed" when you initiate the scan.

If the webhook challenge fails, you will receive an error with the code "42201" and the description "Webhook returned incorrect challenge response" when you initiate the scan.

Webhook Signature Verification

When a customer signs up for the developer platform, Nightfall automatically generates a unique siging secret for them.

This secret is used to sign requests to the customer's configured webhook URL.

If you has any concerns that their signing secret may have leaked, you can request rotation at any time by reaching out to Nightfall Customer Success.

For security purposes, the webhook includes a signature header containing an HMAC-SHA256 digital signature that customers may use to authenticate the client.

In order to authenticate requests to the webhook URL, customers may use the following algorithm:

1. Check for the presence of the headers X-Nightfall-Signature and X-Nightfall-Timestamp. If these headers are not both present, discard the request.

2. Read the entire request body into a string body.

3. Verify that the value in the X-Nightfall-Timestamp header (the POSIX time in seconds) occurred recently. This is to protect against replay attacks, so a threshold on the order of magnitude of minutes should be reasonable. If a request occurred too far in the past, it should be discarded.

4. Concatenate the timestamp and body with a colon delimiter, i.e. timestamp:body.

5. Compute the HMAC SHA-256 hash of the payload from the previous step, using your unique signing secret as the key. Encode this computed value in hex.

6. Compare the value of the X-Nightfall-Signature header to the value computed in the previous step. If the values match, authentication is successful, and processing should proceed. Otherwise, the request must be discarded.

The snippet below shows how you might implement this authentication validation in Python:

from datetime import datetime, timedelta
    import hmac
    import hashlib

    from flask import request

    SIGNING_SECRET = "super-secret"

    given_signature = request.headers.get('X-Nightfall-Signature')
    req_timestamp = request.headers.get('X-Nightfall-Timestamp')
    now = datetime.now()
    if now-timedelta(minutes=5) <= datetime.fromtimestamp(int(req_timestamp)) <= now:
        raise Exception("could not validate timestamp is within the last few minutes")
    computed_signature = hmac.new(
        SIGNING_SECRET.encode(),
        msg=F"{req_timestamp}:{request.get_data(as_text=True)}".encode(),
        digestmod=hashlib.sha256
    ).hexdigest().lower()
    if computed_signature != given_signature:
        raise Exception("could not validate signature of inbound request!")

Example Webhook Server

An example implementation of a simple webhook server is below.

import hmac
import hashlib
from os import getenv, path, mkdir

from flask import Flask, request
import requests

app = Flask(__name__)

output_dir = "findings"

SIGNING_SECRET = getenv("NF_SIGNING_SECRET")


@app.route("/", methods=['POST'])
def hello():
    content = request.get_json(silent=True)
    challenge = content.get("challenge")
    if challenge:
        return challenge
    else:
        verify_signature()

        print(F"Received request metadata: {content['requestMetadata']}")
        print(F"Received errors: {content['errors']}")

        if not content["findingsPresent"]:
            print(F"No findings for {content['uploadID']}")
            return "", 200
        print(F"S3 findings valid until {content['validUntil']}")
        response = requests.get(content["findingsURL"])
        save_findings(content["uploadID"], response.text)
        return "", 200


def verify_signature():
    if SIGNING_SECRET is None:
        return
    given_signature = request.headers.get('X-Nightfall-Signature')
    nonce = request.headers.get('X-Nightfall-Timestamp')
    computed_signature = hmac.new(
        SIGNING_SECRET.encode(),
        msg=F"{nonce}:{request.get_data(as_text=True)}".encode(),
        digestmod=hashlib.sha256
    ).hexdigest().lower()
    if computed_signature != given_signature:
        raise Exception("could not validate signature of inbound request!")


def save_findings(scan_id, finding_json):
    if not path.isdir(output_dir):
        mkdir(output_dir)
    output_path = path.join(output_dir, f"{scan_id}.json")
    with open(output_path, "w+") as out_file:
        out_file.write(finding_json)
    print(F"Findings for {scan_id} written to {output_path}")


if __name__ == "__main__":
    app.run(port=8075)

You can test your webhook with a tool such as ngrok which allows you expose a web server running on your local machine to the internet.

In the above example, the webhook server is running on port 8075. To route ngrok requests to this server, once you run the python script (having installed the necessary dependencies such getenv and Flask), you would run ngrok as follow:

./ngrok http 8075

See the section on Alerting for details about the json payloads for the different messages sent to webhook servers.