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

Question

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

Answer 1

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