reset.py

"""
Performs a factory reset on a Cisco IR800 router through a serial connection.
Automatically finds the correct USB port, logs in using environment variables,
runs write erase, cleans flash and nvram, and reloads the router. After the
reload, the router enters ROMMON-2 mode and fix_rommon.py must be run.
"""

import serial
import time
import os
import glob
import subprocess
import re
from dotenv import load_dotenv


# Load the router login credentials from the .env file
load_dotenv()

username = os.getenv("ROUTER_USERNAME")
password = os.getenv("ROUTER_PASSWORD")
enable_password = os.getenv("ROUTER_ENABLE_PASSWORD")


def close_screen_sessions():
    """
    Closes all active screen sessions to free the serial ports.
    Runs `screen -list` to check for active sessions.
    If any sessions are found, they are closed using `screen -X quit`.
    """

    result = subprocess.run(["screen", "-list"], capture_output=True, text=True)

    if "No Sockets found" not in result.stdout:
        subprocess.run(["screen", "-X", "quit"])
        time.sleep(1)
        print("Screen sessions closed")

    else:
        print("No screen sessions found")


def find_port():
    """
    Searches for and returns an active USB serial port.
    Loops until an active port is found by sending a newline
    and checking if the router responds. Retries every 5 seconds
    if no active port is found.

    Returns:
        str: The path to the active port, for example '/dev/ttyUSB0'.
    """

    print("Searching for USB port...")

    while True:
        ports = glob.glob("/dev/ttyUSB*")

        for port in ports:
            try:
                test = serial.Serial(port=port, baudrate=9600, timeout=2)
                test.write(b"\r\n")
                time.sleep(2)

                output = test.read(test.in_waiting).decode("utf-8", errors="ignore")

                test.close()

                if output:
                    print(f"Found active port: {port}")
                    return port

            except:
                continue

        print("No active port found, retrying in 5 seconds...")
        time.sleep(5)


def send_and_wait(command, wait=3):
    """
    Sends a command through the serial connection and waits for a response
    from the router. Reads the output until a known prompt or keyword
    is found, or until the 15 second timeout is reached.

    Arguments:
        command (str): The command to send to the router.
        wait (float): Seconds to wait before reading output. Default is 3.

    Returns:
        str: The decoded output from the router.
    """

    ser.write(f"{command}\r\n".encode())
    time.sleep(wait)

    output = ""
    deadline = time.time() + 15

    while time.time() < deadline:
        if ser.in_waiting:
            output += ser.read(ser.in_waiting).decode("utf-8", errors="ignore")

        time.sleep(0.2)

        if any(p in output for p in ["#", ">", "confirm", "save", "[OK]", "dialog", "complete", "erase", "yes/no", "filename", "rommon", "deleted", "No such", "Error"]):
            break

    return output


def login_and_enable():
    """
    Logs in to the router and activates enable mode.
    Attempts up to 10 times to log in using the username and password,
    then activates enable mode using the enable password.

    Returns:
        bool: True if enable mode was activated, otherwise False.
    """

    MAX_ATTEMPTS = 10

    for attempt in range(MAX_ATTEMPTS):

        print(f"Login attempt {attempt + 1} of {MAX_ATTEMPTS}...")
        ser.write(b"\r\n")
        time.sleep(6)

        output = ser.read(ser.in_waiting).decode("utf-8", errors="ignore")

        if "(config)" in output:
            send_and_wait("end")

        if "Username" in output:
            output = send_and_wait(username)

            if "Password" in output:
                output = send_and_wait(password)

        if ">" in output:
            output = send_and_wait("enable")

            if "Password" in output:
                output = send_and_wait(enable_password)

        if "(config)" in output:
            send_and_wait("end")

        if "#" in output:
            print("Enable mode activated!")
            return True

        print("Could not reach enable mode, retrying in 5 seconds...")
        time.sleep(5)

    print("Maximum number of attempts reached, exiting")
    return False


def clean_flash():
    """
    Cleans the flash memory and removes all files except protected system files.
    Lists the flash contents and removes all files that do not match
    the protected file list. Short numeric filenames are also skipped.
    """

    print("Cleaning flash...")
    ser.write(b"dir flash:\r\n")

    output = ""
    deadline = time.time() + 20

    while time.time() < deadline:
        if ser.in_waiting:
            output += ser.read(ser.in_waiting).decode("utf-8", errors="ignore")

        if "#" in output:
            break

        time.sleep(0.3)

    # Files that should never be deleted
    never_delete = [
        "eem",
        "managed",
        "ir800-universalk9",
        "ir800-hv",
        "ir800-uk9",
        "gemboa",
        "MC7304",
        "no_usb_emul"
    ]

    # Finds all files and folders in flash
    all_files = re.findall(r"(?:^|\s)([\w\-\.]+)(?:\s|$)", output, re.MULTILINE)

    for fil in all_files:
        if not fil or any(skip in fil for skip in never_delete):
            continue

        if re.match(r"^\d+$", fil) and len(fil) < 6:
            continue

        send_and_wait(f"delete /force /recursive flash:{fil}", wait=3)

    print("Flash cleaned!")


def clean_nvram():
    """
    Cleans nvram and removes all files.
    Lists the nvram contents and removes all files that are not
    short numeric filenames.
    """

    print("Cleaning nvram...")
    ser.write(b"dir nvram:\r\n")

    output = ""
    deadline = time.time() + 20

    while time.time() < deadline:
        if ser.in_waiting:
            output += ser.read(ser.in_waiting).decode("utf-8", errors="ignore")

        if "#" in output:
            break

        time.sleep(0.3)

    # Files that should never be deleted in nvram
    never_delete_nvram = []

    # Finds all files in nvram
    all_files = re.findall(r"(?:^|\s)([\w\-\.]+)(?:\s|$)", output, re.MULTILINE)

    for fil in all_files:
        if not fil or any(skip in fil for skip in never_delete_nvram):
            continue

        if re.match(r"^\d+$", fil) and len(fil) < 6:
            continue

        send_and_wait(f"delete /force nvram:{fil}", wait=3)

    print("Nvram cleaned!")


# Closes screen sessions, finds the port, and connects to the router
close_screen_sessions()
time.sleep(2)

port = find_port()

ser = serial.Serial(port=port, baudrate=9600, timeout=5)


# Exit if login fails
if not login_and_enable():
    ser.close()
    exit()


# Clears startup configuration
print("Running write erase...")

output = send_and_wait("write erase", wait=5)

if "confirm" in output.lower() or "erase" in output.lower():
    send_and_wait("", wait=8)

print("Write erase completed!")


# Cleans flash and nvram
clean_flash()
clean_nvram()


# Reloads the router without saving the configuration
print("Reloading the router...")

output = send_and_wait("reload", wait=5)

if "yes/no" in output.lower() or "save" in output.lower() or "modified" in output.lower():
    send_and_wait("no", wait=5)

if "confirm" in output.lower() or "proceed" in output.lower():
    send_and_wait("", wait=3)

if "confirm" in output.lower() or "proceed" in output.lower():
    send_and_wait("", wait=3)

print("Factory reset completed! The router is rebooting into ROMMON-2.")

print("Run fix_rommon.py to boot the router.")

ser.close()