DEV Community 👩‍💻👨‍💻

Cover image for LGTM Devlog 11: Writing the Serverless Function for receiving GitHub webhooks with Pydantic validation
Yuan Gao
Yuan Gao

Posted on

LGTM Devlog 11: Writing the Serverless Function for receiving GitHub webhooks with Pydantic validation

Now that we know what the github forks look like, it's time to write the function for receiving them. You can see the code matching this post at this commit: 40ac1367

Data validation with Pydantic

First, validating data and extracting it. I need to pull out four or five pieces of data from the json being sent. While this is a relatively simple case of reading the various nested keys from the json, there's problems with that: what if some of the keys don't exist? We need to throw an error, ideally one that explains what the problem is. Suddenly pulling out 4 values becomes having to write four if statements and for error messages.

While in this particular instance, we don't expect the github API to be omitting values, it's still to do things rigorously. One of the best options for this validation problem is to use a module like Pydantic. Not only does it do this validation for you, it generates meaningful errors, has static type annotations, and a whole bunch of very wholesome reasons why you'd use Pydantic over manually decoding JSONs.

In fact, it's so useful that Python API libraries like Starlette/FastAPI built-in support for Pydantic models for defining what API endpoints require as payload, or output. Because we're running on Google's serverless functions, we don't have those built-in support for Pydantic, but we can still make use of it, we just have to manually run the validation.

So first, we define the data model(s). I noticed that the hook json payload has two members: forkee and repository. They both describe a repo. And inside each repo, is an owner that contains the details of the person who owns the repo. So I decided to buidl my data models like this:

(This is the entire contents of the app/utils/models.py file)

""" Models for validation of Github hooks """
from pydantic import BaseModel, Field, Extra  # pylint: disable=no-name-in-module

# pylint: disable=too-few-public-methods,missing-class-docstring
class GitHubUser(BaseModel):
    login: str = Field(..., title="User's user login name")
    id: int = Field(..., title="User's ID")

    class Config:
        extra = Extra.ignore


class GitHubRepository(BaseModel):
    id: int = Field(..., title="Repo's ID")
    full_name: str = Field(..., title="Repo's full name (including owner)")
    owner: GitHubUser = Field(..., title="Owner of repo")
    url: str = Field(..., title="API RUL of the repo")

    class Config:
        extra = Extra.ignore


class GitHubHookFork(BaseModel):
    forkee: GitHubRepository = Field(..., title="The fork created")
    repository: GitHubRepository = Field(..., title="The repository being forked")

    class Config:
        extra = Extra.ignore
Enter fullscreen mode Exit fullscreen mode

The extra config tells Pydantic to not error out when it receives more data than was given, since I don't want to fully define the github object, just the objects I want to access later.

Now in my main function, I can tell it to parse the raw data from the request and return this object:

    try:
        hook_fork = GitHubHookFork.parse_raw(request.data)
    except ValidationError as err:
        logger.err("Validation error", err=err)
        return abort(400, "Validation error")
Enter fullscreen mode Exit fullscreen mode

This hook_fork object has the properties we defined, for example, we can check the username of the person who forked our repo using hook_fork.forkee.owner.login, and! in the case validation fails, we'll get an exception containing what part of the data didn't match.

Signature validation

We discovered in the last post that GitHub can secure webhooks using a signature that is calculated from a pre-shared secret value that you tell it to use, and the payload. I found an example of this implementation on Google, but updated it to use the new SHA256 which GitHub wants you to use. I'm going to store the pre-shared secret in an environmental variable called SECRET (may rename later), so the code looks like this (contents of app/utils/verify.py):

""" Verify the GitHub webhook secret """
import os
import hmac
import hashlib

from flask import Request

SECRET = bytes(os.environ["SECRET"], "utf-8")


def verify_signature(request: Request) -> bool:
    """ Validates the github webhook secret. Will return false if secret not provided """
    expected_signature = hmac.new(
        key=SECRET, msg=request.data, digestmod=hashlib.sha256
    ).hexdigest()
    incoming_signature = request.headers.get("X-Hub-Signature-256", "").removeprefix(
        "sha256="
    )
    return hmac.compare_digest(incoming_signature, expected_signature)
Enter fullscreen mode Exit fullscreen mode

A couple of things to note here: the new Python 3.9 .removeprefix() method for strings, and using hmac.compare_digest() to compare the two digests. This method is equivalent to just doing == but it introduces some random timing to prevent time analysis, which makes it a bit more secure (though in our case that is unlikely to be a major consideration, it's still best-practice)

Main function

So finally, the main function that we will deploy looks like this (full contents of app/main.py):

""" Listens to webhooks from GitHub """

import structlog  # type: ignore
from flask import Request, abort
from pydantic import ValidationError

from utils.verify import verify_signature
from utils.models import GitHubHookFork

logger = structlog.get_logger()

OUR_REPO = "meseta/lgtm"


def github_webhook_listener(request: Request):
    """ A listener for github webhooks """

    # verify
    if not verify_signature(request):
        return abort(403, "Invalid signature")

    # decode
    try:
        hook_fork = GitHubHookFork.parse_raw(request.data)
    except ValidationError as err:
        logger.err("Validation error", err=err)
        return abort(400, "Validation error")

    # output
    if hook_fork.repository.full_name == OUR_REPO:
        logger.info("Got fork", data=hook_fork.dict())

    return "OK"
Enter fullscreen mode Exit fullscreen mode

Note that at this point, I'm not actually doing anything with the data, just decoding it and logging it. Next sprint's task will be to actually take this data and use it to create accounts and whatever.

Tests

To ensure we have full test coverage, I wrote a couple of tests, as well as downloaded the payload and hash for a valid hook, and fabricated an invalid one. The tests/conftest.py file now looks like this, with fixtures to provide the two test payloads as well as the test client:

""" Setup for tests """

import os
import json
import pytest
from functions_framework import create_app  # type: ignore

TEST_FILES = os.path.join(
    os.path.dirname(os.path.realpath(__file__)),
    "test_files",
)


class Payload:  # pylint: disable=too-few-public-methods
    """ Container for holding header/payload pairs during testing"""

    def __init__(self, header_path, payload_path):
        with open(os.path.join(TEST_FILES, header_path)) as fp:
            self.headers = json.load(fp)
        with open(os.path.join(TEST_FILES, payload_path), "rb") as fp:
            self.payload = fp.read()


@pytest.fixture()
def good_fork():
    """ A payload containing (raw) data, this is recorded from GitHub """
    return Payload("good_fork_headers.json", "good_fork.bin")


@pytest.fixture()
def bad_fork():
    """A payload containing (raw) data, that has been edited to be missing stuff
    but paired wiht a valid signature"""
    return Payload("bad_fork_headers.json", "bad_fork.bin")


@pytest.fixture(scope="package")
def client():
    """ Test client """
    return create_app(
        "github_webhook_listener", os.environ["FUNCTION_SOURCE"]
    ).test_client()

Enter fullscreen mode Exit fullscreen mode

And to test our model (contents of tests/test_model.py), we simply ask it to parse the good and the bad payloads to see if it'll work or raise the appropriate exceptions:

""" Tests for pydantic models """

import pytest
from pydantic import ValidationError
from app.utils.models import GitHubHookFork

# pylint: disable=redefined-outer-name
def test_model(good_fork):
    """ Test that our validator works """

    hook_fork = GitHubHookFork.parse_raw(good_fork.payload)
    assert hook_fork.forkee.owner.login


def test_model_invalid(bad_fork):
    """ Test that our validator fails correctly """

    with pytest.raises(ValidationError):
        GitHubHookFork.parse_raw(bad_fork.payload)
Enter fullscreen mode Exit fullscreen mode

And to test our main function, there's four tests that test various combinations of valid/invalid/missing signatures, valid/invalid payloads (contents of tests/test_main.py)

""" Tests for main.py """

# pylint: disable=redefined-outer-name
def test_good_fork(client, good_fork):
    """ For a good fork that's working fine """

    res = client.post("/", headers=good_fork.headers, data=good_fork.payload)
    assert res.status_code == 200


def test_model_validation_fail(client, bad_fork):
    """ Test model validation failure with bad payload but correct signature for it"""

    res = client.post("/", headers=bad_fork.headers, data=bad_fork.payload)
    assert res.status_code == 400


def test_signature_fail(client, good_fork, bad_fork):
    """ Test signature validation failure with good payload but incorrect signature for it"""

    res = client.post("/", headers=bad_fork.headers, data=good_fork.payload)
    assert res.status_code == 403


def test_no_signature(client, good_fork):
    """ When signatures are not supplied """

    res = client.post(
        "/", headers={"Content-Type": "application/json"}, data=good_fork.payload
    )
    assert res.status_code == 403

Enter fullscreen mode Exit fullscreen mode

Our final test results show all tests pass, with 100% code coverage, which isn't definitive that our code is bug free, but gives us a little confidence, and helps us make sure there's no regressions when we edit the code later

Test coverage

Running a real test

To run a real test, I fire up that ngrok endpoint I set up previously (by running ngrok http 5000), started up the test server using pipenv run serve github_webhook_listener, updating the GitHub Webhook URL to the new temporary ngrok path, and telling GitHub to re-send the previous webhook.

The server logs show that a webhook is received, and you can read the decoded variables that are now in the hook_fork object!

Validating the hook

For those interested, all of this is running inside VSCode's Terminal pane, opened into WSL, and inside a tmux session. It's a nice way to run code while developing.


This represents all that is needed for a basic GitHub webhook receiving serverless function, that is built in a reasonably robust way. Next we need to upload it, but I will set up the CI to do this instead of me having to upload it every time.

Top comments (0)

🌚 Browsing with dark mode makes you a better developer by a factor of exactly 40.

It's a scientific fact.