sn32 dumper

sn32_dump.py

import argparse
import time
from telnetlib import Telnet
from contextlib import closing
from pathlib import Path

def auto_int(x):
    return int(x, 0)

def divisible_by_4(x):
    x = int(x, 0)
    if x % 4 != 0:
        raise argparse.ArgumentTypeError("The value 0x{:X} is not divisible by 4".format(x))
    return x

def send_command(tn, command, expected_response, timeout=5):
    try:
        command_bytes = command.encode('ascii') if isinstance(command, str) else command
        tn.read_until(b"> ", timeout=1)  # Wait for the prompt, increase timeout if needed
        tn.write(command_bytes + b"\n")

        # Read the response
        response = tn.read_until(expected_response, timeout=timeout)
        if not response.endswith(expected_response):
            print("Error: Unexpected response. Expected '{}'".format(expected_response.decode('ascii')))
            raise ValueError("Unexpected response")

    except EOFError:
        print("Telnet connection closed unexpectedly.")
        raise

def read_memory(tn, address, count, args):
    word_size = args.word_size
    # Format the address as '0xXXXXXXXX' notation
    address_hex = format(address, '#010X')

    # Gadget magic
    send_command(tn, "reg pc 0x{:X}".format(args.ldr_gadget), b"> ")
    send_command(tn, "reg {} 0x{:X}".format(args.reg2, address), b"> ")
    send_command(tn, "reg {} 0x{:X}".format(args.reg1, address), b"> ")

    # Format the mdw command
    command = "mdw {} {}".format(address_hex, count//word_size)
    tn.write(command.encode('ascii') + b"\n")
    response = tn.read_until(b"> ", timeout=5).decode("ascii")

    print("\rCommand: {} | Response: {}".format(command, response))

    # Check if the response indicates an error
    if "Error" in response:
        print("Error reading memory at address {}. Skipping.".format(address))
        return None

    # Extract and parse the hexadecimal values
    read_values = []
    lines = response.splitlines()
    for line in lines:
        parts = line.split(":")
        if len(parts) == 2:
            hex_values = parts[1].split()
            for val in hex_values:
                try:
                    read_values.append(val)
                except ValueError:
                    print("Error parsing memory value '{}'. Skipping.".format(val))

    if not read_values:
        print("No valid memory values found. Skipping.")
        return None

    return read_values

def verify_and_write_segment(tn, addr, chunk_size, args, outf):
    while True:
        # Read the memory segment
        print("\nReading memory at address {}.. ".format(addr))
        values = read_memory(tn, addr, chunk_size, args)
        if values is None:
            continue  # Retry reading on error

        try:
            # Verify the segment by reading it again
            print("\rVerifying memory at address {}.. ".format(addr))
            verification_values = read_memory(tn, addr, chunk_size, args)
            if verification_values is None:
                continue  # Retry reading on error

            # Compare the original and verification values
            if values != verification_values:
                print("\nVerification mismatch at address {}. Retrying...".format(addr))
                continue  # Retry reading on mismatch

            if values == verification_values:
                print("\rVerification matches at address {}. Continuing...".format(addr))
                #print(f"Read values: {values[:chunk_size]}")
                
                # Write the entire hex string to the file without spaces
                hex_string = ''.join(values[:chunk_size])
                outf.write(bytes.fromhex(hex_string))

        except (ValueError, IndexError):
            print("\nError reading memory at address {}. Skipping.".format(addr))
            continue

        break  # Break out of the loop if the verification is successful

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("start", type=divisible_by_4, help="start address to dump")
    parser.add_argument("size", type=divisible_by_4, help="size to dump")
    parser.add_argument("file", help="output file")
    parser.add_argument("--openocd", required=True, help="host:port of openocd telnet")
    parser.add_argument("--ldr-gadget", type=auto_int, required=True,
                        help="address of a gadget of format ldr Reg1, [Reg2]")
    parser.add_argument("--reg1", required=True, help="register for Reg1 (e.g. r0)")
    parser.add_argument("--reg2", required=True, help="register for Reg2 (e.g. r1)")
    parser.add_argument("--word-size", type=int, default=4, help="word size in bytes")
    parser.add_argument("--chunk-size", type=int, default=256, help="size of each memory read chunk in bytes")

    args = parser.parse_args()
    host, port = args.openocd.split(":")
    port = int(port)

    # set thumb bit
    args.ldr_gadget |= 1

    # Calculate adjusted chunk-size
    adjusted_chunk_size = args.chunk_size
    while adjusted_chunk_size > args.size:
        adjusted_chunk_size -= 4

    print("Adjusted chunk-size to be less than or equal to the requested size. Original chunk-size: {}, Adjusted chunk-size: {}".format(args.chunk_size, adjusted_chunk_size))
    args.chunk_size = adjusted_chunk_size

    output_path = Path(args.file)

    with closing(Telnet(host, port)) as tn, output_path.open("wb") as outf:
        send_command(tn, b"reset halt\n", b"")
        time.sleep(1)  # Add a delay after reset halt

        start_time = time.time()
        first_response_received = False

        for addr in range(args.start, args.start + args.size, args.chunk_size):
            chunk_size = min(args.chunk_size, args.start + args.size - addr)

            # Verify and write the memory segment
            verify_and_write_segment(tn, addr, chunk_size, args, outf)

            processed_bytes = addr - args.start + chunk_size
            elapsed_time = time.time() - start_time
            bytes_per_second = processed_bytes / elapsed_time

            remaining_bytes = args.size - processed_bytes
            remaining_time = remaining_bytes / bytes_per_second

            # Display the processing output with a static time estimate
            print("\rProcessing: [{}/{}] | Time remaining: {:.0f}m {:.0f}s".format(
                processed_bytes, args.size, remaining_time // 60, remaining_time % 60), end="")

            # Calculate and display time estimation after the first response
            if not first_response_received:
                print("\nEstimated time remaining: {:.0f}m {:.0f}s".format(
                    remaining_time // 60, remaining_time % 60))
                first_response_received = True

            if addr + chunk_size >= args.start + args.size:
                break

        print("\nDump completed.")

    # Verify the firmware file size
    with output_path.open("rb") as firmware_file:
        firmware_content = firmware_file.read()

    if len(firmware_content) != args.size:
        print("Warning: Firmware file size does not match the requested size.")
    else:
        print("Firmware file size matches the requested size.")

    # Verify the firmware file content
    expected_flag = b'\x55\x55\xAA\xAA'
    if firmware_content.endswith(expected_flag):
        print("Firmware verification passed.")
    else:
        print("Warning: Firmware verification failed. The expected flag is not present at the end of the file.")

if __name__ == "__main__":
    main()