Skip to content

async_driver

scrapli.driver.core.cisco_nxos.async_driver

AsyncNXOSDriver

Bases: AsyncNetworkDriver, NXOSDriverBase

Source code in driver/core/cisco_nxos/async_driver.py
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
class AsyncNXOSDriver(AsyncNetworkDriver, NXOSDriverBase):
    def __init__(
        self,
        host: str,
        privilege_levels: Optional[Dict[str, PrivilegeLevel]] = None,
        default_desired_privilege_level: str = "privilege_exec",
        port: Optional[int] = None,
        auth_username: str = "",
        auth_password: str = "",
        auth_private_key: str = "",
        auth_private_key_passphrase: str = "",
        auth_strict_key: bool = True,
        auth_bypass: bool = False,
        timeout_socket: float = 15.0,
        timeout_transport: float = 30.0,
        timeout_ops: float = 30.0,
        comms_return_char: str = "\n",
        ssh_config_file: Union[str, bool] = False,
        ssh_known_hosts_file: Union[str, bool] = False,
        on_init: Optional[Callable[..., Any]] = None,
        on_open: Optional[Callable[..., Any]] = None,
        on_close: Optional[Callable[..., Any]] = None,
        transport: str = "system",
        transport_options: Optional[Dict[str, Any]] = None,
        channel_log: Union[str, bool, BytesIO] = False,
        channel_log_mode: str = "write",
        channel_lock: bool = False,
        logging_uid: str = "",
        auth_secondary: str = "",
        failed_when_contains: Optional[List[str]] = None,
        textfsm_platform: str = "cisco_nxos",
        genie_platform: str = "nxos",
    ):
        """
        NXOSDriver Object

        Please see `scrapli.driver.base.base_driver.Driver` for all "base driver" arguments!

        # noqa: DAR101

        Args:
             privilege_levels: optional user provided privilege levels, if left None will default to
                scrapli standard privilege levels
            default_desired_privilege_level: string of name of default desired priv, this is the
                priv level that is generally used to disable paging/set terminal width and things
                like that upon first login, and is also the priv level scrapli will try to acquire
                for normal "command" operations (`send_command`, `send_commands`)
            auth_secondary: password to use for secondary authentication (enable)
            on_open: callable that accepts the class instance as its only argument. this callable,
                if provided, is executed immediately after authentication is completed. Common use
                cases for this callable would be to disable paging or accept any kind of banner
                message that prompts a user upon connection
            on_close: callable that accepts the class instance as its only argument. this callable,
                if provided, is executed immediately prior to closing the underlying transport.
                Common use cases for this callable would be to save configurations prior to exiting,
                or to logout properly to free up vtys or similar.
            textfsm_platform: string name of textfsm parser platform
            genie_platform: string name of cisco genie parser platform
            failed_when_contains: List of strings that indicate a command/config has failed

        Returns:
            None

        Raises:
            N/A

        """
        # somewhere/somehow the mixin is causing mypy to be upset about comms_prompt_pattern...
        self.comms_prompt_pattern: str

        if privilege_levels is None:
            privilege_levels = deepcopy(PRIVS)

        if on_open is None:
            on_open = nxos_on_open
        if on_close is None:
            on_close = nxos_on_close

        if failed_when_contains is None:
            failed_when_contains = FAILED_WHEN_CONTAINS.copy()

        super().__init__(
            host=host,
            port=port,
            auth_username=auth_username,
            auth_password=auth_password,
            auth_private_key=auth_private_key,
            auth_private_key_passphrase=auth_private_key_passphrase,
            auth_strict_key=auth_strict_key,
            auth_bypass=auth_bypass,
            timeout_socket=timeout_socket,
            timeout_transport=timeout_transport,
            timeout_ops=timeout_ops,
            comms_return_char=comms_return_char,
            ssh_config_file=ssh_config_file,
            ssh_known_hosts_file=ssh_known_hosts_file,
            on_init=on_init,
            on_open=on_open,
            on_close=on_close,
            transport=transport,
            transport_options=transport_options,
            channel_log=channel_log,
            channel_log_mode=channel_log_mode,
            channel_lock=channel_lock,
            logging_uid=logging_uid,
            privilege_levels=privilege_levels,
            default_desired_privilege_level=default_desired_privilege_level,
            auth_secondary=auth_secondary,
            failed_when_contains=failed_when_contains,
            textfsm_platform=textfsm_platform,
            genie_platform=genie_platform,
        )

    async def _abort_config(self) -> None:
        """
        Abort NXOS configuration session (if using a config session!)

        Args:
            N/A

        Returns:
            None

        Raises:
            N/A

        """
        # nxos pattern for config sessions should *always* have `config-s`
        if "config\\-s" in self._current_priv_level.pattern:
            await self.channel.send_input(channel_input="abort")
            self._current_priv_level = self.privilege_levels["privilege_exec"]

    def register_configuration_session(self, session_name: str) -> None:
        """
        NXOS specific implementation of register_configuration_session

        Args:
            session_name: name of session to register

        Returns:
            None

        Raises:
            N/A

        """
        self._create_configuration_session(session_name=session_name)
        self.update_privilege_levels()

__init__(host: str, privilege_levels: Optional[Dict[str, PrivilegeLevel]] = None, default_desired_privilege_level: str = 'privilege_exec', port: Optional[int] = None, auth_username: str = '', auth_password: str = '', auth_private_key: str = '', auth_private_key_passphrase: str = '', auth_strict_key: bool = True, auth_bypass: bool = False, timeout_socket: float = 15.0, timeout_transport: float = 30.0, timeout_ops: float = 30.0, comms_return_char: str = '\n', ssh_config_file: Union[str, bool] = False, ssh_known_hosts_file: Union[str, bool] = False, on_init: Optional[Callable[..., Any]] = None, on_open: Optional[Callable[..., Any]] = None, on_close: Optional[Callable[..., Any]] = None, transport: str = 'system', transport_options: Optional[Dict[str, Any]] = None, channel_log: Union[str, bool, BytesIO] = False, channel_log_mode: str = 'write', channel_lock: bool = False, logging_uid: str = '', auth_secondary: str = '', failed_when_contains: Optional[List[str]] = None, textfsm_platform: str = 'cisco_nxos', genie_platform: str = 'nxos')

NXOSDriver Object

Please see scrapli.driver.base.base_driver.Driver for all "base driver" arguments!

noqa: DAR101

Parameters:

Name Type Description Default
privilege_levels Optional[Dict[str, PrivilegeLevel]]

optional user provided privilege levels, if left None will default to scrapli standard privilege levels

None
default_desired_privilege_level: string of name of default desired priv, this is the
    priv level that is generally used to disable paging/set terminal width and things
    like that upon first login, and is also the priv level scrapli will try to acquire
    for normal "command" operations (`send_command`, `send_commands`)
auth_secondary: password to use for secondary authentication (enable)
on_open: callable that accepts the class instance as its only argument. this callable,
    if provided, is executed immediately after authentication is completed. Common use
    cases for this callable would be to disable paging or accept any kind of banner
    message that prompts a user upon connection
on_close: callable that accepts the class instance as its only argument. this callable,
    if provided, is executed immediately prior to closing the underlying transport.
    Common use cases for this callable would be to save configurations prior to exiting,
    or to logout properly to free up vtys or similar.
textfsm_platform: string name of textfsm parser platform
genie_platform: string name of cisco genie parser platform
failed_when_contains: List of strings that indicate a command/config has failed

Returns:

Type Description

None

Source code in driver/core/cisco_nxos/async_driver.py
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def __init__(
    self,
    host: str,
    privilege_levels: Optional[Dict[str, PrivilegeLevel]] = None,
    default_desired_privilege_level: str = "privilege_exec",
    port: Optional[int] = None,
    auth_username: str = "",
    auth_password: str = "",
    auth_private_key: str = "",
    auth_private_key_passphrase: str = "",
    auth_strict_key: bool = True,
    auth_bypass: bool = False,
    timeout_socket: float = 15.0,
    timeout_transport: float = 30.0,
    timeout_ops: float = 30.0,
    comms_return_char: str = "\n",
    ssh_config_file: Union[str, bool] = False,
    ssh_known_hosts_file: Union[str, bool] = False,
    on_init: Optional[Callable[..., Any]] = None,
    on_open: Optional[Callable[..., Any]] = None,
    on_close: Optional[Callable[..., Any]] = None,
    transport: str = "system",
    transport_options: Optional[Dict[str, Any]] = None,
    channel_log: Union[str, bool, BytesIO] = False,
    channel_log_mode: str = "write",
    channel_lock: bool = False,
    logging_uid: str = "",
    auth_secondary: str = "",
    failed_when_contains: Optional[List[str]] = None,
    textfsm_platform: str = "cisco_nxos",
    genie_platform: str = "nxos",
):
    """
    NXOSDriver Object

    Please see `scrapli.driver.base.base_driver.Driver` for all "base driver" arguments!

    # noqa: DAR101

    Args:
         privilege_levels: optional user provided privilege levels, if left None will default to
            scrapli standard privilege levels
        default_desired_privilege_level: string of name of default desired priv, this is the
            priv level that is generally used to disable paging/set terminal width and things
            like that upon first login, and is also the priv level scrapli will try to acquire
            for normal "command" operations (`send_command`, `send_commands`)
        auth_secondary: password to use for secondary authentication (enable)
        on_open: callable that accepts the class instance as its only argument. this callable,
            if provided, is executed immediately after authentication is completed. Common use
            cases for this callable would be to disable paging or accept any kind of banner
            message that prompts a user upon connection
        on_close: callable that accepts the class instance as its only argument. this callable,
            if provided, is executed immediately prior to closing the underlying transport.
            Common use cases for this callable would be to save configurations prior to exiting,
            or to logout properly to free up vtys or similar.
        textfsm_platform: string name of textfsm parser platform
        genie_platform: string name of cisco genie parser platform
        failed_when_contains: List of strings that indicate a command/config has failed

    Returns:
        None

    Raises:
        N/A

    """
    # somewhere/somehow the mixin is causing mypy to be upset about comms_prompt_pattern...
    self.comms_prompt_pattern: str

    if privilege_levels is None:
        privilege_levels = deepcopy(PRIVS)

    if on_open is None:
        on_open = nxos_on_open
    if on_close is None:
        on_close = nxos_on_close

    if failed_when_contains is None:
        failed_when_contains = FAILED_WHEN_CONTAINS.copy()

    super().__init__(
        host=host,
        port=port,
        auth_username=auth_username,
        auth_password=auth_password,
        auth_private_key=auth_private_key,
        auth_private_key_passphrase=auth_private_key_passphrase,
        auth_strict_key=auth_strict_key,
        auth_bypass=auth_bypass,
        timeout_socket=timeout_socket,
        timeout_transport=timeout_transport,
        timeout_ops=timeout_ops,
        comms_return_char=comms_return_char,
        ssh_config_file=ssh_config_file,
        ssh_known_hosts_file=ssh_known_hosts_file,
        on_init=on_init,
        on_open=on_open,
        on_close=on_close,
        transport=transport,
        transport_options=transport_options,
        channel_log=channel_log,
        channel_log_mode=channel_log_mode,
        channel_lock=channel_lock,
        logging_uid=logging_uid,
        privilege_levels=privilege_levels,
        default_desired_privilege_level=default_desired_privilege_level,
        auth_secondary=auth_secondary,
        failed_when_contains=failed_when_contains,
        textfsm_platform=textfsm_platform,
        genie_platform=genie_platform,
    )

register_configuration_session(session_name: str) -> None

NXOS specific implementation of register_configuration_session

Parameters:

Name Type Description Default
session_name str

name of session to register

required

Returns:

Type Description
None

None

Source code in driver/core/cisco_nxos/async_driver.py
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def register_configuration_session(self, session_name: str) -> None:
    """
    NXOS specific implementation of register_configuration_session

    Args:
        session_name: name of session to register

    Returns:
        None

    Raises:
        N/A

    """
    self._create_configuration_session(session_name=session_name)
    self.update_privilege_levels()

nxos_on_close(conn: AsyncNetworkDriver) -> None async

AsyncNXOSDriver default on_close callable

Parameters:

Name Type Description Default
conn AsyncNetworkDriver

NetworkDriver object

required

Returns:

Type Description
None

None

Source code in driver/core/cisco_nxos/async_driver.py
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
async def nxos_on_close(conn: AsyncNetworkDriver) -> None:
    """
    AsyncNXOSDriver default on_close callable

    Args:
        conn: NetworkDriver object

    Returns:
        None

    Raises:
        N/A

    """
    await conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)
    conn.channel.write(channel_input="exit")
    conn.channel.send_return()

nxos_on_open(conn: AsyncNetworkDriver) -> None async

AsyncNXOSDriver default on_open callable

Parameters:

Name Type Description Default
conn AsyncNetworkDriver

NetworkDriver object

required

Returns:

Type Description
None

None

Source code in driver/core/cisco_nxos/async_driver.py
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
async def nxos_on_open(conn: AsyncNetworkDriver) -> None:
    """
    AsyncNXOSDriver default on_open callable

    Args:
        conn: NetworkDriver object

    Returns:
        None

    Raises:
        N/A

    """
    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 511")