A script to test VoIP connection

test_voip.py

import socket
import hashlib
import re
import uuid
import sys
import random

# Target Configuration
SIP_SERVER = "x.x.x.x"
SIP_PORT = 5060
SIP_LOGIN = "login"
SIP_PASSWORD = "pwd"

# Generate unique identifiers required for standard SIP headers
LOCAL_IP = socket.gethostbyname(socket.gethostname())
LOCAL_PORT = random.randint(10000, 60000)
CALL_ID = f"{uuid.uuid4()}@{LOCAL_IP}"
FROM_TAG = uuid.uuid4().hex[:16]

def create_register_payload(nonce=None, realm=None, cseq=1):
    """Generates a raw SIP REGISTER message string."""
    via = f"SIP/2.0/UDP {LOCAL_IP}:{LOCAL_PORT};rport;branch=z9hG4bK-{uuid.uuid4().hex[:8]}"
    from_header = f"<sip:{SIP_LOGIN}@{SIP_SERVER}>;tag={FROM_TAG}"
    to_header = f"<sip:{SIP_LOGIN}@{SIP_SERVER}>"
    contact = f"<sip:{SIP_LOGIN}@{LOCAL_IP}:{LOCAL_PORT}>"

    payload = (
        f"REGISTER sip:{SIP_SERVER} SIP/2.0\r\n"
        f"Via: {via}\r\n"
        f"Max-Forwards: 70\r\n"
        f"From: {from_header}\r\n"
        f"To: {to_header}\r\n"
        f"Call-ID: {CALL_ID}\r\n"
        f"CSeq: {cseq} REGISTER\r\n"
        f"Contact: {contact}\r\n"
        f"Expires: 3600\r\n"
        f"User-Agent: Python-Stdlib-SIP-Tester\r\n"
    )

    # Calculate and inject Digest MD5 response if a challenge was provided
    if nonce and realm:
        # HA1 = MD5(username:realm:password)
        ha1 = hashlib.md5(f"{SIP_LOGIN}:{realm}:{SIP_PASSWORD}".encode()).hexdigest()
        # HA2 = MD5(method:digestURI)
        ha2 = hashlib.md5(f"REGISTER:sip:{SIP_SERVER}".encode()).hexdigest()
        # Response = MD5(HA1:nonce:HA2)
        response = hashlib.md5(f"{ha1}:{nonce}:{ha2}".encode()).hexdigest()

        auth_header = (
            f'Authorization: Digest username="{SIP_LOGIN}", realm="{realm}", '
            f'nonce="{nonce}", uri="sip:{SIP_SERVER}", response="{response}"\r\n'
        )
        payload += auth_header

    payload += "Content-Length: 0\r\n\r\n"
    return payload

def parse_auth_challenge(response_text):
    """Extracts the cryptographic nonce and realm from a 401 response."""
    realm_match = re.search(r'realm="([^"]+)"', response_text)
    nonce_match = re.search(r'nonce="([^"]+)"', response_text)

    realm = realm_match.group(1) if realm_match else None
    nonce = nonce_match.group(1) if nonce_match else None
    return nonce, realm

def test_sip_login():
    print(f"Initializing standard UDP socket on port {LOCAL_PORT}...")

    # Open standard library UDP Socket
    sock = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
    sock.settimeout(5.0) # Stop hanging if the server drops packets

    try:
        # Step 1: Send Initial REGISTER to provoke 401 Challenge
        print("Sending initial unauthenticated REGISTER...")
        initial_payload = create_register_payload(cseq=1)
        sock.sendto(initial_payload.encode(), (SIP_SERVER, SIP_PORT))

        data, _ = sock.recvfrom(4096)
        response_1 = data.decode(errors='ignore')

        if "401 Unauthorized" not in response_1:
            if "200 OK" in response_1:
                print(" Success: Server logged in immediately without a challenge!")
                return True
            print(f" Error: Server rejected request with unexpected message:\n{response_1.splitlines()[0]}")
            return False

        print("Received 401 Unauthorized challenge. Processing MD5 calculations...")
        nonce, realm = parse_auth_challenge(response_1)

        if not nonce or not realm:
            print(" Error: Failed to extract validation variables (nonce/realm) from server.")
            return False

        # Step 2: Send authenticated REGISTER with cryptographic signature
        print("Sending authenticated REGISTER...")
        auth_payload = create_register_payload(nonce=nonce, realm=realm, cseq=2)
        sock.sendto(auth_payload.encode(), (SIP_SERVER, SIP_PORT))

        data, _ = sock.recvfrom(4096)
        response_2 = data.decode(errors='ignore')

        if "200 OK" in response_2:
            print(" Success: Authentication passed!")
            return True
        elif "403" in response_2 or "401" in response_2:
            print(" Error: Authentication failed (Invalid password).")
            return False
        else:
            print(f" Error: Unexpected final response status:\n{response_2.splitlines()[0]}")
            return False

    except socket.timeout:
        print(" Error: Socket timeout. The SIP server or network firewall is ignoring our traffic.")
        return False
    except Exception as e:
        print(f" Unexpected script error: {e}")
        return False
    finally:
        sock.close()

if __name__ == "__main__":
    success = test_sip_login()
    sys.exit(0 if success else 1)