如何使用HMAC SHA256加密和Base64编码通过Websockets对OKEx API V5登录进行签名？

Question

虽然这个问题已经在使用 REST 的 OKEx API 的早期版本中得到了解答，但在使用 websockets 的 API 的最新版本 5 中还没有得到解答。 文档在这里。

我收到以下错误

{"event":"error","msg":"Invalid sign","code":"60007"}

，所以签名字符串算法一定有问题，但我似乎无法识别我在哪里犯了错误。

import hmac
import json
import time
import hashlib
import asyncio
import websockets

passphrase = "XXXX"
secret_key = b"XXXX"
api_key = "XXXX"

timestamp = int(time.time())
print("timestamp: " + str(timestamp))
sign = str(timestamp) + 'GET' + '/users/self/verify'
total_params = bytes(sign, encoding= 'utf-8')
signature = hmac.new(bytes(secret_key, encoding= 'utf-8'), total_params, digestmod=hashlib.sha256).digest()
signature = base64.b64encode(signature)
print("signature = {0}".format(signature))

async def main():
    msg = \
    {
      "op": "login",
      "args": [
        {
          "apiKey": f'{api_key}',
          "passphrase": f'{passphrase}',
          "timestamp": f'{timestamp}',
          "sign": f'{signature}'
        }
      ]
    }

    async with websockets.connect('wss://wspap.okx.com:8443/ws/v5/private?brokerId=9999') as websocket:
        print(msg)
        await websocket.send(json.dumps(msg))
        response = await websocket.recv()
        print(response)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Answer 1

我想通了。在发送之前，签名需要从字节字符串变量转换回字符串变量。

添加以下代码行即可实现此目的：

signature = str(signature, 'utf-8')

import json
import time
import hmac
import hashlib
import base64
import asyncio
import websockets

passphrase = "XXXX"
secret_key = "XXXX"
api_key = "XXXX"

timestamp = int(time.time())
print("timestamp: " + str(timestamp))
sign = str(timestamp) + 'GET' + '/users/self/verify'
total_params = bytes(sign, encoding= 'utf-8')
signature = hmac.new(bytes(secret_key, encoding= 'utf-8'), total_params, digestmod=hashlib.sha256).digest()
signature = base64.b64encode(signature)
signature = str(signature, 'utf-8')

print("signature = {0}".format(signature))

async def main():
    msg = \
    {
      "op": "login",
      "args": [
        {
          "apiKey": f'{api_key}',
          "passphrase": f'{passphrase}',
          "timestamp": f'{timestamp}',
          "sign": f'{signature}'
        }
      ]
    }

    async with websockets.connect('wss://wspap.okx.com:8443/ws/v5/private?brokerId=9999') as websocket:
        print(msg)
        await websocket.send(json.dumps(msg))
        response = await websocket.recv()
        print(response)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Answer 2

您可以使用他们 API 中的函数

import hmac, hashlib, base64
import time

def initLoginParams(apiKey, passphrase, secretKey):
    timestamp = time.time()
    message = str(timestamp) + 'GET' + '/users/self/verify'
    mac = hmac.new(bytes(secretKey, encoding='utf8'), bytes(message, encoding='utf-8'), digestmod='sha256')
    d = mac.digest()
    sign = base64.b64encode(d)
    arg = {"apiKey": apiKey, "passphrase": passphrase, "timestamp": timestamp, "sign": sign.decode("utf-8")}
    payload = {"op": "login", "args": [arg]}
    return payload