Daily Python Practice: Advanced Applications of telnetlib

We will implement batch configuration distribution + device fingerprint recognition + configuration rollback

๐Ÿ›  Pain Points of the Scenario

  • There are dozens or hundreds of old switches/routers, without SSH, only Telnet available
  • Mixed device models: Cisco, H3C, Huawei, etc., with different prompts and commands
  • Every time batch configuration changes, there is a fear of incorrect commands causing service interruptions
  • Rollback can only be done manually, connecting to each device one by one, which is time-consuming and risky

Solution: One-click identification โ†’ Batch distribution โ†’ Difference check โ†’ Automatic rollback on failure

๐Ÿ”‘ Core Functions

  1. Device Fingerprint Recognition

  • Automatically identify vendor and system type through Banner or prompt
  • Batch Configuration Distribution

    • Concurrent pushing of configuration commands, real-time logging
  • Automatic Snapshot Generation

    • Execute <span>show running-config</span> before distribution to save the current configuration
  • One-click Rollback

    • Quickly restore the original configuration in case of configuration anomalies or manual rollback triggers

    ๐Ÿ’ป Complete Example Code

    #!/usr/bin/env python3
    # -*- coding: utf-8 -*-
    
    """
    Telnet Automation Advanced:
    - Automatically identify device type
    - Batch distribution of configuration
    - Snapshot/Rollback mechanism
    """
    
    import telnetlib
    import re
    import time
    from pathlib import Path
    from concurrent.futures import ThreadPoolExecutor, as_completed
    
    LOG_DIR = Path("./logs")
    BACKUP_DIR = Path("./backups")
    LOG_DIR.mkdir(exist_ok=True)
    BACKUP_DIR.mkdir(exist_ok=True)
    
    
    class Device:
        """Encapsulates basic device information"""
        def __init__(self, host, username, password, enable=None):
            self.host = host
            self.username = username
            self.password = password
            self.enable = enable
            self.prompt = b"&gt; "
            self.vendor = "unknown"
            self.tn = None
    
    
    class TelnetHelper:
        """Encapsulates basic Telnet session functions"""
    
        def __init__(self, device: Device, timeout=8, encoding="utf-8"):
            self.dev = device
            self.timeout = timeout
            self.encoding = encoding
    
        def connect(self):
            """Connect and log in"""
            self.dev.tn = telnetlib.Telnet(self.dev.host, port=23, timeout=self.timeout)
            tn = self.dev.tn
            tn.read_until(b"login: ", timeout=self.timeout)
            tn.write(self.dev.username.encode() + b"\n")
            tn.read_until(b"Password: ", timeout=self.timeout)
            tn.write(self.dev.password.encode() + b"\n")
            idx, _, _ = tn.expect([b"&gt; ", b"# "], timeout=self.timeout)
            self.dev.prompt = b"# "if idx == 1elseb"&gt; "
            return True
    
        def identify_device(self):
            """Identify vendor type through banner or prompt"""
            tn = self.dev.tn
            tn.write(b"\n")
            time.sleep(0.5)
            banner = tn.read_very_eager().decode(errors="ignore").lower()
            if "cisco" in banner:
                self.dev.vendor = "cisco"
            elif "huawei" in banner or "vrp" in banner:
                self.dev.vendor = "huawei"
            elif "h3c" in banner:
                self.dev.vendor = "h3c"
            else:
                self.dev.vendor = "unknown"
            return self.dev.vendor
    
        def _run_cmd(self, cmd):
            """Run a single command and return output"""
            tn = self.dev.tn
            tn.write(cmd.encode() + b"\n")
            time.sleep(0.5)
            output = tn.read_very_eager().decode(self.encoding, errors="ignore")
            return output
    
        def backup_config(self):
            """Backup current configuration"""
            if self.dev.vendor == "cisco":
                cmd = "show running-config"
            elif self.dev.vendor == "huawei":
                cmd = "display current-configuration"
            else:
                cmd = "show running-config"
            output = self._run_cmd(cmd)
            with open(BACKUP_DIR / f"{self.dev.host}.cfg", "w", encoding="utf-8") as f:
                f.write(output)
            return output
    
        def push_config(self, commands):
            """Distribute configuration"""
            logs = []
            if self.dev.vendor == "cisco":
                self._run_cmd("conf t")
            for cmd in commands:
                logs.append(f"$ {cmd}\n{self._run_cmd(cmd)}")
            if self.dev.vendor == "cisco":
                self._run_cmd("end")
                self._run_cmd("wr mem")
            with open(LOG_DIR / f"{self.dev.host}.log", "w", encoding="utf-8") as f:
                f.write("\n".join(logs))
            return "\n".join(logs)
    
        def rollback_config(self):
            """Rollback to backup configuration"""
            backup_file = BACKUP_DIR / f"{self.dev.host}.cfg"
            if not backup_file.exists():
                return "No backup file, cannot rollback"
            with open(backup_file, encoding="utf-8") as f:
                lines = f.readlines()
            self.push_config([line.strip() for line in lines if line.strip()])
            return f"{self.dev.host} has been rolled back"
    
        def close(self):
            if self.dev.tn:
                try:
                    self.dev.tn.write(b"exit\n")
                except:
                    pass
                self.dev.tn.close()
    
    
    def job(dev, commands):
        """Single device task: login-identify-backup-push"""
        helper = TelnetHelper(dev)
        try:
            helper.connect()
            vendor = helper.identify_device()
            print(f"[{dev.host}] Identified as {vendor} device")
            helper.backup_config()
            result = helper.push_config(commands)
            return dev.host, True, result
        except Exception as e:
            return dev.host, False, str(e)
        finally:
            helper.close()
    
    
    if __name__ == "__main__":
        # === Batch task example ===
        devices = [
            Device("192.168.1.10", "admin", "pass", "enablepwd"),
            Device("192.168.1.11", "admin", "pass"),
        ]
        commands = [
            "interface vlan 1",
            "description UPDATED_BY_SCRIPT",
            "exit",
        ]
    
        with ThreadPoolExecutor(max_workers=5) as pool:
            futures = [pool.submit(job, dev, commands) for dev in devices]
            for fut in as_completed(futures):
                host, ok, res = fut.result()
                if ok:
                    print(f"โœ… {host} configuration successful")
                else:
                    print(f"โŒ {host} configuration failed: {res}")
    

    ๐Ÿงช Input / Output Examples

    Execution Input

    python telnet_rollback.py
    

    Output Example

    [192.168.1.10] Identified as cisco device
    โœ… 192.168.1.10 configuration successful
    [192.168.1.11] Identified as unknown device
    โŒ 192.168.1.11 configuration failed: Login timeout
    

    Generated Files

    • <span>logs/192.168.1.10.log</span>: Configuration distribution log
    • <span>backups/192.168.1.10.cfg</span>: Current configuration snapshot

    โš ๏ธ Practical Tips

    • Fingerprint Recognition Strategy: Prefer using Banner, then prompt, and if necessary, execute <span>show version</span> for regex matching
    • Prevent Service Interruptions: It is recommended to first use <span>dry-run</span> mode to print commands without executing
    • Backup Strategy: Keep multiple historical versions, supporting <span>rollback --version=2</span>
    • Concurrency Limitation: Control thread pool size when dealing with multiple devices to prevent overload or being blocked by firewalls

    ๐Ÿ“Œ Summary

    • Using <span>telnetlib</span> + device fingerprint recognition, achieve automatic differentiation of vendor devices
    • Automatically backup snapshots before distributing configurations, with one-click rollback in case of anomalies
    • Combined with logging, easily achieve auditable and traceable batch automated operations

    For further evolution, consider combining with:

    • <span>paramiko</span> to replace SSH version
    • <span>rich</span> for colorful progress and beautiful logs
    • <span>sqlite</span> to store device information and change history, achieving a “lightweight CMDB”

    Leave a Comment