pythonwebsockethmacsignhashlib

How to sign OKEx API V5 login with websockets using HMAC SHA256 encryption and Base64 encoding?


Although this question has been answered for previous versions of the OKEx API using REST, it hasn't been for the latest version 5 of the API using websockets. The docs are here.

I am getting the following error {"event":"error","msg":"Invalid sign","code":"60007"} so there must be a problem with the signature string algorithm but I cannot seem to be able to identify where I am making a mistake.

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())

Solution

  • I figured it out. The signature needs to be converted back from bytes-string varible into string variable before it is sent.

    Adding the following line of code does this: 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())