Skip to content

sync_driver

scrapli.driver.network.sync_driver

NetworkDriver

Bases: GenericDriver, BaseNetworkDriver

Source code in driver/network/sync_driver.py
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 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
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
class NetworkDriver(GenericDriver, BaseNetworkDriver):
    def __init__(
        self,
        host: str,
        privilege_levels: Dict[str, PrivilegeLevel],
        default_desired_privilege_level: str,
        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 = "",
        genie_platform: str = "",
    ):
        # ensure type for comms_prompt_pattern exists before setting it in the mixin
        self.comms_prompt_pattern: str

        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,
        )

        self.auth_secondary = auth_secondary
        self.failed_when_contains = failed_when_contains or []
        self.textfsm_platform = textfsm_platform
        self.genie_platform = genie_platform

        self.privilege_levels = privilege_levels
        self.default_desired_privilege_level = default_desired_privilege_level
        self._priv_graph = defaultdict(set)
        self.update_privilege_levels()

    def _escalate(self, escalate_priv: PrivilegeLevel) -> None:
        """
        Escalate to the next privilege level up

        Args:
            escalate_priv: privilege level to escalate to

        Returns:
            None

        Raises:
            ScrapliAuthenticationFailed: if auth escalation timeout

        """
        self._pre_escalate(escalate_priv=escalate_priv)

        if escalate_priv.escalate_auth is False:
            self.channel.send_input(channel_input=escalate_priv.escalate)
        else:
            try:
                super().send_interactive(
                    interact_events=[
                        (escalate_priv.escalate, escalate_priv.escalate_prompt, False),
                        (self.auth_secondary, escalate_priv.pattern, True),
                    ],
                    interaction_complete_patterns=[
                        self.privilege_levels[escalate_priv.previous_priv].pattern,
                        escalate_priv.pattern,
                    ],
                )
            except ScrapliTimeout as exc:
                raise ScrapliAuthenticationFailed(
                    f"failed escalating privilege from '{escalate_priv.previous_priv}' to "
                    f"'{escalate_priv.name}'. do you need to set an 'auth_secondary' password?"
                ) from exc

    def _deescalate(self, current_priv: PrivilegeLevel) -> None:
        """
        Deescalate to the next privilege level down

        Args:
            current_priv: current privilege level

        Returns:
            None

        Raises:
            N/A

        """
        self.channel.send_input(channel_input=current_priv.deescalate)

    def acquire_priv(self, desired_priv: str) -> None:
        """
        Acquire desired priv level

        Args:
            desired_priv: string name of desired privilege level see
                `scrapli.driver.<driver_category.device_type>.driver` for levels

        Returns:
            None

        Raises:
            ScrapliPrivilegeError: if desired_priv cannot be attained

        """
        self._validate_privilege_level_name(privilege_level_name=desired_priv)

        privilege_change_count = 0

        while True:
            current_prompt = self.channel.get_prompt()
            privilege_action, target_priv = self._process_acquire_priv(
                destination_priv=desired_priv,
                current_prompt=current_prompt,
            )

            if privilege_action == PrivilegeAction.NO_ACTION:
                self._current_priv_level = target_priv
                return
            if privilege_action == PrivilegeAction.DEESCALATE:
                self._deescalate(current_priv=target_priv)
            if privilege_action == PrivilegeAction.ESCALATE:
                self._escalate(escalate_priv=target_priv)

            privilege_change_count += 1
            if privilege_change_count > len(self.privilege_levels) * 2:
                msg = f"Failed to acquire requested privilege level {desired_priv}"
                raise ScrapliPrivilegeError(msg)

    def _acquire_appropriate_privilege_level(self, privilege_level: str = "") -> None:
        """
        Acquire the appropriate priv level

        Acquires the "right" priv level based on generic_driver_mode, provided privilege level,
        and default desired privilege level. If in "generic_driver_mode" and no priv level is
        provided, we simply return as we are already at the "right" priv level (since we don't care
        about priv levels in this mode). If we are in "generic_driver_mode" and we are provided a
        priv level (this is only applicable in `send_interactive`) we will try to acquire that
        provided priv level. If a priv name is passed we try to resolve it and use that as the
        privilege level to acquire, otherwise if no priv leve is provided we will acquire the
        default_desired_privilege_level.

        Args:
            privilege_level: optional name of privilege level to acquire

        Returns:
            None

        Raises:
            N/A

        """
        if not privilege_level and self._generic_driver_mode is True:
            return

        if privilege_level:
            self._validate_privilege_level_name(privilege_level_name=privilege_level)
            resolved_privilege_level = privilege_level
        else:
            resolved_privilege_level = self.default_desired_privilege_level

        if self._current_priv_level.name != resolved_privilege_level:
            self.acquire_priv(desired_priv=resolved_privilege_level)

    def send_command(
        self,
        command: str,
        *,
        strip_prompt: bool = True,
        failed_when_contains: Optional[Union[str, List[str]]] = None,
        timeout_ops: Optional[float] = None,
    ) -> Response:
        """
        Send a command

        Super method will raise TypeError if anything but a string is passed here!

        Args:
            command: string to send to device in privilege exec mode
            strip_prompt: True/False strip prompt from returned output
            failed_when_contains: string or list of strings indicating failure if found in response
            timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
                the duration of the operation, value is reset to initial value after operation is
                completed

        Returns:
            Response: Scrapli Response object

        Raises:
            N/A

        """
        self._acquire_appropriate_privilege_level()

        if failed_when_contains is None:
            failed_when_contains = self.failed_when_contains

        response: Response = super().send_command(
            command=command,
            strip_prompt=strip_prompt,
            failed_when_contains=failed_when_contains,
            timeout_ops=timeout_ops,
        )
        self._update_response(response)

        return response

    def send_commands(
        self,
        commands: List[str],
        *,
        strip_prompt: bool = True,
        failed_when_contains: Optional[Union[str, List[str]]] = None,
        stop_on_failed: bool = False,
        eager: bool = False,
        timeout_ops: Optional[float] = None,
    ) -> MultiResponse:
        """
        Send multiple commands

        Super method will raise TypeError if anything but a list of strings is passed here!

        Args:
            commands: list of strings to send to device in privilege exec mode
            strip_prompt: True/False strip prompt from returned output
            failed_when_contains: string or list of strings indicating failure if found in response
            stop_on_failed: True/False stop executing commands if a command fails, returns results
                as of current execution
            eager: if eager is True we do not read until prompt is seen at each command sent to the
                channel. Do *not* use this unless you know what you are doing as it is possible that
                it can make scrapli less reliable!
            timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
                the duration of the operation, value is reset to initial value after operation is
                completed. Note that this is the timeout value PER COMMAND sent, not for the total
                of the commands being sent!

        Returns:
            MultiResponse: Scrapli MultiResponse object

        Raises:
            N/A

        """
        self._acquire_appropriate_privilege_level()

        if failed_when_contains is None:
            failed_when_contains = self.failed_when_contains

        responses = super().send_commands(
            commands=commands,
            strip_prompt=strip_prompt,
            failed_when_contains=failed_when_contains,
            stop_on_failed=stop_on_failed,
            eager=eager,
            timeout_ops=timeout_ops,
        )

        for response in responses:
            self._update_response(response=response)

        return responses

    def send_commands_from_file(
        self,
        file: str,
        *,
        strip_prompt: bool = True,
        failed_when_contains: Optional[Union[str, List[str]]] = None,
        stop_on_failed: bool = False,
        eager: bool = False,
        timeout_ops: Optional[float] = None,
    ) -> MultiResponse:
        """
        Send command(s) from file

        Args:
            file: string path to file
            strip_prompt: True/False strip prompt from returned output
            failed_when_contains: string or list of strings indicating failure if found in response
            stop_on_failed: True/False stop executing commands if a command fails, returns results
                as of current execution
            eager: if eager is True we do not read until prompt is seen at each command sent to the
                channel. Do *not* use this unless you know what you are doing as it is possible that
                it can make scrapli less reliable!
            timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
                the duration of the operation, value is reset to initial value after operation is
                completed. Note that this is the timeout value PER COMMAND sent, not for the total
                of the commands being sent!

        Returns:
            MultiResponse: Scrapli MultiResponse object

        Raises:
            N/A

        """
        self._acquire_appropriate_privilege_level()

        if failed_when_contains is None:
            failed_when_contains = self.failed_when_contains

        return super().send_commands_from_file(
            file=file,
            strip_prompt=strip_prompt,
            failed_when_contains=failed_when_contains,
            stop_on_failed=stop_on_failed,
            eager=eager,
            timeout_ops=timeout_ops,
        )

    def send_interactive(
        self,
        interact_events: Union[List[Tuple[str, str]], List[Tuple[str, str, bool]]],
        *,
        failed_when_contains: Optional[Union[str, List[str]]] = None,
        privilege_level: str = "",
        timeout_ops: Optional[float] = None,
        interaction_complete_patterns: Optional[List[str]] = None,
    ) -> Response:
        """
        Interact with a device with changing prompts per input.

        Used to interact with devices where prompts change per input, and where inputs may be hidden
        such as in the case of a password input. This can be used to respond to challenges from
        devices such as the confirmation for the command "clear logging" on IOSXE devices for
        example. You may have as many elements in the "interact_events" list as needed, and each
        element of that list should be a tuple of two or three elements. The first element is always
        the input to send as a string, the second should be the expected response as a string, and
        the optional third a bool for whether or not the input is "hidden" (i.e. password input)

        An example where we need this sort of capability:

        '''
        3560CX#copy flash: scp:
        Source filename []? test1.txt
        Address or name of remote host []? 172.31.254.100
        Destination username [carl]?
        Writing test1.txt
        Password:

        Password:
         Sink: C0644 639 test1.txt
        !
        639 bytes copied in 12.066 secs (53 bytes/sec)
        3560CX#
        '''

        To accomplish this we can use the following:

        '''
        interact = conn.channel.send_inputs_interact(
            [
                ("copy flash: scp:", "Source filename []?", False),
                ("test1.txt", "Address or name of remote host []?", False),
                ("172.31.254.100", "Destination username [carl]?", False),
                ("carl", "Password:", False),
                ("super_secure_password", prompt, True),
            ]
        )
        '''

        If we needed to deal with more prompts we could simply continue adding tuples to the list of
        interact "events".

        Args:
            interact_events: list of tuples containing the "interactions" with the device
                each list element must have an input and an expected response, and may have an
                optional bool for the third and final element -- the optional bool specifies if the
                input that is sent to the device is "hidden" (ex: password), if the hidden param is
                not provided it is assumed the input is "normal" (not hidden)
            failed_when_contains: list of strings that, if present in final output, represent a
                failed command/interaction
            privilege_level: name of the privilege level to operate in
            timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
                the duration of the operation, value is reset to initial value after operation is
                completed. Note that this is the timeout value PER COMMAND sent, not for the total
                of the commands being sent!
            interaction_complete_patterns: list of patterns, that if seen, indicate the interactive
                "session" has ended and we should exit the interactive session.

        Returns:
            Response: scrapli Response object

        Raises:
            N/A

        """
        self._acquire_appropriate_privilege_level(privilege_level=privilege_level)

        if failed_when_contains is None:
            failed_when_contains = self.failed_when_contains

        # type hint is due to the timeout_modifier wrapper returning `Any` so that we dont anger the
        # asyncio parts (which will get an awaitable not a Response returned)
        response: Response = super().send_interactive(
            interact_events=interact_events,
            failed_when_contains=failed_when_contains,
            timeout_ops=timeout_ops,
            interaction_complete_patterns=interaction_complete_patterns,
        )
        self._update_response(response=response)

        return response

    def _abort_config(self) -> None:
        """
        Abort a configuration operation/session if applicable (for config sessions like junos/iosxr)

        Args:
            N/A

        Returns:
            None

        Raises:
            N/A

        """

    def send_configs(
        self,
        configs: List[str],
        *,
        strip_prompt: bool = True,
        failed_when_contains: Optional[Union[str, List[str]]] = None,
        stop_on_failed: bool = False,
        privilege_level: str = "",
        eager: bool = False,
        timeout_ops: Optional[float] = None,
    ) -> MultiResponse:
        """
        Send configuration(s)

        Args:
            configs: list of strings to send to device in config mode
            strip_prompt: True/False strip prompt from returned output
            failed_when_contains: string or list of strings indicating failure if found in response
            stop_on_failed: True/False stop executing commands if a command fails, returns results
                as of current execution; aborts configuration session if applicable (iosxr/junos or
                eos/nxos if using a configuration session)
            privilege_level: name of configuration privilege level/type to acquire; this is platform
                dependent, so check the device driver for specifics. Examples of privilege_name
                would be "configuration_exclusive" for IOSXRDriver, or "configuration_private" for
                JunosDriver. You can also pass in a name of a configuration session such as
                "my-config-session" if you have registered a session using the
                "register_config_session" method of the EOSDriver or NXOSDriver.
            eager: if eager is True we do not read until prompt is seen at each command sent to the
                channel. Do *not* use this unless you know what you are doing as it is possible that
                it can make scrapli less reliable!
            timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
                the duration of the operation, value is reset to initial value after operation is
                completed. Note that this is the timeout value PER CONFIG sent, not for the total
                of the configs being sent!

        Returns:
            MultiResponse: Scrapli MultiResponse object

        Raises:
            N/A

        """
        resolved_privilege_level, failed_when_contains = self._pre_send_configs(
            configs=configs,
            failed_when_contains=failed_when_contains,
            privilege_level=privilege_level,
        )

        if self._current_priv_level.name != resolved_privilege_level:
            self.acquire_priv(desired_priv=resolved_privilege_level)

        responses = super().send_commands(
            commands=configs,
            strip_prompt=strip_prompt,
            failed_when_contains=failed_when_contains,
            stop_on_failed=stop_on_failed,
            eager=eager,
            timeout_ops=timeout_ops,
        )

        if stop_on_failed and responses.failed:
            self._abort_config()

        return self._post_send_configs(responses=responses)

    def send_config(
        self,
        config: str,
        *,
        strip_prompt: bool = True,
        failed_when_contains: Optional[Union[str, List[str]]] = None,
        stop_on_failed: bool = False,
        privilege_level: str = "",
        eager: bool = False,
        timeout_ops: Optional[float] = None,
    ) -> Response:
        """
        Send configuration string

        Args:
            config: string configuration to send to the device, supports sending multi-line strings
            strip_prompt: True/False strip prompt from returned output
            failed_when_contains: string or list of strings indicating failure if found in response
            stop_on_failed: True/False stop executing commands if a command fails, returns results
                as of current execution; aborts configuration session if applicable (iosxr/junos or
                eos/nxos if using a configuration session)
            privilege_level: name of configuration privilege level/type to acquire; this is platform
                dependent, so check the device driver for specifics. Examples of privilege_name
                would be "configuration_exclusive" for IOSXRDriver, or "configuration_private" for
                JunosDriver. You can also pass in a name of a configuration session such as
                "my-config-session" if you have registered a session using the
                "register_config_session" method of the EOSDriver or NXOSDriver.
            eager: if eager is True we do not read until prompt is seen at each command sent to the
                channel. Do *not* use this unless you know what you are doing as it is possible that
                it can make scrapli less reliable!
            timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
                the duration of the operation, value is reset to initial value after operation is
                completed. Note that this is the timeout value PER CONFIG sent, not for the total
                of the configs being sent!

        Returns:
            Response: Scrapli Response object

        Raises:
            N/A

        """
        split_config = self._pre_send_config(config=config)

        # now that we have a list of configs, just use send_configs to actually execute them
        multi_response = self.send_configs(
            configs=split_config,
            strip_prompt=strip_prompt,
            failed_when_contains=failed_when_contains,
            stop_on_failed=stop_on_failed,
            privilege_level=privilege_level,
            eager=eager,
            timeout_ops=timeout_ops,
        )
        return self._post_send_config(config=config, multi_response=multi_response)

    def send_configs_from_file(
        self,
        file: str,
        *,
        strip_prompt: bool = True,
        failed_when_contains: Optional[Union[str, List[str]]] = None,
        stop_on_failed: bool = False,
        privilege_level: str = "",
        eager: bool = False,
        timeout_ops: Optional[float] = None,
    ) -> MultiResponse:
        """
        Send configuration(s) from a file

        Args:
            file: string path to file
            strip_prompt: True/False strip prompt from returned output
            failed_when_contains: string or list of strings indicating failure if found in response
            stop_on_failed: True/False stop executing commands if a command fails, returns results
                as of current execution; aborts configuration session if applicable (iosxr/junos or
                eos/nxos if using a configuration session)
            privilege_level: name of configuration privilege level/type to acquire; this is platform
                dependent, so check the device driver for specifics. Examples of privilege_name
                would be "exclusive" for IOSXRDriver, "private" for JunosDriver. You can also pass
                in a name of a configuration session such as "session_mysession" if you have
                registered a session using the "register_config_session" method of the EOSDriver or
                NXOSDriver.
            eager: if eager is True we do not read until prompt is seen at each command sent to the
                channel. Do *not* use this unless you know what you are doing as it is possible that
                it can make scrapli less reliable!
            timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
                the duration of the operation, value is reset to initial value after operation is
                completed. Note that this is the timeout value PER CONFIG sent, not for the total
                of the configs being sent!

        Returns:
            MultiResponse: Scrapli MultiResponse object

        Raises:
            N/A

        """
        configs = self._pre_send_from_file(file=file, caller="send_configs_from_file")

        return self.send_configs(
            configs=configs,
            strip_prompt=strip_prompt,
            failed_when_contains=failed_when_contains,
            stop_on_failed=stop_on_failed,
            privilege_level=privilege_level,
            eager=eager,
            timeout_ops=timeout_ops,
        )

acquire_priv(desired_priv: str) -> None

Acquire desired priv level

Parameters:

Name Type Description Default
desired_priv str

string name of desired privilege level see scrapli.driver.<driver_category.device_type>.driver for levels

required

Returns:

Type Description
None

None

Raises:

Type Description
ScrapliPrivilegeError

if desired_priv cannot be attained

Source code in driver/network/sync_driver.py
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
def acquire_priv(self, desired_priv: str) -> None:
    """
    Acquire desired priv level

    Args:
        desired_priv: string name of desired privilege level see
            `scrapli.driver.<driver_category.device_type>.driver` for levels

    Returns:
        None

    Raises:
        ScrapliPrivilegeError: if desired_priv cannot be attained

    """
    self._validate_privilege_level_name(privilege_level_name=desired_priv)

    privilege_change_count = 0

    while True:
        current_prompt = self.channel.get_prompt()
        privilege_action, target_priv = self._process_acquire_priv(
            destination_priv=desired_priv,
            current_prompt=current_prompt,
        )

        if privilege_action == PrivilegeAction.NO_ACTION:
            self._current_priv_level = target_priv
            return
        if privilege_action == PrivilegeAction.DEESCALATE:
            self._deescalate(current_priv=target_priv)
        if privilege_action == PrivilegeAction.ESCALATE:
            self._escalate(escalate_priv=target_priv)

        privilege_change_count += 1
        if privilege_change_count > len(self.privilege_levels) * 2:
            msg = f"Failed to acquire requested privilege level {desired_priv}"
            raise ScrapliPrivilegeError(msg)

send_command(command: str, *, strip_prompt: bool = True, failed_when_contains: Optional[Union[str, List[str]]] = None, timeout_ops: Optional[float] = None) -> Response

Send a command

Super method will raise TypeError if anything but a string is passed here!

Parameters:

Name Type Description Default
command str

string to send to device in privilege exec mode

required
strip_prompt bool

True/False strip prompt from returned output

True
failed_when_contains Optional[Union[str, List[str]]]

string or list of strings indicating failure if found in response

None
timeout_ops Optional[float]

timeout ops value for this operation; only sets the timeout_ops value for the duration of the operation, value is reset to initial value after operation is completed

None

Returns:

Name Type Description
Response Response

Scrapli Response object

Source code in driver/network/sync_driver.py
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def send_command(
    self,
    command: str,
    *,
    strip_prompt: bool = True,
    failed_when_contains: Optional[Union[str, List[str]]] = None,
    timeout_ops: Optional[float] = None,
) -> Response:
    """
    Send a command

    Super method will raise TypeError if anything but a string is passed here!

    Args:
        command: string to send to device in privilege exec mode
        strip_prompt: True/False strip prompt from returned output
        failed_when_contains: string or list of strings indicating failure if found in response
        timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
            the duration of the operation, value is reset to initial value after operation is
            completed

    Returns:
        Response: Scrapli Response object

    Raises:
        N/A

    """
    self._acquire_appropriate_privilege_level()

    if failed_when_contains is None:
        failed_when_contains = self.failed_when_contains

    response: Response = super().send_command(
        command=command,
        strip_prompt=strip_prompt,
        failed_when_contains=failed_when_contains,
        timeout_ops=timeout_ops,
    )
    self._update_response(response)

    return response

send_commands(commands: List[str], *, strip_prompt: bool = True, failed_when_contains: Optional[Union[str, List[str]]] = None, stop_on_failed: bool = False, eager: bool = False, timeout_ops: Optional[float] = None) -> MultiResponse

Send multiple commands

Super method will raise TypeError if anything but a list of strings is passed here!

Parameters:

Name Type Description Default
commands List[str]

list of strings to send to device in privilege exec mode

required
strip_prompt bool

True/False strip prompt from returned output

True
failed_when_contains Optional[Union[str, List[str]]]

string or list of strings indicating failure if found in response

None
stop_on_failed bool

True/False stop executing commands if a command fails, returns results as of current execution

False
eager bool

if eager is True we do not read until prompt is seen at each command sent to the channel. Do not use this unless you know what you are doing as it is possible that it can make scrapli less reliable!

False
timeout_ops Optional[float]

timeout ops value for this operation; only sets the timeout_ops value for the duration of the operation, value is reset to initial value after operation is completed. Note that this is the timeout value PER COMMAND sent, not for the total of the commands being sent!

None

Returns:

Name Type Description
MultiResponse MultiResponse

Scrapli MultiResponse object

Source code in driver/network/sync_driver.py
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
def send_commands(
    self,
    commands: List[str],
    *,
    strip_prompt: bool = True,
    failed_when_contains: Optional[Union[str, List[str]]] = None,
    stop_on_failed: bool = False,
    eager: bool = False,
    timeout_ops: Optional[float] = None,
) -> MultiResponse:
    """
    Send multiple commands

    Super method will raise TypeError if anything but a list of strings is passed here!

    Args:
        commands: list of strings to send to device in privilege exec mode
        strip_prompt: True/False strip prompt from returned output
        failed_when_contains: string or list of strings indicating failure if found in response
        stop_on_failed: True/False stop executing commands if a command fails, returns results
            as of current execution
        eager: if eager is True we do not read until prompt is seen at each command sent to the
            channel. Do *not* use this unless you know what you are doing as it is possible that
            it can make scrapli less reliable!
        timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
            the duration of the operation, value is reset to initial value after operation is
            completed. Note that this is the timeout value PER COMMAND sent, not for the total
            of the commands being sent!

    Returns:
        MultiResponse: Scrapli MultiResponse object

    Raises:
        N/A

    """
    self._acquire_appropriate_privilege_level()

    if failed_when_contains is None:
        failed_when_contains = self.failed_when_contains

    responses = super().send_commands(
        commands=commands,
        strip_prompt=strip_prompt,
        failed_when_contains=failed_when_contains,
        stop_on_failed=stop_on_failed,
        eager=eager,
        timeout_ops=timeout_ops,
    )

    for response in responses:
        self._update_response(response=response)

    return responses

send_commands_from_file(file: str, *, strip_prompt: bool = True, failed_when_contains: Optional[Union[str, List[str]]] = None, stop_on_failed: bool = False, eager: bool = False, timeout_ops: Optional[float] = None) -> MultiResponse

Send command(s) from file

Parameters:

Name Type Description Default
file str

string path to file

required
strip_prompt bool

True/False strip prompt from returned output

True
failed_when_contains Optional[Union[str, List[str]]]

string or list of strings indicating failure if found in response

None
stop_on_failed bool

True/False stop executing commands if a command fails, returns results as of current execution

False
eager bool

if eager is True we do not read until prompt is seen at each command sent to the channel. Do not use this unless you know what you are doing as it is possible that it can make scrapli less reliable!

False
timeout_ops Optional[float]

timeout ops value for this operation; only sets the timeout_ops value for the duration of the operation, value is reset to initial value after operation is completed. Note that this is the timeout value PER COMMAND sent, not for the total of the commands being sent!

None

Returns:

Name Type Description
MultiResponse MultiResponse

Scrapli MultiResponse object

Source code in driver/network/sync_driver.py
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
def send_commands_from_file(
    self,
    file: str,
    *,
    strip_prompt: bool = True,
    failed_when_contains: Optional[Union[str, List[str]]] = None,
    stop_on_failed: bool = False,
    eager: bool = False,
    timeout_ops: Optional[float] = None,
) -> MultiResponse:
    """
    Send command(s) from file

    Args:
        file: string path to file
        strip_prompt: True/False strip prompt from returned output
        failed_when_contains: string or list of strings indicating failure if found in response
        stop_on_failed: True/False stop executing commands if a command fails, returns results
            as of current execution
        eager: if eager is True we do not read until prompt is seen at each command sent to the
            channel. Do *not* use this unless you know what you are doing as it is possible that
            it can make scrapli less reliable!
        timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
            the duration of the operation, value is reset to initial value after operation is
            completed. Note that this is the timeout value PER COMMAND sent, not for the total
            of the commands being sent!

    Returns:
        MultiResponse: Scrapli MultiResponse object

    Raises:
        N/A

    """
    self._acquire_appropriate_privilege_level()

    if failed_when_contains is None:
        failed_when_contains = self.failed_when_contains

    return super().send_commands_from_file(
        file=file,
        strip_prompt=strip_prompt,
        failed_when_contains=failed_when_contains,
        stop_on_failed=stop_on_failed,
        eager=eager,
        timeout_ops=timeout_ops,
    )

send_config(config: str, *, strip_prompt: bool = True, failed_when_contains: Optional[Union[str, List[str]]] = None, stop_on_failed: bool = False, privilege_level: str = '', eager: bool = False, timeout_ops: Optional[float] = None) -> Response

Send configuration string

Parameters:

Name Type Description Default
config str

string configuration to send to the device, supports sending multi-line strings

required
strip_prompt bool

True/False strip prompt from returned output

True
failed_when_contains Optional[Union[str, List[str]]]

string or list of strings indicating failure if found in response

None
stop_on_failed bool

True/False stop executing commands if a command fails, returns results as of current execution; aborts configuration session if applicable (iosxr/junos or eos/nxos if using a configuration session)

False
privilege_level str

name of configuration privilege level/type to acquire; this is platform dependent, so check the device driver for specifics. Examples of privilege_name would be "configuration_exclusive" for IOSXRDriver, or "configuration_private" for JunosDriver. You can also pass in a name of a configuration session such as "my-config-session" if you have registered a session using the "register_config_session" method of the EOSDriver or NXOSDriver.

''
eager bool

if eager is True we do not read until prompt is seen at each command sent to the channel. Do not use this unless you know what you are doing as it is possible that it can make scrapli less reliable!

False
timeout_ops Optional[float]

timeout ops value for this operation; only sets the timeout_ops value for the duration of the operation, value is reset to initial value after operation is completed. Note that this is the timeout value PER CONFIG sent, not for the total of the configs being sent!

None

Returns:

Name Type Description
Response Response

Scrapli Response object

Source code in driver/network/sync_driver.py
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
def send_config(
    self,
    config: str,
    *,
    strip_prompt: bool = True,
    failed_when_contains: Optional[Union[str, List[str]]] = None,
    stop_on_failed: bool = False,
    privilege_level: str = "",
    eager: bool = False,
    timeout_ops: Optional[float] = None,
) -> Response:
    """
    Send configuration string

    Args:
        config: string configuration to send to the device, supports sending multi-line strings
        strip_prompt: True/False strip prompt from returned output
        failed_when_contains: string or list of strings indicating failure if found in response
        stop_on_failed: True/False stop executing commands if a command fails, returns results
            as of current execution; aborts configuration session if applicable (iosxr/junos or
            eos/nxos if using a configuration session)
        privilege_level: name of configuration privilege level/type to acquire; this is platform
            dependent, so check the device driver for specifics. Examples of privilege_name
            would be "configuration_exclusive" for IOSXRDriver, or "configuration_private" for
            JunosDriver. You can also pass in a name of a configuration session such as
            "my-config-session" if you have registered a session using the
            "register_config_session" method of the EOSDriver or NXOSDriver.
        eager: if eager is True we do not read until prompt is seen at each command sent to the
            channel. Do *not* use this unless you know what you are doing as it is possible that
            it can make scrapli less reliable!
        timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
            the duration of the operation, value is reset to initial value after operation is
            completed. Note that this is the timeout value PER CONFIG sent, not for the total
            of the configs being sent!

    Returns:
        Response: Scrapli Response object

    Raises:
        N/A

    """
    split_config = self._pre_send_config(config=config)

    # now that we have a list of configs, just use send_configs to actually execute them
    multi_response = self.send_configs(
        configs=split_config,
        strip_prompt=strip_prompt,
        failed_when_contains=failed_when_contains,
        stop_on_failed=stop_on_failed,
        privilege_level=privilege_level,
        eager=eager,
        timeout_ops=timeout_ops,
    )
    return self._post_send_config(config=config, multi_response=multi_response)

send_configs(configs: List[str], *, strip_prompt: bool = True, failed_when_contains: Optional[Union[str, List[str]]] = None, stop_on_failed: bool = False, privilege_level: str = '', eager: bool = False, timeout_ops: Optional[float] = None) -> MultiResponse

Send configuration(s)

Parameters:

Name Type Description Default
configs List[str]

list of strings to send to device in config mode

required
strip_prompt bool

True/False strip prompt from returned output

True
failed_when_contains Optional[Union[str, List[str]]]

string or list of strings indicating failure if found in response

None
stop_on_failed bool

True/False stop executing commands if a command fails, returns results as of current execution; aborts configuration session if applicable (iosxr/junos or eos/nxos if using a configuration session)

False
privilege_level str

name of configuration privilege level/type to acquire; this is platform dependent, so check the device driver for specifics. Examples of privilege_name would be "configuration_exclusive" for IOSXRDriver, or "configuration_private" for JunosDriver. You can also pass in a name of a configuration session such as "my-config-session" if you have registered a session using the "register_config_session" method of the EOSDriver or NXOSDriver.

''
eager bool

if eager is True we do not read until prompt is seen at each command sent to the channel. Do not use this unless you know what you are doing as it is possible that it can make scrapli less reliable!

False
timeout_ops Optional[float]

timeout ops value for this operation; only sets the timeout_ops value for the duration of the operation, value is reset to initial value after operation is completed. Note that this is the timeout value PER CONFIG sent, not for the total of the configs being sent!

None

Returns:

Name Type Description
MultiResponse MultiResponse

Scrapli MultiResponse object

Source code in driver/network/sync_driver.py
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
def send_configs(
    self,
    configs: List[str],
    *,
    strip_prompt: bool = True,
    failed_when_contains: Optional[Union[str, List[str]]] = None,
    stop_on_failed: bool = False,
    privilege_level: str = "",
    eager: bool = False,
    timeout_ops: Optional[float] = None,
) -> MultiResponse:
    """
    Send configuration(s)

    Args:
        configs: list of strings to send to device in config mode
        strip_prompt: True/False strip prompt from returned output
        failed_when_contains: string or list of strings indicating failure if found in response
        stop_on_failed: True/False stop executing commands if a command fails, returns results
            as of current execution; aborts configuration session if applicable (iosxr/junos or
            eos/nxos if using a configuration session)
        privilege_level: name of configuration privilege level/type to acquire; this is platform
            dependent, so check the device driver for specifics. Examples of privilege_name
            would be "configuration_exclusive" for IOSXRDriver, or "configuration_private" for
            JunosDriver. You can also pass in a name of a configuration session such as
            "my-config-session" if you have registered a session using the
            "register_config_session" method of the EOSDriver or NXOSDriver.
        eager: if eager is True we do not read until prompt is seen at each command sent to the
            channel. Do *not* use this unless you know what you are doing as it is possible that
            it can make scrapli less reliable!
        timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
            the duration of the operation, value is reset to initial value after operation is
            completed. Note that this is the timeout value PER CONFIG sent, not for the total
            of the configs being sent!

    Returns:
        MultiResponse: Scrapli MultiResponse object

    Raises:
        N/A

    """
    resolved_privilege_level, failed_when_contains = self._pre_send_configs(
        configs=configs,
        failed_when_contains=failed_when_contains,
        privilege_level=privilege_level,
    )

    if self._current_priv_level.name != resolved_privilege_level:
        self.acquire_priv(desired_priv=resolved_privilege_level)

    responses = super().send_commands(
        commands=configs,
        strip_prompt=strip_prompt,
        failed_when_contains=failed_when_contains,
        stop_on_failed=stop_on_failed,
        eager=eager,
        timeout_ops=timeout_ops,
    )

    if stop_on_failed and responses.failed:
        self._abort_config()

    return self._post_send_configs(responses=responses)

send_configs_from_file(file: str, *, strip_prompt: bool = True, failed_when_contains: Optional[Union[str, List[str]]] = None, stop_on_failed: bool = False, privilege_level: str = '', eager: bool = False, timeout_ops: Optional[float] = None) -> MultiResponse

Send configuration(s) from a file

Parameters:

Name Type Description Default
file str

string path to file

required
strip_prompt bool

True/False strip prompt from returned output

True
failed_when_contains Optional[Union[str, List[str]]]

string or list of strings indicating failure if found in response

None
stop_on_failed bool

True/False stop executing commands if a command fails, returns results as of current execution; aborts configuration session if applicable (iosxr/junos or eos/nxos if using a configuration session)

False
privilege_level str

name of configuration privilege level/type to acquire; this is platform dependent, so check the device driver for specifics. Examples of privilege_name would be "exclusive" for IOSXRDriver, "private" for JunosDriver. You can also pass in a name of a configuration session such as "session_mysession" if you have registered a session using the "register_config_session" method of the EOSDriver or NXOSDriver.

''
eager bool

if eager is True we do not read until prompt is seen at each command sent to the channel. Do not use this unless you know what you are doing as it is possible that it can make scrapli less reliable!

False
timeout_ops Optional[float]

timeout ops value for this operation; only sets the timeout_ops value for the duration of the operation, value is reset to initial value after operation is completed. Note that this is the timeout value PER CONFIG sent, not for the total of the configs being sent!

None

Returns:

Name Type Description
MultiResponse MultiResponse

Scrapli MultiResponse object

Source code in driver/network/sync_driver.py
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
def send_configs_from_file(
    self,
    file: str,
    *,
    strip_prompt: bool = True,
    failed_when_contains: Optional[Union[str, List[str]]] = None,
    stop_on_failed: bool = False,
    privilege_level: str = "",
    eager: bool = False,
    timeout_ops: Optional[float] = None,
) -> MultiResponse:
    """
    Send configuration(s) from a file

    Args:
        file: string path to file
        strip_prompt: True/False strip prompt from returned output
        failed_when_contains: string or list of strings indicating failure if found in response
        stop_on_failed: True/False stop executing commands if a command fails, returns results
            as of current execution; aborts configuration session if applicable (iosxr/junos or
            eos/nxos if using a configuration session)
        privilege_level: name of configuration privilege level/type to acquire; this is platform
            dependent, so check the device driver for specifics. Examples of privilege_name
            would be "exclusive" for IOSXRDriver, "private" for JunosDriver. You can also pass
            in a name of a configuration session such as "session_mysession" if you have
            registered a session using the "register_config_session" method of the EOSDriver or
            NXOSDriver.
        eager: if eager is True we do not read until prompt is seen at each command sent to the
            channel. Do *not* use this unless you know what you are doing as it is possible that
            it can make scrapli less reliable!
        timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
            the duration of the operation, value is reset to initial value after operation is
            completed. Note that this is the timeout value PER CONFIG sent, not for the total
            of the configs being sent!

    Returns:
        MultiResponse: Scrapli MultiResponse object

    Raises:
        N/A

    """
    configs = self._pre_send_from_file(file=file, caller="send_configs_from_file")

    return self.send_configs(
        configs=configs,
        strip_prompt=strip_prompt,
        failed_when_contains=failed_when_contains,
        stop_on_failed=stop_on_failed,
        privilege_level=privilege_level,
        eager=eager,
        timeout_ops=timeout_ops,
    )

send_interactive(interact_events: Union[List[Tuple[str, str]], List[Tuple[str, str, bool]]], *, failed_when_contains: Optional[Union[str, List[str]]] = None, privilege_level: str = '', timeout_ops: Optional[float] = None, interaction_complete_patterns: Optional[List[str]] = None) -> Response

Interact with a device with changing prompts per input.

Used to interact with devices where prompts change per input, and where inputs may be hidden such as in the case of a password input. This can be used to respond to challenges from devices such as the confirmation for the command "clear logging" on IOSXE devices for example. You may have as many elements in the "interact_events" list as needed, and each element of that list should be a tuple of two or three elements. The first element is always the input to send as a string, the second should be the expected response as a string, and the optional third a bool for whether or not the input is "hidden" (i.e. password input)

An example where we need this sort of capability:

''' 3560CX#copy flash: scp: Source filename []? test1.txt Address or name of remote host []? 172.31.254.100 Destination username [carl]? Writing test1.txt Password:

Password

Sink: C0644 639 test1.txt

! 639 bytes copied in 12.066 secs (53 bytes/sec) 3560CX# '''

To accomplish this we can use the following:

''' interact = conn.channel.send_inputs_interact( [ ("copy flash: scp:", "Source filename []?", False), ("test1.txt", "Address or name of remote host []?", False), ("172.31.254.100", "Destination username [carl]?", False), ("carl", "Password:", False), ("super_secure_password", prompt, True), ] ) '''

If we needed to deal with more prompts we could simply continue adding tuples to the list of interact "events".

Parameters:

Name Type Description Default
interact_events Union[List[Tuple[str, str]], List[Tuple[str, str, bool]]]

list of tuples containing the "interactions" with the device each list element must have an input and an expected response, and may have an optional bool for the third and final element -- the optional bool specifies if the input that is sent to the device is "hidden" (ex: password), if the hidden param is not provided it is assumed the input is "normal" (not hidden)

required
failed_when_contains Optional[Union[str, List[str]]]

list of strings that, if present in final output, represent a failed command/interaction

None
privilege_level str

name of the privilege level to operate in

''
timeout_ops Optional[float]

timeout ops value for this operation; only sets the timeout_ops value for the duration of the operation, value is reset to initial value after operation is completed. Note that this is the timeout value PER COMMAND sent, not for the total of the commands being sent!

None
interaction_complete_patterns Optional[List[str]]

list of patterns, that if seen, indicate the interactive "session" has ended and we should exit the interactive session.

None

Returns:

Name Type Description
Response Response

scrapli Response object

Source code in driver/network/sync_driver.py
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
def send_interactive(
    self,
    interact_events: Union[List[Tuple[str, str]], List[Tuple[str, str, bool]]],
    *,
    failed_when_contains: Optional[Union[str, List[str]]] = None,
    privilege_level: str = "",
    timeout_ops: Optional[float] = None,
    interaction_complete_patterns: Optional[List[str]] = None,
) -> Response:
    """
    Interact with a device with changing prompts per input.

    Used to interact with devices where prompts change per input, and where inputs may be hidden
    such as in the case of a password input. This can be used to respond to challenges from
    devices such as the confirmation for the command "clear logging" on IOSXE devices for
    example. You may have as many elements in the "interact_events" list as needed, and each
    element of that list should be a tuple of two or three elements. The first element is always
    the input to send as a string, the second should be the expected response as a string, and
    the optional third a bool for whether or not the input is "hidden" (i.e. password input)

    An example where we need this sort of capability:

    '''
    3560CX#copy flash: scp:
    Source filename []? test1.txt
    Address or name of remote host []? 172.31.254.100
    Destination username [carl]?
    Writing test1.txt
    Password:

    Password:
     Sink: C0644 639 test1.txt
    !
    639 bytes copied in 12.066 secs (53 bytes/sec)
    3560CX#
    '''

    To accomplish this we can use the following:

    '''
    interact = conn.channel.send_inputs_interact(
        [
            ("copy flash: scp:", "Source filename []?", False),
            ("test1.txt", "Address or name of remote host []?", False),
            ("172.31.254.100", "Destination username [carl]?", False),
            ("carl", "Password:", False),
            ("super_secure_password", prompt, True),
        ]
    )
    '''

    If we needed to deal with more prompts we could simply continue adding tuples to the list of
    interact "events".

    Args:
        interact_events: list of tuples containing the "interactions" with the device
            each list element must have an input and an expected response, and may have an
            optional bool for the third and final element -- the optional bool specifies if the
            input that is sent to the device is "hidden" (ex: password), if the hidden param is
            not provided it is assumed the input is "normal" (not hidden)
        failed_when_contains: list of strings that, if present in final output, represent a
            failed command/interaction
        privilege_level: name of the privilege level to operate in
        timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
            the duration of the operation, value is reset to initial value after operation is
            completed. Note that this is the timeout value PER COMMAND sent, not for the total
            of the commands being sent!
        interaction_complete_patterns: list of patterns, that if seen, indicate the interactive
            "session" has ended and we should exit the interactive session.

    Returns:
        Response: scrapli Response object

    Raises:
        N/A

    """
    self._acquire_appropriate_privilege_level(privilege_level=privilege_level)

    if failed_when_contains is None:
        failed_when_contains = self.failed_when_contains

    # type hint is due to the timeout_modifier wrapper returning `Any` so that we dont anger the
    # asyncio parts (which will get an awaitable not a Response returned)
    response: Response = super().send_interactive(
        interact_events=interact_events,
        failed_when_contains=failed_when_contains,
        timeout_ops=timeout_ops,
        interaction_complete_patterns=interaction_complete_patterns,
    )
    self._update_response(response=response)

    return response