Webhooks Verification

1
ts = "1592570791"
2
payload = u'<REQUEST BODY>'
3
sig_key = "shared symmetric key"
4
sig_input = "{ts}.{payload}".format(ts=ts, payload=payload)
5
6
hash_obj = hmac.new(key=bytes(sig_key, 'ascii'), msg=bytes(sig_input, 'utf-8'), digestmod=hashlib.sha256)
7
generated_sig = hash_obj.hexdigest() # "5257a869e7...."
8
return generated_signature == header_signature

1
'''
2
This code is purely for educational purposes and should not be used
3
in any type of production/live environment.
4
'''
5
6
import hashlib
7
import hmac
8
import json
9
from http.server import BaseHTTPRequestHandler, HTTPServer
10
11
PORT = 8000
12
SHARED_SYMMETRIC_KEY = "super-secret-webhooks-verification-key"
13
14
def validate_signature(header_ts, header_signature, post_body):
15
    sig_input = f"{header_ts}.{post_body}"
16
    print(f'sig_input: {sig_input}')
17
    hash_obj = hmac.new(
18
        key=bytes(SHARED_SYMMETRIC_KEY, 'ascii'),
19
        msg=bytes(sig_input, 'utf-8'),
20
        digestmod=hashlib.sha256
21
    )
22
    generated_signature = hash_obj.hexdigest()
23
    print(f'{generated_signature} ?==? {header_signature}')
24
    return generated_signature == header_signature
25
26
27
class handler(BaseHTTPRequestHandler):
28
    def do_POST(self):
29
        print(self.headers)
30
31
        # POST body extraction
32
        content_len = int(self.headers.get('Content-Length'))
33
        post_body = self.rfile.read(content_len).decode("utf-8")
34
        print(f'post_body: {post_body}')
35
36
        # Signature extraction from header
37
        header_ts, header_signature = self.headers['OrderGroove-Signature'].split(',')
38
        print(header_ts, header_signature)
39
        _, header_ts = header_ts.split('=')
40
        _, header_signature = header_signature.split('=')
41
        print(header_ts, header_signature)
42
43
        # Set response headers
44
        self.send_response(200)
45
        self.send_header('Content-type', 'text/html')
46
        self.end_headers()
47
48
        # Verify signature from header and send response
49
        if validate_signature(header_ts, header_signature, post_body):
50
            self.wfile.write(bytes("SUCCESS", "utf8"))
51
        else:
52
            self.wfile.write(bytes("SIGNATURE VERIFICATION FAILED!", "utf8"))
53
54
55
with HTTPServer(('', PORT), handler) as server:
56
    print(f"Server starting - listening on port {PORT}")
57
    server.serve_forever()

1
import hashlib
2
import hmac
3
import json
4
import time
5
6
SHARED_SYMMETRIC_KEY = "super-secret-webhooks-verification-key"
7
8
event_payload = {'a': {'webhook': 'event'}} # You can modify this as you like
9
payload_str = json.dumps(event_payload, separators=(',', ':'))
10
11
ts = str(1592570791) # You can modify this as you like - use int(time.time()) if you prefer
12
13
sig_input = f'{ts}.{payload_str}'
14
sig = hmac.new(
15
  key=bytes(SHARED_SYMMETRIC_KEY, 'ascii'), 
16
  msg=bytes(sig_input, 'utf-8'), 
17
  digestmod=hashlib.sha256
18
).hexdigest()
19
20
print(f'''
21
curl -X POST -L http://localhost:8000\\
22
 -H "Content-Length: {len(payload_str)}"\\
23
 -H "Content-Type: application/json"\\
24
 -H "OrderGroove-Signature: ts={ts},sig={sig}"\\
25
 -d '{payload_str}'
26
''')