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.
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
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
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