Talking TL1 from Python: A Working Client, and Where It Stops Being Enough

Talking TL1 from Python: A Working Client, and Where It Stops Being Enough

Sooner or later someone asks why the optical shelves aren't in the nightly backup with everything else. Usually the answer is that nobody wrote the script. So here's the script.

It's about 200 lines of Python on top of paramiko. It logs in to a Ciena 6500 gateway over SSH, finds the remote NEs behind it, pulls inventory from each one and writes a file per NE. Along the way it deals with the four things that break most first attempts at TL1: login happening inside the session, alarms landing in the middle of your answer, acknowledgements that look like answers, and replies that arrive in pages.

We wrote it against the framing rConfig Sim uses for its Ciena personality, and tested it against a stand-in NE that throws all four of those at it. It's Python 3.10+ and paramiko 5. Run it against the simulator or a lab shelf before you point it at anything carrying traffic.

Why not just use Netmiko?

Because TL1 isn't a CLI. Netmiko and its relatives are built around a prompt: send a line, read until the prompt comes back, done. TL1 sends answers framed by a header and a lone ;, may send something else before your answer, and on some platforms prints the prompt in the middle of a reply. You can bend a prompt-based library to fit, but you end up rewriting the reading loop anyway, so we start from paramiko and write it properly.

If TL1 itself is new to you, the TL1 commands and protocol guide covers the command format, TIDs, AIDs and CTAGs. This post assumes you've read it or are happy to look things up.

Opening the session

def __init__(self, host, port, username, password, timeout=30):
    # No prompt matching anywhere. Messages are framed by header and closing ";",
    # so it doesn't matter that Ciena prompts with "<" and Infinera with ">".
    self.timeout = timeout
    self.autonomous = []  # alarms and events that arrived while we were waiting
    self._ctags = itertools.count(1)
    self._buf = ""

    self.ssh = paramiko.SSHClient()
    self.ssh.load_system_host_keys()
    # Refuse unknown host keys. Add the NE to known_hosts first.
    self.ssh.set_missing_host_key_policy(paramiko.RejectPolicy())
    self.ssh.connect(
        host,
        port=port,
        username=username,
        password=password,
        look_for_keys=False,
        allow_agent=False,
        timeout=timeout,
    )
    self.chan = self.ssh.invoke_shell(width=512)
    self.chan.settimeout(1.0)

invoke_shell, not exec_command: the NE wants an interactive session it can talk back on. Unknown host keys are refused, so add the shelf to known_hosts first rather than switching that off. Depending on how the shelf is set up, SSH may ask for credentials as well as TL1. The client sends the same ones to both.

Framing: stop waiting for the prompt

def _next_message(self, deadline):
    """Return the next complete TL1 message, an ack tuple, or keep reading."""
    current = None
    for line in self._read_lines(deadline):
        if current is None:
            ack = ACK.match(line.strip())
            if ack:
                return ("ack", ack.group(1), ack.group(2))
            head = HEADER.match(line)
            if head:
                current = {"sid": head.group(1), "lines": [line]}
            # Anything else outside a message is echo, prompt or blank: skip it.
            continue
        current["lines"].append(line)
        if line.strip() == ";":
            return self._build(current)

@staticmethod
def _build(current):
    code_line = current["lines"][1].split()
    if code_line[0] == "M":
        kind, tag, code = "response", code_line[1], code_line[2]
    else:
        kind, tag, code = "autonomous", code_line[1], code_line[0]
    body = [line for line in current["lines"][2:-1] if line.strip()]
    return Message(
        current["sid"], kind, tag, code, body, "\n".join(current["lines"])
    )

This is the part most scripts get wrong. A TL1 message starts with a header line (the SID, then a YY-MM-DD HH:MM:SS timestamp) and ends with a line holding nothing but ;. So that's what we frame on. The echo of your own command, the prompt and the blank lines all sit outside a message and get skipped.

Not matching the prompt pays off twice. The Infinera DTN-X prompts with > where Ciena uses <, which strands any client waiting for the wrong character. And the DTN-X prints the prompt between the pages of a long reply, so a client that stops at the first prompt leaves the rest of the answer sitting in the socket.

The second line decides what kind of message it is. M means a response to a command, with the CTAG and completion code after it. Anything else (*C, **, *, A) is an autonomous message: an alarm or event the NE sent without being asked.

One CTAG per command, and matching on it

def command(self, verb, tid="", aid="", params=None):
    """Send one command and return (messages, records) for its CTAG."""
    ctag = str(next(self._ctags))
    text = f"{verb}:{tid}:{aid}:{ctag}"
    text += f"::{params};" if params is not None else ";"
    self.chan.sendall((text + "\r\n").encode("ascii"))

    deadline = time.monotonic() + self.timeout
    parts = []
    while True:
        msg = self._next_message(deadline)
        if isinstance(msg, tuple):  # IP or PF: the real answer is still coming
            continue
        if msg.kind == "autonomous":
            self.autonomous.append(msg)
            continue
        if msg.tag != ctag:
            continue  # a late answer to an earlier command that timed out
        if msg.code == "DENY":
            reason = msg.lines[0].strip() if msg.lines else "UNKNOWN"
            raise TL1Error(ctag, reason)
        parts.append(msg)
        if msg.code == "COMPLD":  # RTRV means more blocks follow
            records = [
                parse_record(line)
                for part in parts
                for line in part.lines
                if line.strip().startswith('"')
            ]
            return parts, records

Every command gets its own CTAG from a counter, and the loop only returns when a response carrying that CTAG completes. On the way it deals with everything else the NE might send:

• An acknowledgement like IP 6 (in progress) means the answer is still coming. Keep reading.
• An alarm that turns up mid-command goes into session.autonomous instead of being mistaken for the answer.
• A response with someone else's CTAG is a late answer to an earlier command that timed out. Drop it.
• DENY raises TL1Error with the four-letter code, so a refusal can never be saved as a backup.
• A completion code of RTRV means more blocks follow. Keep collecting until COMPLD.

The command string keeps every colon, VERB:TID:AID:CTAG, even when the TID and AID are empty. Some NEs accept a shorter form, but the full one is what the standard describes.

Logging in with ACT-USER

def login(self, username, password):
    # Always quote it: GR-831 needs quotes for anything beyond letters and digits.
    self.command("ACT-USER", aid=username, params=f'"{password}"')

The SSH session is a pipe. Authentication is a TL1 command, ACT-USER, with the username in the AID field and the password in the parameter block. Until it returns COMPLD, the NE answers everything with DENY PLNA, which command() turns into an exception. If login fails you find out straight away, not three commands later when every answer is empty.

The password is always quoted. A bare password works until someone sets one with a colon or a semicolon in it, and then the NE reads half of it as the next field. A password containing a double quote will still break this. Don't use one.

Parsing records

def split_outside_quotes(text, sep):
    """Split on sep, ignoring separators inside double quotes."""
    parts, buf, quoted = [], [], False
    for ch in text:
        if ch == '"':
            quoted = not quoted
        if ch == sep and not quoted:
            parts.append("".join(buf))
            buf = []
        else:
            buf.append(ch)
    parts.append("".join(buf))
    return parts


def parse_record(line):
    """Turn one quoted TL1 record into its positional blocks, each a list of items."""
    body = line.strip()
    if body.startswith('"') and body.endswith('"'):
        body = body[1:-1]
    body = body.replace('\\"', '"')
    return [
        split_outside_quotes(block, ",") for block in split_outside_quotes(body, ":")
    ]


def keywords(record):
    """Collect every KEY=VALUE item in a parsed record into a dict."""
    out = {}
    for block in record:
        for item in block:
            if "=" in item:
                key, value = item.split("=", 1)
                out[key] = value.strip('"')
    return out

A record is a quoted line, split into positional blocks by : and into items by ,, except where either sits inside quotes. Ciena nests escaped quotes inside records, which is why the \" gets unescaped first:

>>> rec = parse_record('"SHELF-1::SID=\\"RNE-GALWAY\\",NENAME=\\"RNE-GALWAY\\",GNE=NO"')
>>> keywords(rec)
{'SID': 'RNE-GALWAY', 'NENAME': 'RNE-GALWAY', 'GNE': 'NO'}

keywords() only helps on platforms that label their fields. The Cisco ONS 15454's RTRV-MAP-NETWORK records are purely positional (address, node name, product), so there you read the blocks by index and keywords() hands back an empty dict.

Walking the gateway

def backup_gateway(host, port, username, password, outdir="backups"):
    session = TL1Session(host, port, username, password)
    failures = {}
    try:
        session.login(username, password)

        _, sys_records = session.command("RTRV-SYS")
        gateway = keywords(sys_records[0])["SID"]

        # The NE list leaves out the gateway itself, so start with it.
        targets = [gateway]
        _, ne_records = session.command("RTRV-NE-LIST")
        targets += [keywords(r)["SID"] for r in ne_records]

        stamp = time.strftime("%Y%m%d-%H%M%S")
        for tid in targets:
            try:
                parts, _ = session.command("RTRV-EQPT", tid=tid, aid="ALL")
            except TL1Error as err:
                failures[tid] = err.code
                continue
            path = os.path.join(outdir, tid, f"{stamp}.txt")
            os.makedirs(os.path.dirname(path), exist_ok=True)
            with open(path, "w") as fh:
                fh.write("\n".join(p.raw for p in parts) + "\n")
    finally:
        session.close()

    for msg in session.autonomous:
        print(f"autonomous {msg.code} from {msg.sid}: {msg.lines[0].strip()}")
    return targets, failures

Two details matter here. RTRV-NE-LIST lists the remote NEs but not the gateway itself, so the gateway's own SID (from RTRV-SYS) goes on the list first. And a DENY on one remote, typically IIAC for an RNE that's unreachable, is recorded as a failure and the walk carries on. The gateway answered that command, not the RNE, so saving it would give you a tidy backup of a shelf you never reached.

Running it

pip install paramiko
export TL1_HOST=192.0.2.10 TL1_USER=yourname TL1_PASS='your password'
python tl1client.py

Against our stand-in gateway, with three remotes behind it, one of them unreachable and an alarm thrown in mid-session:

autonomous ** from CIENA-LAX-1001: "SLOT-7:MJ,T-LOS,NSA,,,,:\"Loss of signal\""
collected 3 of 4 NEs, failed: {'RNE-NOWHERE': 'IIAC'}

You get a folder per NE under backups/, with a timestamped file each time it runs.

Where this stops being enough

One gateway, a handful of shelves and a cron job: this is fine. Keep it.

At fifty gateways, here's what you end up adding:

• Credentials somewhere better than environment variables, rotated, with an audit of who used them.
• Concurrency, with timeouts and retries per NE, so one slow gateway doesn't hold up the night.
• Diffs. This script writes files. It doesn't tell you what changed since yesterday.
• Alerting on every DENY and timeout, so a dead RNE shows up the next morning, not at the next audit.
• The Cisco ONS and Infinera dialects, which differ in more than the prompt.
• History you can hand an auditor, next to the same history for the routers and switches.

None of that is hard on its own. Together it's a product to maintain. It's also what rConfig's TL1 drivers already do: TL1 configuration management covers the Ciena 6500, Cisco ONS 15454 and Infinera DTN-X, on the same schedules and in the same inventory as the IP side. Plenty of teams run both, the script for odd jobs and rConfig for the estate.

The whole script

"""A small TL1-over-SSH client for a Ciena 6500 gateway. Python 3.10+, paramiko."""

import itertools
import os
import re
import time
from dataclasses import dataclass, field

import paramiko

HEADER = re.compile(r"^\s+(\S+) (\d{2}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s*$")
ACK = re.compile(r"^(IP|PF|OK|NA|NG|RL) (\S+)\s*$")


class TL1Error(Exception):
    def __init__(self, ctag, code):
        super().__init__(f"CTAG {ctag} denied: {code}")
        self.ctag = ctag
        self.code = code


@dataclass
class Message:
    sid: str
    kind: str  # "response" or "autonomous"
    tag: str  # CTAG for responses, ATAG for autonomous messages
    code: str  # COMPLD, DENY, RTRV ... or the alarm code (*C, **, *, A)
    lines: list = field(default_factory=list)
    raw: str = ""


def split_outside_quotes(text, sep):
    """Split on sep, ignoring separators inside double quotes."""
    parts, buf, quoted = [], [], False
    for ch in text:
        if ch == '"':
            quoted = not quoted
        if ch == sep and not quoted:
            parts.append("".join(buf))
            buf = []
        else:
            buf.append(ch)
    parts.append("".join(buf))
    return parts


def parse_record(line):
    """Turn one quoted TL1 record into its positional blocks, each a list of items."""
    body = line.strip()
    if body.startswith('"') and body.endswith('"'):
        body = body[1:-1]
    body = body.replace('\\"', '"')
    return [
        split_outside_quotes(block, ",") for block in split_outside_quotes(body, ":")
    ]


def keywords(record):
    """Collect every KEY=VALUE item in a parsed record into a dict."""
    out = {}
    for block in record:
        for item in block:
            if "=" in item:
                key, value = item.split("=", 1)
                out[key] = value.strip('"')
    return out


class TL1Session:
    def __init__(self, host, port, username, password, timeout=30):
        # No prompt matching anywhere. Messages are framed by header and closing ";",
        # so it doesn't matter that Ciena prompts with "<" and Infinera with ">".
        self.timeout = timeout
        self.autonomous = []  # alarms and events that arrived while we were waiting
        self._ctags = itertools.count(1)
        self._buf = ""

        self.ssh = paramiko.SSHClient()
        self.ssh.load_system_host_keys()
        # Refuse unknown host keys. Add the NE to known_hosts first.
        self.ssh.set_missing_host_key_policy(paramiko.RejectPolicy())
        self.ssh.connect(
            host,
            port=port,
            username=username,
            password=password,
            look_for_keys=False,
            allow_agent=False,
            timeout=timeout,
        )
        self.chan = self.ssh.invoke_shell(width=512)
        self.chan.settimeout(1.0)

    # --- reading -----------------------------------------------------------

    def _read_lines(self, deadline):
        """Yield complete lines from the channel until the deadline passes."""
        while True:
            while "\n" in self._buf:
                line, self._buf = self._buf.split("\n", 1)
                yield line.rstrip("\r")
            if time.monotonic() > deadline:
                raise TimeoutError("no complete TL1 response before the deadline")
            try:
                chunk = self.chan.recv(65536)
            except TimeoutError:
                continue
            if not chunk:
                raise ConnectionError("NE closed the session")
            self._buf += chunk.decode("ascii", errors="replace")

    def _next_message(self, deadline):
        """Return the next complete TL1 message, an ack tuple, or keep reading."""
        current = None
        for line in self._read_lines(deadline):
            if current is None:
                ack = ACK.match(line.strip())
                if ack:
                    return ("ack", ack.group(1), ack.group(2))
                head = HEADER.match(line)
                if head:
                    current = {"sid": head.group(1), "lines": [line]}
                # Anything else outside a message is echo, prompt or blank: skip it.
                continue
            current["lines"].append(line)
            if line.strip() == ";":
                return self._build(current)

    @staticmethod
    def _build(current):
        code_line = current["lines"][1].split()
        if code_line[0] == "M":
            kind, tag, code = "response", code_line[1], code_line[2]
        else:
            kind, tag, code = "autonomous", code_line[1], code_line[0]
        body = [line for line in current["lines"][2:-1] if line.strip()]
        return Message(
            current["sid"], kind, tag, code, body, "\n".join(current["lines"])
        )

    # --- commands ----------------------------------------------------------

    def command(self, verb, tid="", aid="", params=None):
        """Send one command and return (messages, records) for its CTAG."""
        ctag = str(next(self._ctags))
        text = f"{verb}:{tid}:{aid}:{ctag}"
        text += f"::{params};" if params is not None else ";"
        self.chan.sendall((text + "\r\n").encode("ascii"))

        deadline = time.monotonic() + self.timeout
        parts = []
        while True:
            msg = self._next_message(deadline)
            if isinstance(msg, tuple):  # IP or PF: the real answer is still coming
                continue
            if msg.kind == "autonomous":
                self.autonomous.append(msg)
                continue
            if msg.tag != ctag:
                continue  # a late answer to an earlier command that timed out
            if msg.code == "DENY":
                reason = msg.lines[0].strip() if msg.lines else "UNKNOWN"
                raise TL1Error(ctag, reason)
            parts.append(msg)
            if msg.code == "COMPLD":  # RTRV means more blocks follow
                records = [
                    parse_record(line)
                    for part in parts
                    for line in part.lines
                    if line.strip().startswith('"')
                ]
                return parts, records

    def login(self, username, password):
        # Always quote it: GR-831 needs quotes for anything beyond letters and digits.
        self.command("ACT-USER", aid=username, params=f'"{password}"')

    def close(self):
        try:
            self.command("CANC-USER")
        except (TL1Error, TimeoutError):
            pass  # not every NE (or simulator) implements CANC-USER
        self.ssh.close()


def backup_gateway(host, port, username, password, outdir="backups"):
    session = TL1Session(host, port, username, password)
    failures = {}
    try:
        session.login(username, password)

        _, sys_records = session.command("RTRV-SYS")
        gateway = keywords(sys_records[0])["SID"]

        # The NE list leaves out the gateway itself, so start with it.
        targets = [gateway]
        _, ne_records = session.command("RTRV-NE-LIST")
        targets += [keywords(r)["SID"] for r in ne_records]

        stamp = time.strftime("%Y%m%d-%H%M%S")
        for tid in targets:
            try:
                parts, _ = session.command("RTRV-EQPT", tid=tid, aid="ALL")
            except TL1Error as err:
                failures[tid] = err.code
                continue
            path = os.path.join(outdir, tid, f"{stamp}.txt")
            os.makedirs(os.path.dirname(path), exist_ok=True)
            with open(path, "w") as fh:
                fh.write("\n".join(p.raw for p in parts) + "\n")
    finally:
        session.close()

    for msg in session.autonomous:
        print(f"autonomous {msg.code} from {msg.sid}: {msg.lines[0].strip()}")
    return targets, failures


if __name__ == "__main__":
    done, failed = backup_gateway(
        os.environ["TL1_HOST"],
        int(os.environ.get("TL1_PORT", "22")),
        os.environ["TL1_USER"],
        os.environ["TL1_PASS"],
    )
    print(f"collected {len(done) - len(failed)} of {len(done)} NEs, failed: {failed}")

Further reading

• TL1 commands and protocol guide: command anatomy, responses, alarms and error codes
• Ciena 6500 TL1 GNE/RNE routing, with real session output
• rConfig Sim on GitHub, to run this against simulated Ciena, Cisco and Infinera NEs
• Ciena TL1 examples from the simulator docs