APIs & Integrations

dataops2tonnes
Member

Webhook signature v3 - Can't manage to match header using Python

SOLVE

Hello !

I'm trying to follow the documentation to validate the v3 request signature for webhooks using Python

But i can't manage to match the v3 signature provided in the request header with the signature that I compute on my side.

 

For my test, I used a sandbox environment with a private app that I set up. Then i created a webhook subscription using Hubspot UI ( in the webhook section of my private app), and triggered a test request in the details of my webhook subscription.

 

 

 

 

import hmac
from base64 import b64encode
from hashlib import sha256
import json

# Request data
header_signature = "..."
header_timestamp = "1732715701368"
request_payload = [
  {
    "appId": 4708013,
    "eventId": 100,
    "subscriptionId": 2963051,
    "portalId": 145106154,
    "occurredAt": 1732715551737,
    "subscriptionType": "deal.deletion",
    "attemptNumber": 0,
    "objectId": 123,
    "changeSource": "CRM",
    "changeFlag": "DELETED"
  }
]
# Private app data
webhook_url = "https://my.domain.com/webhook-path"
secret_key = "xxxxxxx-xxxx-xxxx-xxxxxxxxxxx"


msg = "POST" + webhook_url + json.dumps(request_payload) + header_timestamp
print(string)
# POSThttps://my.domain.com/webhook-path[{"appId": 4708013, "eventId": 100, "subscriptionId": 2963051, "portalId": 145106154, "occurredAt": 1732715551737, "subscriptionType": "deal.deletion", "attemptNumber": 0, "objectId": 123, "changeSource": "CRM", "changeFlag": "DELETED"}]1732715701368

computed_signature = b64encode(hmac.new(secret_key.encode("utf-8"), msg=msg.encode("utf-8"), digestmod=sha256).digest()).decode("utf-8")

print(computed_signature == header_signature)
# False

 

 

 

But my computed signature doesn't match with the header signature.

I made sure to use the app's private key, and the webhook url doesn't have any URL-encoded characters.

 

Does anyone have managed to make it work using Python ?

Or can help me solve this issue ?

 

Big thanks !

0 Upvotes
1 Accepted solution
zach_threadint
Solution
Top Contributor

Webhook signature v3 - Can't manage to match header using Python

SOLVE

Hi @dataops2tonnes 👋

 

I struggled a lot when trying to validate HubSpot requests in Python, so I'm very happy to share with you an example that works for me. Please note, my example uses Python's Flask app framework. If you're not using Flask, you'll have to adjust accordingly (it also relies on 2 variables declared in the given project's .env file). It supports HubSpot's validation signature versions 2 and 3.

"""
The following example:
1. Uses Python's Flask app framework
2. Relies on .env file variables, for example:
.env
----
service_domain="https://my.service.domain"
hs_app_client_secret="abc-def"
"""

from datetime import datetime, timezone
import sys
import os
import urllib.parse
import hmac
import base64
import hashlib
import json
from flask import Flask, request

# HubSpot Request Validation Function (Python)
# -- Supports HubSpot Signature Versions 2 and 3
# ==============================================
def hs_request_validation(request_method, request_headers_dict, request_path, request_body):
    """
    Returns True (boolean) for validated requests, False (boolean) for unvalidated requests
    """
    request_headers = {}
    for header in request_headers_dict.keys():
        request_headers[header.lower()] = request_headers_dict[header]
    if "x-hubspot-signature-v3" in request_headers.keys():
        # attempt to validate using V3 request signature
        # reject if request header timestamp is older than 5 minutes
        now_utc = datetime.now(tz=timezone.utc).timestamp()*1000
        request_timestamp = int(request_headers.get("x-hubspot-request-timestamp"))
        if now_utc - request_timestamp > 300000:
            # request timestamp is older than 5 mins, reject
            print(f"** request not validated - timestamp older than 5 mins - {now_utc}, {request_timestamp}")
            sys.stdout.flush()
            return False
        else:
            # continue validation process
            test_string = request_method + os.environ["service_domain"] + urllib.parse.unquote_plus(request_path) + request_body + request_headers.get("x-hubspot-request-timestamp")
            signature = base64.b64encode(
                hmac.new(
                    key=bytes(os.environ["hs_app_client_secret"], "utf-8"),
                    msg=bytes(test_string, "utf-8"),
                    digestmod=hashlib.sha256
                ).digest()
            ).decode()
            if signature == request_headers["x-hubspot-signature-v3"]:
                # request validated - v3
                return True
            else:
                print(f"** request not validated")
                print(f"test_string: {test_string}")
                print(f"test signature: {signature}")
                print(f"hs signature: {request_headers['x-hubspot-signature-v3']}")
                sys.stdout.flush()
                return False
    elif request_headers.get("x-hubspot-signature-version") == "v2":
        # v3 signature not in headers
        # attempt to validate using v2 request signature
        test_string = os.environ["hs_app_client_secret"] + request_method + os.environ["service_domain"] + urllib.parse.unquote_plus(request_path) + request_body
        signature = hashlib.sha256(test_string.encode('utf-8')).hexdigest()
        if signature == request_headers["x-hubspot-signature"]:
            # request validated - v2
            return True
        else:
            print(f"** request not validated")
            print(f"test_string: {test_string}")
            print(f"test signature: {signature}")
            print(f"hs signature: {request_headers['x-hubspot-signature']}")
            sys.stdout.flush()
            return False
    else:
        # invalid request
        print(f"** request not validated - unexpected hs headers")
        print(request_headers)
        sys.stdout.flush()
        return False

# initiate Flask app
app = Flask(__name__)

# define your service / app endpoint
# this is the endpoint that receives the given request from HubSpot
@app.route("/your-endpoint", methods=['POST'])
def your_endpoint():
    request_validated = hs_request_validation(request.method, dict(request.headers), request.path, json.dumps(request.get_json(), ensure_ascii=False, separators=(",", ":")))
    if request_validated:
        # request validated
        print("request validated")
        sys.stdout.flush()
    else:
        # request not validated
        pass
        print("request not validated")
        sys.stdout.flush()

I hope that proves helpful. Please let me know if you have any follow-up questions.

All the best,

Zach

--

Zach Klein
HubSpot Integrations & App Developer
Meanjin / Brisbane, Australia



Say g'day


If my post helped answer your query, please consider marking it as a solution.


View solution in original post

0 Upvotes
4 Replies 4
zach_threadint
Solution
Top Contributor

Webhook signature v3 - Can't manage to match header using Python

SOLVE

Hi @dataops2tonnes 👋

 

I struggled a lot when trying to validate HubSpot requests in Python, so I'm very happy to share with you an example that works for me. Please note, my example uses Python's Flask app framework. If you're not using Flask, you'll have to adjust accordingly (it also relies on 2 variables declared in the given project's .env file). It supports HubSpot's validation signature versions 2 and 3.

"""
The following example:
1. Uses Python's Flask app framework
2. Relies on .env file variables, for example:
.env
----
service_domain="https://my.service.domain"
hs_app_client_secret="abc-def"
"""

from datetime import datetime, timezone
import sys
import os
import urllib.parse
import hmac
import base64
import hashlib
import json
from flask import Flask, request

# HubSpot Request Validation Function (Python)
# -- Supports HubSpot Signature Versions 2 and 3
# ==============================================
def hs_request_validation(request_method, request_headers_dict, request_path, request_body):
    """
    Returns True (boolean) for validated requests, False (boolean) for unvalidated requests
    """
    request_headers = {}
    for header in request_headers_dict.keys():
        request_headers[header.lower()] = request_headers_dict[header]
    if "x-hubspot-signature-v3" in request_headers.keys():
        # attempt to validate using V3 request signature
        # reject if request header timestamp is older than 5 minutes
        now_utc = datetime.now(tz=timezone.utc).timestamp()*1000
        request_timestamp = int(request_headers.get("x-hubspot-request-timestamp"))
        if now_utc - request_timestamp > 300000:
            # request timestamp is older than 5 mins, reject
            print(f"** request not validated - timestamp older than 5 mins - {now_utc}, {request_timestamp}")
            sys.stdout.flush()
            return False
        else:
            # continue validation process
            test_string = request_method + os.environ["service_domain"] + urllib.parse.unquote_plus(request_path) + request_body + request_headers.get("x-hubspot-request-timestamp")
            signature = base64.b64encode(
                hmac.new(
                    key=bytes(os.environ["hs_app_client_secret"], "utf-8"),
                    msg=bytes(test_string, "utf-8"),
                    digestmod=hashlib.sha256
                ).digest()
            ).decode()
            if signature == request_headers["x-hubspot-signature-v3"]:
                # request validated - v3
                return True
            else:
                print(f"** request not validated")
                print(f"test_string: {test_string}")
                print(f"test signature: {signature}")
                print(f"hs signature: {request_headers['x-hubspot-signature-v3']}")
                sys.stdout.flush()
                return False
    elif request_headers.get("x-hubspot-signature-version") == "v2":
        # v3 signature not in headers
        # attempt to validate using v2 request signature
        test_string = os.environ["hs_app_client_secret"] + request_method + os.environ["service_domain"] + urllib.parse.unquote_plus(request_path) + request_body
        signature = hashlib.sha256(test_string.encode('utf-8')).hexdigest()
        if signature == request_headers["x-hubspot-signature"]:
            # request validated - v2
            return True
        else:
            print(f"** request not validated")
            print(f"test_string: {test_string}")
            print(f"test signature: {signature}")
            print(f"hs signature: {request_headers['x-hubspot-signature']}")
            sys.stdout.flush()
            return False
    else:
        # invalid request
        print(f"** request not validated - unexpected hs headers")
        print(request_headers)
        sys.stdout.flush()
        return False

# initiate Flask app
app = Flask(__name__)

# define your service / app endpoint
# this is the endpoint that receives the given request from HubSpot
@app.route("/your-endpoint", methods=['POST'])
def your_endpoint():
    request_validated = hs_request_validation(request.method, dict(request.headers), request.path, json.dumps(request.get_json(), ensure_ascii=False, separators=(",", ":")))
    if request_validated:
        # request validated
        print("request validated")
        sys.stdout.flush()
    else:
        # request not validated
        pass
        print("request not validated")
        sys.stdout.flush()

I hope that proves helpful. Please let me know if you have any follow-up questions.

All the best,

Zach

--

Zach Klein
HubSpot Integrations & App Developer
Meanjin / Brisbane, Australia



Say g'day


If my post helped answer your query, please consider marking it as a solution.


0 Upvotes
dataops2tonnes
Member

Webhook signature v3 - Can't manage to match header using Python

SOLVE

Hello !
I'm following Hubspot documentation on validating the v3 request signature, but i can't manage to make it works using Python

 

Here is my code

 

 

import hmac
from base64 import b64encode
from hashlib import sha256
import json

# Request data
header_signature = "P8ucqETtHzOBLWuDCCxNpEvUhXHBwvISTqTfM7DVQCM="
header_timestamp = "1732715701368"
request_payload = [
    {
        "appId": 4708013,
        "eventId": 100,
        "subscriptionId": 2963051,
        "portalId": 145106154,
        "occurredAt": 1732715551737,
        "subscriptionType": "deal.deletion",
        "attemptNumber": 0,
        "objectId": 123,
        "changeSource": "CRM",
        "changeFlag": "DELETED",
    }
]
# Private app data
webhook_url = "https://my.domain.com/webhook-path"
secret_key = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"


string = "POST" + webhook_url + json.dumps(request_payload) + header_timestamp
print(string)
# POSThttps://my.domain.com/webhook-path[{"appId": 4708013, "eventId": 100, "subscriptionId": 2963051, "portalId": 145106154, "occurredAt": 1732715551737, "subscriptionType": "deal.deletion", "attemptNumber": 0, "objectId": 123, "changeSource": "CRM", "changeFlag": "DELETED"}]1732715701368

computed_string = b64encode(
    hmac.new(
        key=secret_key.encode("utf-8"), 
        msg=string.encode("utf-8"), 
        digestmod=sha256
    ).digest()
).decode("utf-8")

print(computed_string == header_signature)
# False

 

 

For the context, i'm using a sandbox account where I configured a private app.

I then created a webhook subscription in my app using Hubspot UI, and fired a test request in my webhook subscription detail

 

Does anyone have managed to make it work using Python ?

Or can help me solve my issue ?

 

Big thanks !

0 Upvotes
dataops2tonnes
Member

Webhook signature v3 - Can't manage to match header using Python

SOLVE

Hey Zach

 

Thanks for your quick reply.

I do use Flask too.
I looked at your code and I did manage to make it works !

 

Crazy... Issue was in the jsonification of the request body. It's good to know that Hubspot doesn't have whitespaces in the json output when generating the signature, which is not the default behavior of Python's json.dumps() method

 

import json

request_payload = [
    {
        "appId": 4708013,
        "eventId": 100,
        "subscriptionId": 2963051,
        "portalId": 145106154,
        "occurredAt": 1732783951647,
        "subscriptionType": "deal.deletion",
        "attemptNumber": 0,
        "objectId": 123,
        "changeSource": "CRM",
        "changeFlag": "DELETED",
    }
]

# Works
print(json.dumps(request_payload, ensure_ascii=False, separators=(",", ":")))
# Doesn't works
print(json.dumps(request_payload))

# Output
# [{"appId":4708013,"eventId":100,"subscriptionId":2963051,"portalId":145106154,"occurredAt":1732783951647,"subscriptionType":"deal.deletion","attemptNumber":0,"objectId":123,"changeSource":"CRM","changeFlag":"DELETED"}]

# [{"appId": 4708013, "eventId": 100, "subscriptionId": 2963051, "portalId": 145106154, "occurredAt": 1732783951647, "subscriptionType": "deal.deletion", "attemptNumber": 0, "objectId": 123, "changeSource": "CRM", "changeFlag": "DELETED"}]

 

 

Big thanks Zach, it was such a headache haha

I owe you a coffee / a beer / whatever you want

 

Jimmy Eung

Developer at 2tonnes

Paris, France

zach_threadint
Top Contributor

Webhook signature v3 - Can't manage to match header using Python

SOLVE

@dataops2tonnes I'm really glad it helped  -- I remember how long it took me to get to the bottom of that issue, so I'm more than happy to help!

All the best,

Zach

--

Zach Klein
HubSpot Integrations & App Developer
Meanjin / Brisbane, Australia



Say g'day


If my post helped answer your query, please consider marking it as a solution.


0 Upvotes