Wrapping Legacy Network Gear in a Modern API

Articles · July 2026 · Ioannis (John) Konstas

Plenty of infrastructure predates the API era. Virtualized router fleets often expose their consoles as serial-over-network — raw Telnet sessions mapped to a range of ports — which is fine for a human with a terminal and hostile to automation: unpredictable timing, escape-character noise, no authentication, no structure. The goal of this project was to abstract those legacy serial interfaces into a clean, authenticated REST API without exposing the underlying fleet to anything. Here's the architecture, modernized.

Foundation: total isolation

The security model starts in the network, not the code. The hypervisor hosting the routers has no default gateway; it reaches the outside world only through a management host over a Layer-3 point-to-point link. The router fleet is unreachable from the internet by any direct path — every management action funnels through one controlled choke point. Nothing below survives if this principle is violated, so it comes first.

The bridge, in three tiers

The stack converts stateless HTTP requests into stateful terminal sessions:

1. An async API service — authenticates the caller, validates the target, and enqueues the work.
2. A session controller — manages the fragile timing of virtual serial consoles.
3. A transport relay — bridges local ports to the hypervisor's console ports.

The original built this on Flask and shell-invoked Telnet with hand-tuned sleeps. The modernized version below is fully async, uses a real Telnet client library instead of shelling out, enforces API keys and per-target allow-listing, and returns structured JSON with proper timeouts.

Tier 1 — the API (FastAPI, async, validated)

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, field_validator
import asyncio, os, secrets

ALLOWED_PORTS = set(range(2301, 2321))          # the console range, explicit
API_KEY = os.environ["ROUTER_API_KEY"]          # never hard-coded
HYPERVISOR = "198.51.100.10"                        # reachable only via mgmt host

app = FastAPI(title="Router Console API")

class Command(BaseModel):
    port: int
    command: str

    @field_validator("port")
    @classmethod
    def port_in_range(cls, v):
        if v not in ALLOWED_PORTS:
            raise ValueError("port outside allowed console range")
        return v

    @field_validator("command")
    @classmethod
    def no_config_mode(cls, v):
        # Read-only gateway: refuse config/exec that could change state
        banned = ("conf", "reload", "erase", "write", "delete")
        if v.strip().lower().startswith(banned):
            raise ValueError("only show/diagnostic commands permitted")
        return v

def require_key(x_api_key: str = Header(...)):
    if not secrets.compare_digest(x_api_key, API_KEY):
        raise HTTPException(status_code=401, detail="bad api key")

@app.post("/console")
async def run(cmd: Command, _=Header(None, alias="X-API-Key")):
    require_key(_)
    try:
        output = await asyncio.wait_for(
            execute(cmd.port, cmd.command), timeout=15.0)
    except asyncio.TimeoutError:
        raise HTTPException(504, "console session timed out")
    return {"port": cmd.port, "router": f"R{cmd.port - 2300}",
            "output": scrub(output)}

Tier 2 — the session controller (no more magic sleeps)

Virtual serial consoles drop characters if you type faster than the buffer drains. The old fix was a fixed sleep ladder; the robust fix is to wait for the prompt — read until the device signals it's ready, then send the next line. This is faster and doesn't break when a device is slow.

import telnetlib3   # asyncio-native Telnet

PROMPT = re.compile(rb"[\r\n][\w.-]+[#>]\s*$")

async def execute(port: int, command: str) -> str:
    reader, writer = await telnetlib3.open_connection(HYPERVISOR, port)
    try:
        await _read_until_prompt(reader)          # wake the console
        writer.write("terminal length 0\r\n")     # disable pagination
        await _read_until_prompt(reader)
        writer.write(command + "\r\n")
        output = await _read_until_prompt(reader)
        return output
    finally:
        writer.close()                             # always release the session

async def _read_until_prompt(reader, timeout=8.0) -> str:
    buf = bytearray()
    while not PROMPT.search(buf):
        chunk = await asyncio.wait_for(reader.read(1024), timeout)
        if not chunk:
            break
        buf.extend(chunk.encode() if isinstance(chunk, str) else chunk)
    return buf.decode(errors="replace")

Tier 3 — transport relay and output hygiene

If the API host can't reach the console ports directly, a socat relay bridges them, forking per connection so concurrent sessions don't collide:

# One relay per console port; TCP-LISTEN forks a child per client
for p in $(seq 2301 2320); do
  socat TCP-LISTEN:${p},reuseaddr,fork \
        TCP:198.51.100.10:${p} &
done

Finally, strip the Telnet negotiation bytes, ANSI escapes, and echoed command so callers get clean text:

import re
_ANSI = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
_IAC  = re.compile(rb"\xff[\xfa-\xff].")

def scrub(raw: str) -> str:
    text = _ANSI.sub("", raw)
    return "\n".join(l.rstrip() for l in text.splitlines()
                     if l.strip() and not l.startswith("terminal length"))

Reaching it safely from anywhere

No inbound firewall port needs opening. An outbound-only tunnel (Cloudflare Tunnel, or an SSH reverse tunnel) publishes the API to an authenticated HTTPS endpoint while the management host initiates the connection. An engineer anywhere runs one authenticated request:

curl -X POST https://console.example.internal/console \
  -H "X-API-Key: $ROUTER_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"port": 2301, "command": "show ip interface brief"}'

{
  "port": 2301,
  "router": "R1",
  "output": "Interface        IP-Address      OK? Status  Protocol\nGig1  10.0.0.1  YES up  up"
}

What modernizing changed

• Prompt-driven timing replaced fixed sleeps — faster and reliable under load.
• Async throughout — many concurrent sessions without threads or crashes.
• Validation and a read-only guard at the edge — the API cannot be tricked into config mode or out-of-range ports.
• Secrets from the environment, constant-time key comparison, structured JSON out.
• Isolation preserved — the fleet is still unreachable except through one outbound-only path.

Legacy interfaces don't need replacing to be usable — they need a disciplined bridge. Isolate the fleet at the network layer, wait for the prompt instead of guessing, validate everything at the edge, and never open an inbound port.

The alternative: Netmiko instead of a gateway

The API gateway above earns its keep when you actually want a REST surface — a caller anywhere, authenticated over HTTPS, that doesn't need Python installed. If the consumer is a script rather than an HTTP client, though, building a whole gateway is often more than the job needs. Netmiko — a Python library that wraps SSH or Telnet and hides the vendor-specific quirks (prompts, paging, enable mode) behind one consistent interface — talks to this exact fleet directly, no gateway required: Netmiko's cisco_ios_telnet device type handles the same forwarded console ports Tier 3 relays. SSH fleets use the plain cisco_ios type shown below; swap it for cisco_ios_telnet and the rest of the code is identical for a Telnet-only fleet like this one.

Start read-only — it can't break anything and proves the connection works. Netmiko can parse output into structured data via TextFSM/Genie instead of leaving you to regex a text blob:

from netmiko import ConnectHandler

device = {
    "device_type": "cisco_ios",
    "host": host,
    "username": "admin",
    "password": get_secret("r1"),     # never hard-code; pull from a vault/env
    "fast_cli": True,
}

with ConnectHandler(**device) as conn:      # context manager -> always disconnects
    # use_textfsm parses the table into a list of dicts, not a blob of text
    interfaces = conn.send_command("show ip interface brief", use_textfsm=True)

for intf in interfaces:
    if intf["status"] != "up":
        print(f"DOWN: {intf['interface']} ({intf['ip_address']})")

Pushing config is where automation earns trust or loses it. Send config as a list — Netmiko enters config mode for you — verify the change applied, and only then save:

commands = ["interface Loopback100", "description managed-by-automation", "ip address 10.100.0.1 255.255.255.255"]

with ConnectHandler(**device) as conn:
    output = conn.send_config_set(commands)     # enters/exits config mode
    check = conn.send_command("show ip interface brief | include Loopback100")
    if "10.100.0.1" in check:
        conn.save_config()                       # write mem / copy run start
    else:
        raise RuntimeError(f"change did not apply cleanly: {output}")

One device at a time is a demo; the point is a hundred at once. Network I/O is the bottleneck, so a thread pool scales cleanly — and the rule that separates professionals from outages is that one bad device can't abort the batch:

from concurrent.futures import ThreadPoolExecutor, as_completed

def audit(host: str) -> tuple[str, str]:
    dev = {**base, "host": host}
    try:
        with ConnectHandler(**dev) as c:
            return host, c.send_command("show version | include uptime")
    except Exception as e:                        # unreachable, auth fail, timeout
        return host, f"ERROR: {e}"

results = {}
with ThreadPoolExecutor(max_workers=20) as pool:
    futures = {pool.submit(audit, h): h for h in inventory}
    for f in as_completed(futures):
        host, out = f.result()
        results[host] = out                       # one bad device can't stop the rest

Netmiko is the low-level SSH engine. Above it sit richer frameworks: NAPALM for vendor-neutral getters and config replace/rollback, and Nornir for a pure-Python automation framework with inventory and concurrency built in — it often uses Netmiko underneath. The natural progression is to script tasks with Netmiko first, then graduate to Nornir/NAPALM as inventory size and safety requirements grow.

Where this goes next

The gateway above solves the "last mile" — console access via a REST API. The next layer is a professional-grade automation platform on top of it: structured device interaction, parallel fleet operations, live topology graphing, and packet-level diagnostics, all served over local HTTPS with no external cloud dependency.

← All articles

Research and direction by Ioannis (John) Konstas; compiled and finalized with AI assistance.