智能小程序OPENCARD
接入指南

服务端 Webhook

Python Webhook

对于使用 Python 的 Opencard 应用, 开发者可以参考以下 Python 示例代码,来实现 Webhook 服务。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
#!/usr/bin/env python
# encoding: utf-8

import BaseHTTPServer
import time

# jwcrypto 不是标准库,需额外安装: pip install jwcrypto
from jwcrypto import jwk, jwe
from jwcrypto.common import json_encode, json_decode

# 配置项,测试时请修改以下配置
HOST_NAME = 'localhost'
PORT_NUMBER = 8000
PSK_TABLE = {'0':jwk.JWK.from_json('{"kty":"oct","k":"MDEyMzQ1Njc4OWFiY2RlZg"}')}

# Webhook 基本协议实现
#
# 以下代码为示例,未妥善处理业务字段或异常等因素
class Webhook(BaseHTTPServer.BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(404)

def do_POST(self):
content_length = int(self.headers['Content-Length'])
req_body = self.rfile.read(content_length)
jwetoken = jwe.JWE()
# 从 protected 字段中解出来 kid
jwetoken.deserialize(req_body)
kid = jwetoken.jose_header["kid"]
# 通过 kid 确定要用哪个解密 key
key = PSK_TABLE[kid]
# 可以将 rid 打印到日志中,以便问题追查时 join 日志
print time.asctime(), "Req head %s" % json_encode(jwetoken.jose_header)
# 用 key 解密取得解密后的内容
jwetoken.decrypt(key)
print time.asctime(), "Req json %s" % jwetoken.payload.decode("unicode_escape")
req = json_decode(jwetoken.payload)
# 如果资源方有多张卡片,从 req 中解析出来 srcid,处理相应的业务逻辑
if req["srcid"] == "123":
# do something
res = {"status":0,"msg":""}
res["data"] = {
"item_list":[{"title":"故宫博物院"}],
"jump_url":"/path/to/page3"
}
else:
res = {"status":2,"msg":"Invalid srcid"}
# 处理业务响应结果
payload = json_encode(res)
print time.asctime(), "Res json %s" % payload.decode("unicode_escape")
# 用请求的 header 和 key 生成加密结果,保持加密算法和头内容(kid/rid)与请求一致
jwetoken = jwe.JWE(plaintext=payload,
protected=json_encode(jwetoken.jose_header),
recipient=key)
res_body = jwetoken.serialize(compact=True)
self.send_response(200)
self.end_headers()
self.wfile.write(res_body)

if __name__ == '__main__':
httpd = BaseHTTPServer.HTTPServer((HOST_NAME, PORT_NUMBER), Webhook)
print time.asctime(), "Webhook Starts - http://%s:%s" % (HOST_NAME, PORT_NUMBER)
try:
httpd.serve_forever()
except KeyboardInterrupt:
pass
httpd.server_close()
print time.asctime(), "Webhook Stops - http://%s:%s" % (HOST_NAME, PORT_NUMBER)
反 馈帮 助 回 到顶 部