scrapli IOS-XR handling of standby RSP console

scrapli_xr_console_on_open.py

import asyncio
import re
from typing import TYPE_CHECKING, cast

if TYPE_CHECKING:
    from scrapli.driver import AsyncNetworkDriver
    
class RPNotActiveError(Exception):
    pass
 

IOS_XR_CONSOLE_RE = re.compile(rb"rp node is not ready or active for login|user access verification|username|^[a-z0-9.\-@()/:]{1,48}[#>$]\s*$", re.M | re.I)

async def ios_xr_console_on_open(conn: "AsyncNetworkDriver") -> None:
    """Custom on_open for IOS-XR console handling inactive standby RSP console"""
    conn.channel.send_return()
    buf = b""
    while True:
        new_buf = await conn.channel.read()
        buf += new_buf.lower()
        if re_match := IOS_XR_CONSOLE_RE.search(buf):
            match = re_match.group(0)
            if b"rp node is not ready or active" in match:
                raise RPNotActiveError(f"{conn.host}:{conn.port} is standby RP")
            elif b"user access verification" in match or b"username" in match:
                await conn.channel.channel_authenticate_telnet(conn.auth_username, conn.auth_password)
            break

    await conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)
    await conn.send_command(command="terminal length 0")
    await conn.send_command(command="terminal width 512")
    
async def main() -> None:
    conn_data = {
        "host": "terminal-server1",
        "port": 2001,
        "auth_bypass": True,
        "auth_username": "cisco",
        "auth_password": "cisco",
        "transport": "asyncssh",
        "auth_strict_key": False,
        "platform": "cisco_iosxr",
        "on_open": ios_xr_console_on_open,
    }
    async with AsyncScrapli(**conn_data) as conn:
        conn = cast("AsyncNetworkDriver", conn)
        response = await conn.send_command("show version")
        print(response.result)


if __name__ == '__main__':
    asyncio.run(main())