Python

Handle Twitch hub.secret logic in python

How to check a request authenticity made to a webhook

When subscribing to a webhook, Twitch suggests that we should pass a hub.secret param to the request body to authenticate the post requests done on our callback URL.

This is important because if someone finds your callback URL, they can make a post request with the same structure as Twitch and trick you to think an event happened when it didn’t.

By sending hub.secret, twitch will hash the payload with your secret, allowing you to check the authenticity of the request and confirm that it was sent by Twitch and not some dodgy third-party.

Checking Twitch documentation on webhooks you can see that Twitch will use your secret to sign the payload and will include an X-Hub-Signature to the header of the request, this signature is generated by sha256(secret, notification_bytes).

When Twitch sends the request you will have to use your hub.secret together with the payload received from Twitch to get the sha256 hash. Then you need to compare if this hash is equal to the one received on the X-Hub-Signature header.


Note that the X-Hub-Signature has the following format sha256=<sha hash>.


With this information, we can build our validation checker function to see if the request made to our URL did come from Twitch or not. The thing that we have to keep in mind is that we will always work with bytes.

I’m using opsdroid to connect to Twitch and subscribe to webhooks. Opsdroid uses AIOHTTP to create the web server as well to handle requests.

In aiohttp we can do await request.read() to read the response as bytes, then we just need to create our sha256 hash with this data together with the secret that we passed to hub.secret when subscribed to the webhook.

We can achieve this with the hashlib module.

python
1import hashlib
2
3async def validate_request(request, secret):
4 # Get x-hub-signature from header and remove 'sha256=' from it
5 signature = request.headers.get("x-hub-signature").replace("sha256=", '')
6
7 payload = await request.read()
8 validator = hashlib.sha256()
9
10 # Add payload together with secret encoded to bytes
11 validator.update(payload + secret.encode())
12
13 # Get hex representation
14 representation = validate.hexdigest()
15
16 return signature == representation

This function will achieve what we want but seems a bit messy and we could perhaps refactor it a bit more to make it more readable. We can use the hmac module to pass the payload and the secret as parameters and get the hex representation of our hash.

python
1async def validate_request(request, secret):
2 signature = request.headers.get("x-hub-signature").replace("sha256=", '')
3
4 payload = await request.read()
5
6 computed_hash = hmac.new(secret.encode(), msg=payload, digestmod=hashlib.sha256).hexdigest()
7
8 return signature == computed_hash

Now that we have our validator working, we can use it to check if a request did come from Twitch and then handle that request. If the request didn’t come from Twitch we can just send an error message back.


References:

Webmentions

0 Like 0 Comment

You might also like these

How to return an attribute from a many-to-many object relationship from a Django Ninja API endpoint.

Read More
Python

Django Ninja Schemas and Many To Many

Django Ninja Schemas and Many To Many

How I solved the issue of testing a function that should call sys.exit() when a yaml file couldn't be safely loaded.

Read More
Python

Unittest - How to test for sys.exit

Unittest -  How to test for sys.exit

An example from opsdroid on how to test if a logging call was made successfully.

Read More
Python

Test: Was Logging called

Test: Was Logging called

This is an example on how to use the side_effect function from the unittest module to test if the aiohttp exception ClientOSError was raised.

Read More
Python

Test for aiohttp's ClientErrorOS

Test for aiohttp's ClientErrorOS