Files
es24n-conf/modules/workflow_serial.py
scott d8a74f02dc Prompt for IOM1 or IOM2 in serial configure workflow
Previously hardcoded to IOM1. Now asks which IOM the serial cable is
connected to after login, and threads that choice through all Redfish
paths in fetch_current_config and collect_network_config. apply_configuration
and print_summary already used cfg.iom1.iom for paths so needed no changes.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-16 11:39:34 -04:00

640 lines
23 KiB
Python

"""
workflow_serial.py — Serial-based ES24N IOM network configuration workflow.
Connects via USB serial cable, logs in to the IOM console, and configures
network settings by running curl commands through the shell session.
The IOM's Redfish API is only accessible locally at https://127.0.0.1 from
within the IOM's own shell — not directly from the host over the serial cable.
All Redfish operations are therefore performed by issuing curl commands over
the serial connection and parsing the JSON responses.
"""
import glob
import json
import os
import re
import subprocess
import time
from typing import Optional
from models import IOMConfig, ShelfConfig
from serial_port import SerialPort
from ui import (
_c, C,
banner, rule, draw_table, draw_box,
info, ok, warn, error,
prompt, prompt_ip, prompt_yn, prompt_password,
)
# Strip ANSI escape sequences from serial terminal output.
# Catches CSI sequences (ESC[...X) and other two-character ESC sequences.
_ANSI_RE = re.compile(r'\x1b(?:\[[0-9;]*[A-Za-z]|[^[])')
def _at_shell_prompt(text: str) -> bool:
"""
Return True if any line in text looks like a root shell prompt.
Checks whether a non-empty line ends with '#' or '# ' — the pattern
produced by prompts like 'root@hostname:~#' — without being tripped
up by the word 'login' appearing elsewhere in the output (e.g. in
'Last login: ...' messages shown after a session is resumed).
"""
for line in text.splitlines():
stripped = line.rstrip()
if stripped.endswith("#") or stripped.endswith("# "):
return True
return False
def _extract_serial_from_hostname(hostname: str) -> Optional[str]:
"""
Extract the BMC serial number from an IOM hostname.
The serial number (which doubles as the root password) is a
dash-separated segment matching MXE followed by alphanumerics.
e.g. 'ves-ves-vds2249r-MXE3000048LHA03C-mgr1''MXE3000048LHA03C'
Returns None if no matching segment is found.
"""
for segment in hostname.split("-"):
if re.match(r"^MXE[A-Z0-9]+$", segment, re.IGNORECASE):
return segment
return None
def _detect_password_from_shell(ser: SerialPort) -> Optional[str]:
"""
When already logged in, run 'hostname' on the IOM shell and extract
the BMC serial number from the output.
"""
ser.send_line("hostname", delay=0.3)
out = _ANSI_RE.sub("", ser.read_until_quiet(quiet_period=0.5))
for line in out.splitlines():
pw = _extract_serial_from_hostname(line.strip())
if pw:
return pw
return None
# ── Serial Redfish transport ───────────────────────────────────────────────────
def _login_serial_console(ser: SerialPort) -> tuple:
"""
Perform the root login sequence on the IOM serial console.
Auto-detects the root password from the IOM hostname — the BMC serial
number (e.g. MXE3000048LHA03C) is embedded in both the login prompt
hostname and the output of 'hostname', so no manual entry is needed.
- Login prompt present: extracts serial from 'hostname login:' line.
- Already logged in: runs 'hostname' on the shell to extract the serial.
- Falls back to prompting the user if auto-detection fails either way.
Returns (success: bool, password: str).
"""
info("Logging in to IOM serial console...")
# Send a newline to get the current state of the console
ser.send_line("", delay=0.5)
response = _ANSI_RE.sub("", ser.read_until_quiet(quiet_period=0.5))
# Already at a shell prompt — extract password via 'hostname' command
if _at_shell_prompt(response):
ok("Already at shell prompt.")
password = _detect_password_from_shell(ser)
if password:
ok(f"Password auto-detected: {_c(C.BOLD, password)}")
return True, password
warn("Could not auto-detect password from hostname.")
return True, prompt_password()
low = response.lower()
# Try to extract the password from the login prompt hostname before
# sending credentials — format: 'ves-ves-model-SERIAL-mgr1 login:'
password = None
login_match = re.search(r"(\S+)\s+login:", response, re.IGNORECASE)
if login_match:
password = _extract_serial_from_hostname(login_match.group(1))
if password:
ok(f"Password auto-detected from login prompt: {_c(C.BOLD, password)}")
# Send username if we see a login prompt (or an empty/unknown response)
if "login" in low or not response.strip():
ser.send_line("root", delay=0.5)
response = _ANSI_RE.sub("", ser.read_until_quiet(quiet_period=0.5))
# Fall back to asking the user if auto-detection failed
if not password:
warn("Could not auto-detect password from login prompt.")
password = prompt_password()
# Send password
if "password" in response.lower():
ser.send_line(password, delay=0.5)
response = _ANSI_RE.sub("", ser.read_until_quiet(quiet_period=2.0))
if _at_shell_prompt(response):
ok("Logged in to IOM console.")
return True, password
error(f"Login failed. Console response: {response.strip()[:120]}")
return False, ""
def _serial_redfish_request(ser: SerialPort, password: str, method: str,
path: str, payload: Optional[dict] = None) -> tuple:
"""
Issue a Redfish request by running curl on the IOM's serial console.
The Redfish API is available at https://127.0.0.1 from within the IOM shell.
Appends -w '\\nHTTP_CODE:%{http_code}' to every curl call so the HTTP
status can be parsed independently of the response body.
Returns (success: bool, data: dict | str).
"""
url = f"https://127.0.0.1{path}"
cmd = f"curl -sk -u root:{password} -X {method}"
if payload:
body = json.dumps(payload, separators=(",", ":"))
cmd += f" -H 'Content-Type: application/json' -d '{body}'"
cmd += f" -w '\\nHTTP_CODE:%{{http_code}}' '{url}'"
ser.send_line(cmd, delay=0.3)
raw = _ANSI_RE.sub("", ser.read_until_quiet(quiet_period=1.5, timeout=30.0))
# ── Parse HTTP status code ─────────────────────────────────────────────────
# Use the LAST occurrence of HTTP_CODE: so the echoed curl -w argument
# (which also contains the literal text "HTTP_CODE:") doesn't interfere.
http_code = 0
if "HTTP_CODE:" in raw:
try:
http_code = int(raw.split("HTTP_CODE:")[-1].strip()[:3])
except ValueError:
pass
# ── Extract JSON body ──────────────────────────────────────────────────────
# The terminal echoes the curl command before executing it, and the
# -w format string contains '%{http_code}' which includes a literal '{'.
# When the terminal wraps long command lines, that '{' can land at a
# line boundary, causing find("{") to return the wrong position.
#
# Strategy: narrow the search to the text before HTTP_CODE: (the actual
# curl response body appears there), then find the LAST newline-prefixed
# '{' — because the JSON response always starts on its own line, while
# the echoed '{http_code}' is embedded mid-line in the curl command echo.
# Use rfind (not find) — the -w argument in the echoed curl command also
# contains the literal text "HTTP_CODE:", so find() would land there instead
# of at the actual HTTP_CODE:200 output that curl appends at the end.
http_code_pos = raw.rfind("HTTP_CODE:")
search_area = raw[:http_code_pos].rstrip() if http_code_pos >= 0 else raw
data: dict = {}
json_end = search_area.rfind("}") + 1
if json_end > 0:
json_start = -1
for nl in ("\r\n", "\n"):
pos = search_area.rfind(nl + "{")
if pos >= 0:
json_start = pos + len(nl)
break
if json_start < 0:
json_start = search_area.find("{") # fallback if no newline found
if json_start >= 0:
try:
data = json.loads(search_area[json_start:json_end])
except json.JSONDecodeError:
pass
# ── Determine success ──────────────────────────────────────────────────────
if http_code >= 400:
msg = (data.get("error", {}).get("message", "")
if isinstance(data, dict) else "")
snippet = search_area[json_start:json_start + 120] if json_start >= 0 else search_area[-120:]
return False, f"HTTP {http_code}: {msg or snippet}"
if http_code >= 200 or data:
return True, data
# No HTTP code and no JSON — check for curl-level error
if "curl:" in raw.lower():
return False, f"curl error: {raw.strip()[:120]}"
return True, data
# ── Step 1: Detect serial device ──────────────────────────────────────────────
def detect_serial_device() -> Optional[str]:
rule("Step 1 of 5 -- Serial Cable & Device Detection")
print(" Connect the serial cable from the ES24N IOM1 port")
print(" to the active F-Series controller USB port.")
print()
prompt("Press Enter when the cable is connected")
for attempt in range(1, 4):
info(f"Scanning for USB serial devices (attempt {attempt}/3)...")
time.sleep(1)
# FreeBSD: /dev/ttyU* Linux: /dev/ttyUSB*, /dev/ttyACM*
patterns = ["/dev/ttyUSB*", "/dev/ttyACM*", "/dev/ttyU*"]
ports = sorted({p for pat in patterns for p in glob.glob(pat)})
if ports:
break
if attempt < 3:
warn("No device found yet — retrying in 2 seconds...")
time.sleep(2)
if not ports:
error("No USB serial device detected after 3 attempts.")
print()
print(" Troubleshooting:")
print(" - Ensure the serial cable is fully seated at both ends.")
print(" - Try a different USB port on the controller.")
print(" - Confirm the ES24N is powered on.")
return None
if len(ports) == 1:
ok(f"Device found: {_c(C.BOLD, ports[0])}")
_fix_permissions(ports[0])
return ports[0]
# Multiple devices — let the user choose
print()
draw_table(
["#", "Device"],
[[str(i), p] for i, p in enumerate(ports, 1)],
[4, 24],
)
print()
while True:
val = prompt(f"Select device number [1-{len(ports)}]")
if val.isdigit() and 1 <= int(val) <= len(ports):
selected = ports[int(val) - 1]
ok(f"Selected: {_c(C.BOLD, selected)}")
_fix_permissions(selected)
return selected
warn(f"Please enter a number between 1 and {len(ports)}.")
def _fix_permissions(device: str):
try:
result = subprocess.run(
["sudo", "chown", ":wheel", device],
capture_output=True, timeout=5,
)
if result.returncode == 0:
ok(f"Permissions updated on {device}")
return
except Exception:
pass
try:
os.chmod(device, 0o666)
ok(f"Permissions updated on {device}")
except PermissionError:
warn("Could not update device permissions automatically.")
warn("If the connection fails, re-run this script with sudo.")
# ── Step 2: Open serial connection & wake IOM console ─────────────────────────
def open_serial_connection(device: str) -> Optional[SerialPort]:
rule("Step 2 of 5 -- Opening Serial Connection")
info(f"Opening {device} at 115200 baud (8N1)...")
ser = SerialPort(device, baudrate=115200, timeout=5.0)
try:
ser.open()
except OSError as e:
error(str(e))
return None
ok(f"Port opened: {device}")
info("Sending wake signal to IOM console...")
ser.send_line("", delay=0.5)
ser.send_line("", delay=0.5)
response = _ANSI_RE.sub("", ser.read_until_quiet(quiet_period=0.5))
print()
if response.strip():
print(f" {_c(C.DIM, '+-- IOM Console Response ' + '-' * 31)}")
for line in response.strip().splitlines():
print(f" {_c(C.DIM, '|')} {line}")
print(f" {_c(C.DIM, '+' + '-' * 56)}")
print()
low = response.lower()
if any(kw in low for kw in ("login", "$", "#", "password")):
ok("IOM console is responsive.")
else:
warn("Unexpected response — the IOM may still be booting.")
warn("You can continue; login will be attempted next.")
else:
warn("No response received from IOM console.")
warn("Login will be attempted after the password is entered.")
print()
return ser
# ── Step 3: Fetch & display current IOM network settings ─────────────────────
def fetch_current_config(cfg: ShelfConfig, ser: SerialPort, iom: str) -> bool:
"""
Query Redfish for the current network config of the connected IOM via curl
over the serial console session. Only the IOM whose console port the serial
cable is plugged into is reachable.
Populates cfg.iom1 with live data (regardless of whether IOM1 or IOM2).
Returns True if the IOM responded successfully.
"""
rule(f"Step 3 of 5 -- Current {iom} Network Settings")
info(f"Querying {iom} Redfish API via serial console...")
print()
path = f"/redfish/v1/Managers/{iom}/EthernetInterfaces/1"
ok_flag, data = _serial_redfish_request(ser, cfg.password, "GET", path)
if ok_flag and isinstance(data, dict):
dhcp_enabled = (
data.get("DHCPv4", {}).get("DHCPEnabled", False) or
data.get("DHCPv6", {}).get("DHCPEnabled", False)
)
# Prefer IPv4StaticAddresses; fall back to IPv4Addresses
addrs = data.get("IPv4StaticAddresses") or data.get("IPv4Addresses", [])
if addrs:
addr_rec = addrs[0]
ip = addr_rec.get("Address", "--")
gateway = addr_rec.get("Gateway", "--")
netmask = addr_rec.get("SubnetMask", "--")
else:
ip = gateway = netmask = "--"
cfg.iom1 = IOMConfig(
iom = iom,
dhcp = dhcp_enabled,
ip = ip if ip != "--" else "",
gateway = gateway if gateway != "--" else "",
netmask = netmask if netmask != "--" else "",
)
mode_str = _c(C.CYN, "DHCP") if dhcp_enabled else _c(C.GRN, "Static")
draw_table(
["IOM", "Mode", "IP Address", "Gateway", "Subnet Mask"],
[[iom, mode_str, ip, gateway, netmask]],
[5, 8, 15, 15, 15],
)
print()
return True
else:
draw_table(
["IOM", "Mode", "IP Address", "Gateway", "Subnet Mask"],
[[iom, _c(C.RED, "No response"), "--", "--", "--"]],
[5, 8, 15, 15, 15],
)
print()
error(f"{iom} query failed: {data}")
error(f"Check that the serial cable is connected and {iom} is booted.")
print()
return False
# ── Step 4: Prompt user — change config or exit ───────────────────────────────
def collect_network_config(cfg: ShelfConfig, iom: str) -> bool:
"""
Show current settings, ask user what to do.
Returns True to proceed with applying changes, False to skip.
"""
rule("Step 4 of 5 -- Change Configuration?")
print(f" {_c(C.BOLD, '1')} Change network configuration")
print(f" {_c(C.BOLD, '2')} Leave settings as-is and disconnect")
print()
while True:
choice = prompt("Select option [1/2]")
if choice in ("1", "2"):
break
warn("Please enter 1 or 2.")
if choice == "2":
info("No changes requested.")
return False
# ── User wants to change settings ─────────────────────────────────────────
print()
print(f" How should {iom} be configured?")
print(f" {_c(C.BOLD, '1')} Static IP address")
print(f" {_c(C.BOLD, '2')} DHCP")
print()
while True:
mode = prompt("Select mode [1/2]")
if mode in ("1", "2"):
break
warn("Please enter 1 or 2.")
use_dhcp = (mode == "2")
print()
if use_dhcp:
cfg.iom1 = IOMConfig(iom, dhcp=True)
ok(f"{iom} will be set to DHCP.")
return True
# Static
info(f"Static network details for {_c(C.BOLD, iom)}:")
ip = prompt_ip(f" {iom} IP address ")
gw = prompt_ip(f" {iom} Gateway ")
nm = prompt_ip(f" {iom} Subnet Mask")
cfg.iom1 = IOMConfig(iom, dhcp=False, ip=ip, gateway=gw, netmask=nm)
return True
# ── Step 5a: Apply configuration via Redfish ──────────────────────────────────
def apply_configuration(cfg: ShelfConfig, ser: SerialPort) -> bool:
rule("Step 5 of 5 -- Applying Configuration via Redfish API")
info("Sending Redfish PATCH request to IOM1 via serial console curl...")
print()
success, detail = _apply_iom(cfg.password, cfg.iom1, ser)
status = _c(C.GRN, "OK") if success else _c(C.RED, "FAIL")
draw_table(["IOM", "Result", "Detail"], [["IOM1", status, detail]], [6, 8, 50])
print()
return success
def _apply_iom(password: str, iom_cfg: IOMConfig, ser: SerialPort) -> tuple:
"""
Apply network config to a single IOM via curl over the serial console.
DHCP: single PATCH enabling DHCPv4.
Static: two sequential PATCHes to work around a firmware bug in the
current ES24N release that prevents disabling DHCP and setting a static
address in the same request.
Pass 1 -- set the static IP/gateway/netmask (DHCP still on)
Pass 2 -- disable DHCP (address is already committed)
"""
path = f"/redfish/v1/Managers/{iom_cfg.iom}/EthernetInterfaces/1"
if iom_cfg.dhcp:
ok_flag, data = _serial_redfish_request(
ser, password, "PATCH", path,
{"DHCPv4": {"DHCPEnabled": True}},
)
if ok_flag:
return True, "Configured: DHCP"
return False, str(data)[:80]
# Static -- Pass 1: set address while DHCP is still enabled
info(f" {iom_cfg.iom} pass 1/2 -- setting static address {iom_cfg.ip}...")
ok_flag, data = _serial_redfish_request(
ser, password, "PATCH", path,
{
"IPv4StaticAddresses": [{
"Address": iom_cfg.ip,
"Gateway": iom_cfg.gateway,
"SubnetMask": iom_cfg.netmask,
}]
},
)
if not ok_flag:
return False, f"Pass 1 failed: {str(data)[:70]}"
# Brief pause to allow the IOM to commit the address before the next call
time.sleep(1)
# Static -- Pass 2: disable DHCP now that the static address is committed
info(f" {iom_cfg.iom} pass 2/2 -- disabling DHCP...")
ok_flag, data = _serial_redfish_request(
ser, password, "PATCH", path,
{"DHCPv4": {"DHCPEnabled": False}},
)
if not ok_flag:
return False, f"Pass 2 failed: {str(data)[:70]}"
return True, f"Configured: Static {iom_cfg.ip}"
# ── Step 5b: Print applied-settings summary ───────────────────────────────────
def print_summary(cfg: ShelfConfig, changed: bool):
rule("Summary")
iom = cfg.iom1
if iom.dhcp:
mode, ip, gateway, netmask = "DHCP", "-- (DHCP)", "-- (DHCP)", "-- (DHCP)"
else:
mode, ip, gateway, netmask = "Static", iom.ip, iom.gateway, iom.netmask
draw_table(
["Setting", "IOM1"],
[
["Mode", mode],
["IP Address", ip],
["Gateway", gateway],
["Subnet Mask", netmask],
["Serial Port", cfg.device],
["Changes", "Yes" if changed else "None"],
],
[12, 28],
)
if changed:
print()
draw_box([
f"{_c(C.YEL, 'IMPORTANT -- Per the ES24N Service Guide:')}",
"",
"Remove the serial cable ONLY after verifying the",
"expander appears in TrueNAS with matching drives.",
"",
"TrueNAS > System Settings > Enclosure >",
"NVMe-oF Expansion Shelves",
], colour=C.YEL)
print()
# ── Disconnect ────────────────────────────────────────────────────────────────
def close_serial_connection(ser: SerialPort, device: str):
if ser and ser.is_open:
ser.close()
ok(f"Serial port {device} closed.")
print()
prompt("Disconnect the serial cable, then press Enter to continue")
ok("Serial cable disconnected. Shelf complete.")
# ── Full shelf configuration cycle ────────────────────────────────────────────
def configure_shelf() -> bool:
"""Run one complete shelf cycle. Returns True if user wants another shelf."""
banner()
cfg = ShelfConfig()
# 1 — Detect device
device = detect_serial_device()
if not device:
error("Could not detect a serial device. Returning to main menu.")
time.sleep(2)
return True
cfg.device = device
# 2 — Open serial port & wake IOM console
ser = open_serial_connection(device)
if not ser:
error("Could not open serial port. Returning to main menu.")
time.sleep(2)
return True
# Log in to the IOM console (auto-detects password from hostname)
print()
logged_in, cfg.password = _login_serial_console(ser)
if not logged_in:
error("Could not log in to IOM console. Returning to main menu.")
close_serial_connection(ser, device)
time.sleep(2)
return True
# Prompt for which IOM the serial cable is connected to
print()
print(" Which IOM is the serial cable connected to?")
print(f" {_c(C.BOLD, '1')} IOM1")
print(f" {_c(C.BOLD, '2')} IOM2")
print()
while True:
iom_choice = prompt("Select [1/2]")
if iom_choice in ("1", "2"):
break
warn("Please enter 1 or 2.")
iom = "IOM1" if iom_choice == "1" else "IOM2"
# 3 — Fetch & display current settings
fetch_current_config(cfg, ser, iom)
# 4 — Ask user: change or leave alone?
apply_changes = collect_network_config(cfg, iom)
# 5 — Apply if requested
changed = False
if apply_changes:
print()
rule("Ready to Apply")
info("Redfish PATCH requests will be sent via curl on the IOM console.")
print()
if prompt_yn("Apply configuration now?", default=True):
apply_configuration(cfg, ser)
changed = True
else:
warn("Configuration skipped — no changes were made.")
# Summary
print_summary(cfg, changed)
# Disconnect
close_serial_connection(ser, device)
print()
return prompt_yn("Configure another ES24N shelf?", default=False)