Python Websockets and connecting to ws-data-stream problem

[mikkkogu]

So I'm trying to connect websocket stream, but getting error from camera:

! failing connection with code 1006
x half-closing TCP connection
= connection is CLOSED
server rejected WebSocket connection: HTTP 426

MyCODE:

import requests, re
import websockets
import asyncio
import hashlib
import base64
import uuid
import logging
logger = logging.getLogger('websockets')
logger.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler())

def calculate_digest_response(username, realm, password, uri, method, nonce, nc, cnonce):
    ha1 = hashlib.md5(f"{username}:{realm}:{password}".encode()).hexdigest()
    ha2 = hashlib.md5(f"{method}:{uri}".encode()).hexdigest()
    response = hashlib.md5(
        f"{ha1}:{nonce}:{nc}:{cnonce}:auth:{ha2}".encode()).hexdigest()
    return response


async def connect(nonce):
    username = 'root'
    password = 'mypassword'
    realm = 'AXIS_SOMETHING'
    uri = '/vapix/ws-data-stream?sources=events'
    method = 'GET'

    nc = '00000001'
    cnonce = str(uuid.uuid4())

    response = calculate_digest_response(
        username, realm, password, uri, method, nonce, nc, cnonce)
    print("Digest calculated response:", response, "\n")
    authorization_header = f'Digest username="{username}", realm="{realm}", nonce="{nonce}", uri="{uri}", response="{response}", qop=auth, nc={nc}, cnonce="{cnonce}"'
    print(" ---- Digest header ---- \n"+authorization_header+"\n")
    websocket_uri = "ws://myipaddress/vapix/ws-data-stream?sources=events"
    websocket_headers = {

        "Authorization": authorization_header,
        
        "Origin": "Python",
        "User-Agent": "Python Websockets",
        "Accept-Encoding": "gzip, deflate",
        "Accept-Language": "et,en-US;q=0.9,en;q=0.8",
        "Sec-GPC": "1",
        "Sec-WebSocket-Extensions": "permessage-deflate; client_max_window_bits",
        "Upgrade": "websocket"
    }
    
    try:
        async with websockets.connect(websocket_uri, extra_headers=websocket_headers, user_agent_header=None) as websocket:
            print("WebSocket connection established.")

            # Prepare the JSON payload
            payload = {
                "apiVersion": "1.0",
                "method": "events:configure",
                "params": {
                    "eventFilterList": [
                        {
                            "topicFilter": "tnsaxis:CameraApplicationPlatform/FenceGuard/Camera1ProfileANY"
                        }
                    ]
                }
            }

            # Convert the payload to JSON string
            payload_json = json.dumps(payload)

            # Send the JSON payload
            await websocket.send(payload_json)
            print("JSON payload sent.")

            # Continue with WebSocket communication
            while True:
                message = await websocket.recv()
                print(f"Received message: {message}")
    except Exception as e:
        print(e)        


def extract_value(header, key):
    match = re.search(fr'{key}="([^"]+)"', header)
    if match:
        return match.group(1)
    return None

def perform_http_request():
    username = 'root'
    password = 'mypassword'
    url = 'http://myipaddress/axis-cgi/login.cgi'

    # Make a GET request to obtain the Digest authentication headers
    response = requests.get(url)
    #print(response.text)
    # Extract the required headers from the response
    nonce = response.headers.get('WWW-Authenticate')
    #print(nonce)
    match = re.search(r'nonce="([^"]+)"', nonce)
    if match:
        nonce = match.group(1)
        print("Nonce from response:", nonce,"\n")
    else:
        print("Nonce not found in the string.")
    response = requests.get(url, auth=requests.auth.HTTPDigestAuth(username, password))
    # Continue with the WebSocket connection using the obtained headers
    asyncio.run(connect(nonce))


if __name__ == "__main__":
    perform_http_request()

Can somebody help. Or is there any easy way get data from rstp stream?

[vivekatoffice]

Hi @mikkkogu ,
Thank you so much for sharing your sample code here. After few changes in code, I was able to get the results. But I really appreciate your work and effort you put for Digest authentication for WebSockets in Python. Even on Websocket Documentation they mentioned it is not supported for client.

Python code

Please note: realm = 'AXIS_B8XXXXXXXXXX' #B8XXXXXXXXXX is the the MAC address of the camera.
Check device properties

import json
import requests, re
import websockets
import asyncio
import hashlib
import base64
import uuid
import logging
logger = logging.getLogger('websockets')
logger.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler())

def calculate_digest_response(username, realm, password, uri, method, nonce, nc, cnonce):
    ha1 = hashlib.md5(f"{username}:{realm}:{password}".encode()).hexdigest()
    ha2 = hashlib.md5(f"{method}:{uri}".encode()).hexdigest()
    response = hashlib.md5(
        f"{ha1}:{nonce}:{nc}:{cnonce}:auth:{ha2}".encode()).hexdigest()
    return response


async def connect(nonce):
    username = 'root'
    password = 'password'
    realm = 'AXIS_B8XXXXXXXXXX'  #B8XXXXXXXXXX is the the MAC address of the camera.
    uri = '/vapix/ws-data-stream?sources=events'
    method = 'GET'

    nc = '00000001'
    cnonce = str(uuid.uuid4())

    response = calculate_digest_response(
        username, realm, password, uri, method, nonce, nc, cnonce)
    print("Digest calculated response:", response, "\n")
    authorization_header = f'Digest username="{username}", realm="{realm}", nonce="{nonce}", uri="{uri}", response="{response}", qop=auth, nc={nc}, cnonce="{cnonce}"'
    print(" ---- Digest header ---- \n"+authorization_header+"\n")
    websocket_uri = "ws://192.168.2.13/vapix/ws-data-stream?sources=events"
    websocket_headers = {
        "Authorization": authorization_header
    }
    
    try:
        async with websockets.connect(websocket_uri, extra_headers=websocket_headers, user_agent_header=None) as websocket:
            print("WebSocket connection established.")

            # Prepare the JSON payload
            payload = {
                    "apiVersion": "1.0",
                    "method": "events:configure",
                    "params": {
                            "eventFilterList":[
                            {
                            "topicFilter":"tns1:Device/tnsaxis:IO/VirtualPort"
                            }
                             ]
                            }
                        }
           

            # Convert the payload to JSON string
            payload_json = json.dumps(payload)

            # Send the JSON payload
            await websocket.send(payload_json)
            print("JSON payload sent.")

            # Continue with WebSocket communication
            while True:
                message = await websocket.recv()
                print(f"Received message: {message}")
    except Exception as e:
        print(e)        


def extract_value(header, key):
    match = re.search(fr'{key}="([^"]+)"', header)
    if match:
        return match.group(1)
    return None

def perform_http_request():
    username = 'root'
    password = 'password'
    url = 'http://192.168.2.13/axis-cgi/login.cgi'

    # Make a GET request to obtain the Digest authentication headers
    response = requests.get(url)
    #print(response.text)
    # Extract the required headers from the response
    nonce = response.headers.get('WWW-Authenticate')
    #print(nonce)
    match = re.search(r'nonce="([^"]+)"', nonce)
    if match:
        nonce = match.group(1)
        print("Nonce from response:", nonce,"\n")
    else:
        print("Nonce not found in the string.")
    response = requests.get(url, auth=requests.auth.HTTPDigestAuth(username, password))
    # Continue with the WebSocket connection using the obtained headers
    asyncio.run(connect(nonce))


if __name__ == "__main__":
    perform_http_request()

Output

[janssen70]

Hi,

I made a few changes to @vivekatoffice 's version:

• Also get the realm from the Authentication-header
• Get IP address and credentials from commandline

So now it works for every supporting device without changing the code

import json
import requests, re
import websockets
import asyncio
import hashlib
import base64
import uuid
import logging
import argparse

logger = logging.getLogger('websockets')
logger.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler())

def calculate_digest_response(username, realm, password, uri, method, nonce, nc, cnonce):
    ha1 = hashlib.md5(f"{username}:{realm}:{password}".encode()).hexdigest()
    ha2 = hashlib.md5(f"{method}:{uri}".encode()).hexdigest()
    response = hashlib.md5(
        f"{ha1}:{nonce}:{nc}:{cnonce}:auth:{ha2}".encode()).hexdigest()
    return response


async def connect(device, username, password, nonce, realm):
    uri = '/vapix/ws-data-stream?sources=events'
    method = 'GET'

    nc = '00000001'
    cnonce = str(uuid.uuid4())

    response = calculate_digest_response(
        username, realm, password, uri, method, nonce, nc, cnonce)
    print("Digest calculated response:", response, "\n")
    authorization_header = f'Digest username="{username}", realm="{realm}", nonce="{nonce}", uri="{uri}", response="{response}", qop=auth, nc={nc}, cnonce="{cnonce}"'
    print(f' ---- Digest header ---- \n{authorization_header}\n')
    websocket_uri = f'ws://{device}/vapix/ws-data-stream?sources=events'
    websocket_headers = {
        "Authorization": authorization_header
    }
    
    try:
        async with websockets.connect(websocket_uri, extra_headers=websocket_headers, user_agent_header=None) as websocket:
            print("WebSocket connection established.")

            # Prepare the JSON payload
            payload = {
                    "apiVersion": "1.0",
                    "method": "events:configure",
                    "params": {
                            "eventFilterList":[
                            {
                            "topicFilter":"tns1:Device/tnsaxis:IO/VirtualPort"
                            }
                             ]
                            }
                        }
           

            # Convert the payload to JSON string
            payload_json = json.dumps(payload)

            # Send the JSON payload
            await websocket.send(payload_json)
            print("JSON payload sent.")

            # Continue with WebSocket communication
            while True:
                message = await websocket.recv()
                print(f"Received message: {message}")
    except Exception as e:
        print(e)        


def extract_value(header, key):
    match = re.search(fr'{key}="([^"]+)"', header)
    if match:
        return match.group(1)
    return None

def perform_http_request(device, username, password):
    url = f'http://{device}/axis-cgi/login.cgi'

    # Make a GET request to obtain the Digest authentication headers
    response = requests.get(url)
    #print(response.text)
    # Extract the required headers from the response
    nonce = None
    realm = None
    auth_info = response.headers.get('WWW-Authenticate')
    match = re.search(r'nonce="([^"]+)"', auth_info)
    if match:
        nonce = match.group(1)
        print("Nonce from response:", nonce)
    match = re.search(r'realm="([^"]+)"', auth_info)
    if match:
        realm = match.group(1)
        print("Realm from response:", realm)
    if nonce and realm:
        response = requests.get(url, auth=requests.auth.HTTPDigestAuth(username, password))
        # Continue with the WebSocket connection using the obtained headers
        asyncio.run(connect(device, username, password, nonce, realm))
    else:
       print('Failed to obtain authentication prerequisites')

if __name__ == "__main__":
    
    parser = argparse.ArgumentParser(
       description='Websocket test'
    )
 
    parser.add_argument(
         '-c', '--camera', type=str, default='192.168.2.13',
         help='Hostname/IP address of the device(s) to interact with')
    parser.add_argument(
         '-u', '--user', type=str, default='root',
         help='username to login with (root)')
    parser.add_argument(
         '-p', '--password', type=str, default='pass',
         help='password to login with (pass)')
 
    args = parser.parse_args()
    perform_http_request(args.camera, args.user, args.password)