如何使用 Python 向 Viber 机器人发送消息?
How to send message to Viber bot with Python?
我有以下 HTTPS 服务器:
from flask import Flask, request, Response
from viberbot import Api
from viberbot.api.bot_configuration import BotConfiguration
from viberbot.api.messages import VideoMessage
from viberbot.api.messages.text_message import TextMessage
import logging
from viberbot.api.viber_requests import ViberConversationStartedRequest
from viberbot.api.viber_requests import ViberFailedRequest
from viberbot.api.viber_requests import ViberMessageRequest
from viberbot.api.viber_requests import ViberSubscribedRequest
from viberbot.api.viber_requests import ViberUnsubscribedRequest
logger = logging.getLogger(__name__)
app = Flask(__name__)
viber = Api(BotConfiguration(
name='PythonSampleBot',
avatar='http://www.clker.com/cliparts/3/m/v/Y/E/V/small-red-apple-hi.png',
auth_token='xxx-xxx-xxx'
))
@app.route('/', methods=['POST'])
def incoming():
logger.debug("received request. post data: {0}".format(request.get_data()))
# every viber message is signed, you can verify the signature using this method
if not viber.verify_signature(request.get_data(), request.headers.get('X-Viber-Content-Signature')):
return Response(status=403)
# this library supplies a simple way to receive a request object
viber_request = viber.parse_request(request.get_data())
if isinstance(viber_request, ViberMessageRequest):
message = viber_request.message
# lets echo back
viber.send_messages(viber_request.sender.id, [
message
])
elif isinstance(viber_request, ViberSubscribedRequest):
viber.send_messages(viber_request.get_user.id, [
TextMessage(text="thanks for subscribing!")
])
elif isinstance(viber_request, ViberFailedRequest):
logger.warn(
"client failed receiving message. failure: {0}".format(viber_request))
return Response(status=200)
def set_webhook(viber_bot):
viber_bot.set_webhook('https://xxx.xxx.xxx.xxx:4443')
logging.info("Web hook has been set")
if __name__ == "__main__":
context = ('certificate.pem', 'key.pem')
app.run(host='0.0.0.0', port=4443, debug=True, ssl_context=context)
正在尝试发送消息:
import json
import requests
webhook_url = 'https://xxx.xxx.xxx.xxx:4443'
data = {
"receiver": "xxx-xxx-xxx",
"type": "text",
"text": "Hello world!"
}
response = requests.post(
webhook_url, data=json.dumps(data),
headers={'Content-Type': 'application/json'},
verify='E:\Docs\learn_py\viberbot\certificate.pem'
)
if response.status_code != 200:
raise ValueError(
'Request returned an error %s, the response is:\n%s'
% (response.status_code, response.text)
)
我收到 403 错误
ValueError: Request returned an error 403, the response is:
更新:
403 来自:
if not viber.verify_signature(request.get_data(), request.headers.get('X-Viber-Content-Signature')):
return Response(status=403)
已编辑 由于更新。您在 verify_signature
.
上遇到错误
verify_signature
的定义:
def verify_signature(self, request_data, signature):
return signature == self._calculate_message_signature(request_data)
您正在向那里发送 request.headers.get('X-Viber-Content-Signature')
作为“签名”。因此,您的解决方案是检查 __calculate_message_signature(request_data)
的结果
requiest_data = request.get_data() 你的情况。
_calculate_message_signature
的定义是:
def _calculate_message_signature(self, message):
return hmac.new(
bytes(self._bot_configuration.auth_token.encode('ascii')),
msg=message,
digestmod=hashlib.sha256)\
.hexdigest()
我会检查你的 auth_token
用于 self._bot_configuration.auth_token.encode('ascii')
。它是否包含非 ascii 字符?如果是,那么你有理由。 (例如)
尝试比较以下结果:
hmac.new(bytes(self._bot_configuration.auth_token.encode('ascii')),
msg=request.get_data(),
digestmod=hashlib.sha256).hexdigest()
至:
request.headers.get('X-Viber-Content-Signature')
这是不同的,这就是您收到禁止消息的原因。
您收到 403 错误有两个原因。要模拟来自 Viber 的 Webhook 请求,您必须发送 X-Viber-Content-Signature
header。此外,此值必须是使用身份验证令牌和 webhook 负载计算的 SHA256 哈希值,如 Callbacks.
下的 API 文档中所述
我相信你在这里有两个选择。如果您只想验证您的代码是否正确接收了 webhook,您可以暂时注释掉 verify_signature()
行。 Viber(或任何 webhook 源)不需要验证 webhook 请求。通常,开发人员会假设像 Viber 提供的库这样的库正确地测试了他们的代码,因此通常不需要再次测试他们的功能。您还可以考虑模拟该函数,因为在这种情况下这非常简单。
如果你真的想测试 Viber 的签名验证,那么你将需要实现我首先提到的 2 个原因。这基本上是您需要在测试 webhook 发送代码中执行的操作。请注意,我只在下面包含了您需要的新代码,请合并到您的其他测试代码中。
import json
import hmac
import hashlib
# Compute SHA256 hex digest signature using auth token and payload.
auth_token = 'xxx-xxx-xxx'
signature = hmac.new(
key=auth_token.encode('ascii'),
msg=data.encode('ascii'),
digestmod=hashlib.sha256
).hexdigest()
# Send test webhook request with computed signature in header.
response = requests.post(
webhook_url,
data=json.dumps(data),
headers={
'X-Viber-Content-Signature': signature,
'Content-Type': 'application/json'
},
verify='E:\Docs\learn_py\viberbot\certificate.pem'
)
请注意,@tukan 在 viber-bot-python 存储库中指出了 _calculate_message_signature()
函数,它显示了签名是如何计算的。
我有以下 HTTPS 服务器:
from flask import Flask, request, Response
from viberbot import Api
from viberbot.api.bot_configuration import BotConfiguration
from viberbot.api.messages import VideoMessage
from viberbot.api.messages.text_message import TextMessage
import logging
from viberbot.api.viber_requests import ViberConversationStartedRequest
from viberbot.api.viber_requests import ViberFailedRequest
from viberbot.api.viber_requests import ViberMessageRequest
from viberbot.api.viber_requests import ViberSubscribedRequest
from viberbot.api.viber_requests import ViberUnsubscribedRequest
logger = logging.getLogger(__name__)
app = Flask(__name__)
viber = Api(BotConfiguration(
name='PythonSampleBot',
avatar='http://www.clker.com/cliparts/3/m/v/Y/E/V/small-red-apple-hi.png',
auth_token='xxx-xxx-xxx'
))
@app.route('/', methods=['POST'])
def incoming():
logger.debug("received request. post data: {0}".format(request.get_data()))
# every viber message is signed, you can verify the signature using this method
if not viber.verify_signature(request.get_data(), request.headers.get('X-Viber-Content-Signature')):
return Response(status=403)
# this library supplies a simple way to receive a request object
viber_request = viber.parse_request(request.get_data())
if isinstance(viber_request, ViberMessageRequest):
message = viber_request.message
# lets echo back
viber.send_messages(viber_request.sender.id, [
message
])
elif isinstance(viber_request, ViberSubscribedRequest):
viber.send_messages(viber_request.get_user.id, [
TextMessage(text="thanks for subscribing!")
])
elif isinstance(viber_request, ViberFailedRequest):
logger.warn(
"client failed receiving message. failure: {0}".format(viber_request))
return Response(status=200)
def set_webhook(viber_bot):
viber_bot.set_webhook('https://xxx.xxx.xxx.xxx:4443')
logging.info("Web hook has been set")
if __name__ == "__main__":
context = ('certificate.pem', 'key.pem')
app.run(host='0.0.0.0', port=4443, debug=True, ssl_context=context)
正在尝试发送消息:
import json
import requests
webhook_url = 'https://xxx.xxx.xxx.xxx:4443'
data = {
"receiver": "xxx-xxx-xxx",
"type": "text",
"text": "Hello world!"
}
response = requests.post(
webhook_url, data=json.dumps(data),
headers={'Content-Type': 'application/json'},
verify='E:\Docs\learn_py\viberbot\certificate.pem'
)
if response.status_code != 200:
raise ValueError(
'Request returned an error %s, the response is:\n%s'
% (response.status_code, response.text)
)
我收到 403 错误
ValueError: Request returned an error 403, the response is:
更新:
403 来自:
if not viber.verify_signature(request.get_data(), request.headers.get('X-Viber-Content-Signature')):
return Response(status=403)
已编辑 由于更新。您在 verify_signature
.
verify_signature
的定义:
def verify_signature(self, request_data, signature):
return signature == self._calculate_message_signature(request_data)
您正在向那里发送 request.headers.get('X-Viber-Content-Signature')
作为“签名”。因此,您的解决方案是检查 __calculate_message_signature(request_data)
requiest_data = request.get_data() 你的情况。
_calculate_message_signature
的定义是:
def _calculate_message_signature(self, message):
return hmac.new(
bytes(self._bot_configuration.auth_token.encode('ascii')),
msg=message,
digestmod=hashlib.sha256)\
.hexdigest()
我会检查你的 auth_token
用于 self._bot_configuration.auth_token.encode('ascii')
。它是否包含非 ascii 字符?如果是,那么你有理由。 (例如)
尝试比较以下结果:
hmac.new(bytes(self._bot_configuration.auth_token.encode('ascii')),
msg=request.get_data(),
digestmod=hashlib.sha256).hexdigest()
至:
request.headers.get('X-Viber-Content-Signature')
这是不同的,这就是您收到禁止消息的原因。
您收到 403 错误有两个原因。要模拟来自 Viber 的 Webhook 请求,您必须发送 X-Viber-Content-Signature
header。此外,此值必须是使用身份验证令牌和 webhook 负载计算的 SHA256 哈希值,如 Callbacks.
我相信你在这里有两个选择。如果您只想验证您的代码是否正确接收了 webhook,您可以暂时注释掉 verify_signature()
行。 Viber(或任何 webhook 源)不需要验证 webhook 请求。通常,开发人员会假设像 Viber 提供的库这样的库正确地测试了他们的代码,因此通常不需要再次测试他们的功能。您还可以考虑模拟该函数,因为在这种情况下这非常简单。
如果你真的想测试 Viber 的签名验证,那么你将需要实现我首先提到的 2 个原因。这基本上是您需要在测试 webhook 发送代码中执行的操作。请注意,我只在下面包含了您需要的新代码,请合并到您的其他测试代码中。
import json
import hmac
import hashlib
# Compute SHA256 hex digest signature using auth token and payload.
auth_token = 'xxx-xxx-xxx'
signature = hmac.new(
key=auth_token.encode('ascii'),
msg=data.encode('ascii'),
digestmod=hashlib.sha256
).hexdigest()
# Send test webhook request with computed signature in header.
response = requests.post(
webhook_url,
data=json.dumps(data),
headers={
'X-Viber-Content-Signature': signature,
'Content-Type': 'application/json'
},
verify='E:\Docs\learn_py\viberbot\certificate.pem'
)
请注意,@tukan 在 viber-bot-python 存储库中指出了 _calculate_message_signature()
函数,它显示了签名是如何计算的。