我正在尝试通过 twitter 文档中给出的 curl 命令在 twitter 上将 url 注册为 webhook。我认为 twitter 文档有点过时或不正确。
curl --request POST --url "https://api.twitter.com/1.1/account_activity/all/prod/webhooks.json?url=https%3A%2F%2FXXXX.com%2Fwebhook%2Ftwitter" --header "authorization: OAuth oauth_consumer_key='XXXXXXXXXXXXXXXXXXXXXX', oauth_nonce='ODgyNjc5NjQ0MTM3NzI4NTcwMjY4NDQ0', oauth_signature='7daMyzB1JClE4xv8hXNCimWpGtA%3D', oauth_signature_method='HMAC-SHA1', oauth_timestamp='1568620293', oauth_token='XXXXXXXXXXXXXXXXXXXXX', oauth_version='1.0'"
运行此代码后，我收到错误消息
{"errors":[{"code":215,"message":"Bad Authentication data."}]}
下面给出了我用来生成时间戳、oauth_nonce、oauth_signature 的代码。这些代码在 python3 中。我不太确定他们是否给出了正确的输出。
oauth_timestamp = str(int(time.time()))
#oauth_nonce
def get_nonce():
nonce = ''.join(random.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits)for x in range(32))
return nonce
#oauth_signature
def escape(s):
"""Percent Encode the passed in string"""
return urllib.parse.quote_plus(s, safe='~')
def create_auth_header(parameters):
"""For all collected parameters, order them and create auth header"""
ordered_parameters = {}
ordered_parameters = collections.OrderedDict(sorted(parameters.items()))
auth_header = (
'%s="%s"' % (k, v) for k, v in ordered_parameters.items())
val = "OAuth " + ', '.join(auth_header)
return val
def generate_signature(method, url, url_parameters, oauth_parameters, oauth_consumer_key, oauth_consumer_secret, oauth_token_secret=None, status=None):
"""Create the signature base string"""
#Combine parameters into one hash
temp = collect_parameters(oauth_parameters, status, url_parameters)
#Create string of combined url and oauth parameters
parameter_string = stringify_parameters(temp)
#Create your Signature Base String
signature_base_string = (
method.upper() + '&' +
escape(str(url)) + '&' +
escape(parameter_string)
)
#Get the signing key
signing_key = create_signing_key(oauth_consumer_secret, oauth_token_secret)
return calculate_signature(signing_key, signature_base_string)
def stringify_parameters(parameters):
"""Orders parameters, and generates string representation of parameters"""
output = ''
ordered_parameters = {}
ordered_parameters = collections.OrderedDict(sorted(parameters.items()))
counter = 1
for k, v in ordered_parameters.items():
output += escape(str(k)) + '=' + escape(str(v))
if counter < len(ordered_parameters):
output += '&'
counter += 1
return output
def collect_parameters(oauth_parameters, status, url_parameters):
"""Combines oauth, url and status parameters"""
#Add the oauth_parameters to temp hash
temp = oauth_parameters.copy()
#Add the status, if passed in. Used for posting a new tweet
if status is not None:
temp['status'] = status
#Add the url_parameters to the temp hash
for k, v in url_parameters.items():
temp[k] = v
return temp
def calculate_signature(signing_key, signature_base_string):
"""Calculate the signature using SHA1"""
hashed = hmac.new(signing_key, signature_base_string.encode('utf-8'), hashlib.sha1)
sig = binascii.b2a_base64(hashed.digest())[:-1]
return escape(sig)
def create_signing_key(oauth_consumer_secret, oauth_token_secret):
"""Create key to sign request with"""
signing_key = escape(oauth_consumer_secret) + '&'
signing_key += escape(oauth_token_secret)
return signing_key.encode('utf-8')
oauth_parameters = {
'oauth_timestamp': str(int(time.time())),
'oauth_signature_method': "HMAC-SHA1",
'oauth_version': "1.0",
'oauth_token': "vvvvvvvvvvvvvvv",
'oauth_nonce': get_nonce(),
'oauth_consumer_key': 'bbbbbbbbbbbbbbb'
}
oauth_parameters['oauth_signature'] = generate_signature(
method,
url,
url_parameters, oauth_parameters,
consumer_key,
consumer_secret,
access_token_secret
)
auth_headers = {'Authorization': create_auth_header(oauth_parameters),
'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8'
}
auth_data = {
'grant_type': 'client_credentials'
}
webhook_endpoint ='https%3A%2F%2Fmy-site.com%2Fwebhook%2Ftwitter'
url = 'https://api.twitter.com/1.1/account_activity/all/env/webhooks.json?url={}'.format(webhook_endpoint)
method = "post"
url_parameters = {
'exclude_replies': 'true'
}
r = requests.post(url, headers=auth_headers, data=auth_data)
请告诉我哪个代码不正确或者我做错了什么？如果您可以在答案中编写以下代码，那将是有益的。