Skip to content

sync_channel

scrapli.channel.sync_channel

Channel

Bases: BaseChannel

Source code in channel/sync_channel.py
 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
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
class Channel(BaseChannel):
    def __init__(
        self,
        transport: Transport,
        base_channel_args: BaseChannelArgs,
    ) -> None:
        super().__init__(
            transport=transport,
            base_channel_args=base_channel_args,
        )
        self.transport: Transport

        self.channel_lock: Optional[Lock] = None
        if self._base_channel_args.channel_lock:
            self.channel_lock = Lock()

    @contextmanager
    def _channel_lock(self) -> Iterator[None]:
        """
        Lock the channel during public channel operations if channel_lock is enabled

        Args:
            N/A

        Yields:
            None

        Raises:
            N/A

        """
        if self.channel_lock:
            with self.channel_lock:
                yield
        else:
            yield

    def read(self) -> bytes:
        """
        Read chunks of output from the channel

        Replaces any r"\r" characters that sometimes get stuffed into the output from the devices

        Args:
            N/A

        Returns:
            bytes: output read from channel

        Raises:
            N/A

        """
        buf = self.transport.read()
        buf = buf.replace(b"\r", b"")

        self.logger.debug(f"read: {buf!r}")

        if self.channel_log:
            self.channel_log.write(buf)

        if b"\x1b" in buf.lower():
            buf = self._strip_ansi(buf=buf)

        return buf

    def _read_until_input(self, channel_input: bytes) -> bytes:
        """
        Read until all channel_input has been read on the channel

        Args:
            channel_input: bytes that should have been written to the channel

        Returns:
            bytes: output read from channel while checking for the input in the channel stream

        Raises:
            N/A

        """
        buf = b""

        if not channel_input:
            return buf

        # squish all channel input words together and cast to lower to make comparison easier
        processed_channel_input = b"".join(channel_input.lower().split())

        while True:
            buf += self.read()

            # replace any backspace chars (particular problem w/ junos), and remove any added spaces
            # this is just for comparison of the inputs to what was read from channel
            if processed_channel_input in b"".join(buf.lower().replace(b"\x08", b"").split()):
                return buf

    def _read_until_prompt(self, buf: bytes = b"") -> bytes:
        """
        Read until expected prompt is seen.

        This reads until the "normal" `_base_channel_args.comms_prompt_pattern` is seen. The
        `_read_until_explicit_prompt` method can be used to read until some pattern in an arbitrary
        list of patterns is seen.

        Args:
            buf: output from previous reads if needed (used by scrapli netconf)

        Returns:
            bytes: output read from channel

        Raises:
            N/A

        """
        search_pattern = self._get_prompt_pattern(
            class_pattern=self._base_channel_args.comms_prompt_pattern
        )

        read_buf = BytesIO(buf)

        while True:
            read_buf.write(self.read())

            search_buf = self._process_read_buf(read_buf=read_buf)

            channel_match = re.search(
                pattern=search_pattern,
                string=search_buf,
            )

            if channel_match:
                return read_buf.getvalue()

    def _read_until_explicit_prompt(self, prompts: List[str]) -> bytes:
        """
        Read until expected prompt is seen.

        This method is for *explicit* prompt patterns instead of the "standard" prompt patterns
        contained in the `_base_channel_args.comms_prompt_pattern` attribute. Generally this is
        only used for `send_interactive` though it could be used elsewhere as well.

        Args:
            prompts: list of prompt patterns to look for, will return upon seeing any match

        Returns:
            bytes: output read from channel

        Raises:
            N/A

        """
        search_patterns = [
            self._get_prompt_pattern(
                class_pattern=self._base_channel_args.comms_prompt_pattern, pattern=prompt
            )
            for prompt in prompts
        ]

        read_buf = BytesIO(b"")

        while True:
            read_buf.write(self.read())

            search_buf = self._process_read_buf(read_buf=read_buf)

            for search_pattern in search_patterns:
                channel_match = re.search(
                    pattern=search_pattern,
                    string=search_buf,
                )

                if channel_match:
                    return read_buf.getvalue()

    def _read_until_prompt_or_time(
        self,
        buf: bytes = b"",
        channel_outputs: Optional[List[bytes]] = None,
        read_duration: Optional[float] = None,
    ) -> bytes:
        """
        Read until expected prompt is seen, outputs are seen, for duration, whichever comes first.

        As transport reading may block, transport timeout is temporarily set to the read_duration
        and any `ScrapliTimeout` that is raised while reading is ignored.

        Args:
            buf: bytes from previous reads if needed
            channel_outputs: List of bytes to search for in channel output, if any are seen, return
                read output
            read_duration: duration to read from channel for

        Returns:
            bytes: output read from channel

        Raises:
            N/A

        """
        search_pattern = self._get_prompt_pattern(
            class_pattern=self._base_channel_args.comms_prompt_pattern,
        )

        if channel_outputs is None:
            channel_outputs = []
        if read_duration is None:
            read_duration = 2.5

        regex_channel_outputs_pattern = self._join_and_compile(channel_outputs=channel_outputs)

        _transport_args = self.transport._base_transport_args  # pylint: disable=W0212
        previous_timeout_transport = _transport_args.timeout_transport
        _transport_args.timeout_transport = int(read_duration)

        read_buf = BytesIO(buf)

        start = time.time()
        while True:
            with suppress(ScrapliTimeout):
                read_buf.write(self.read())

            search_buf = self._process_read_buf(read_buf=read_buf)

            if (time.time() - start) > read_duration:
                break
            if any((channel_output in search_buf for channel_output in channel_outputs)):
                break
            if re.search(pattern=regex_channel_outputs_pattern, string=search_buf):
                break
            if re.search(pattern=search_pattern, string=search_buf):
                break

        _transport_args.timeout_transport = previous_timeout_transport

        return read_buf.getvalue()

    @timeout_wrapper
    def channel_authenticate_ssh(
        self, auth_password: str, auth_private_key_passphrase: str
    ) -> None:
        """
        Handle SSH Authentication for transports that only operate "in the channel" (i.e. system)

        Args:
            auth_password: password to authenticate with
            auth_private_key_passphrase: passphrase for ssh key if necessary

        Returns:
            None

        Raises:
            ScrapliAuthenticationFailed: if password prompt seen more than twice
            ScrapliAuthenticationFailed: if passphrase prompt seen more than twice

        """
        self.logger.debug("attempting in channel ssh authentication")

        password_count = 0
        passphrase_count = 0
        authenticate_buf = b""

        (
            password_pattern,
            passphrase_pattern,
            prompt_pattern,
        ) = self._pre_channel_authenticate_ssh()

        with self._channel_lock():
            while True:
                buf = self.read()
                authenticate_buf += buf.lower()

                self._ssh_message_handler(output=authenticate_buf)

                if re.search(
                    pattern=password_pattern,
                    string=authenticate_buf,
                ):
                    # clear the authentication buffer so we don't re-read the password prompt
                    authenticate_buf = b""
                    password_count += 1
                    if password_count > 2:
                        msg = "password prompt seen more than once, assuming auth failed"
                        self.logger.critical(msg)
                        raise ScrapliAuthenticationFailed(msg)
                    self.write(channel_input=auth_password, redacted=True)
                    self.send_return()

                if re.search(
                    pattern=passphrase_pattern,
                    string=authenticate_buf,
                ):
                    # clear the authentication buffer so we don't re-read the passphrase prompt
                    authenticate_buf = b""
                    passphrase_count += 1
                    if passphrase_count > 2:
                        msg = "passphrase prompt seen more than once, assuming auth failed"
                        self.logger.critical(msg)
                        raise ScrapliAuthenticationFailed(msg)
                    self.write(channel_input=auth_private_key_passphrase, redacted=True)
                    self.send_return()

                if re.search(
                    pattern=prompt_pattern,
                    string=authenticate_buf,
                ):
                    return

    @timeout_wrapper
    def channel_authenticate_telnet(  # noqa: c901
        self, auth_username: str = "", auth_password: str = ""
    ) -> None:
        """
        Handle Telnet Authentication

        Args:
            auth_username: username to use for telnet authentication
            auth_password: password to use for telnet authentication

        Returns:
            None

        Raises:
            ScrapliAuthenticationFailed: if password prompt seen more than twice
            ScrapliAuthenticationFailed: if login prompt seen more than twice

        """
        self.logger.debug("attempting in channel telnet authentication")

        username_count = 0
        password_count = 0
        authenticate_buf = b""

        (
            username_pattern,
            password_pattern,
            prompt_pattern,
            auth_start_time,
            return_interval,
        ) = self._pre_channel_authenticate_telnet()

        return_attempts = 1

        with self._channel_lock():
            while True:
                try:
                    buf = self.read()
                except ScrapliConnectionError:
                    # telnet transport socket can send us an EOF which gets raised as a connection
                    # error, if we see that we can try to send a return and go back to the top...
                    # this first cropped up with telnet on asa devices in:
                    # https://github.com/carlmontanari/scrapli/issues/278
                    self.send_return()
                    return_attempts += 1
                    continue

                if not buf:
                    current_iteration_time = datetime.now().timestamp()
                    if (current_iteration_time - auth_start_time) > (
                        return_interval * return_attempts
                    ):
                        self.send_return()
                        return_attempts += 1

                authenticate_buf += buf.lower()

                if re.search(
                    pattern=username_pattern,
                    string=authenticate_buf,
                ):
                    # clear the authentication buffer so we don't re-read the username prompt
                    authenticate_buf = b""
                    username_count += 1
                    if username_count > 2:
                        msg = "username/login prompt seen more than once, assuming auth failed"
                        self.logger.critical(msg)
                        raise ScrapliAuthenticationFailed(msg)
                    self.write(channel_input=auth_username)
                    self.send_return()

                if re.search(
                    pattern=password_pattern,
                    string=authenticate_buf,
                ):
                    # clear the authentication buffer so we don't re-read the password prompt
                    authenticate_buf = b""
                    password_count += 1
                    if password_count > 2:
                        msg = "password prompt seen more than once, assuming auth failed"
                        self.logger.critical(msg)
                        raise ScrapliAuthenticationFailed(msg)
                    self.write(channel_input=auth_password, redacted=True)
                    self.send_return()

                if re.search(
                    pattern=prompt_pattern,
                    string=authenticate_buf,
                ):
                    return

    @timeout_wrapper
    def get_prompt(self) -> str:
        """
        Get current channel prompt

        Args:
            N/A

        Returns:
            str: string of the current prompt

        Raises:
            N/A

        """
        buf = b""

        search_pattern = self._get_prompt_pattern(
            class_pattern=self._base_channel_args.comms_prompt_pattern
        )

        with self._channel_lock():
            self.send_return()

            while True:
                buf += self.read()

                channel_match = re.search(
                    pattern=search_pattern,
                    string=buf,
                )

                if channel_match:
                    current_prompt = channel_match.group(0)
                    return current_prompt.decode().strip()

    @timeout_wrapper
    def send_input(
        self,
        channel_input: str,
        *,
        strip_prompt: bool = True,
        eager: bool = False,
    ) -> Tuple[bytes, bytes]:
        """
        Primary entry point to send data to devices in shell mode; accept input and returns result

        Args:
            channel_input: string input to send to channel
            strip_prompt: strip prompt or not, defaults to True (yes, strip the prompt)
            eager: eager mode reads and returns the `_read_until_input` value, but does not attempt
                to read to the prompt pattern -- this should not be used manually! (only used by
                `send_configs` with the eager flag set)

        Returns:
            Tuple[bytes, bytes]: tuple of "raw" output and "processed" (cleaned up/stripped) output

        Raises:
            N/A

        """
        self._pre_send_input(channel_input=channel_input)

        buf = b""
        bytes_channel_input = channel_input.encode()

        self.logger.info(
            f"sending channel input: {channel_input}; strip_prompt: {strip_prompt}; eager: {eager}"
        )

        with self._channel_lock():
            self.write(channel_input=channel_input)
            _buf_until_input = self._read_until_input(channel_input=bytes_channel_input)
            self.send_return()

            if not eager:
                buf += self._read_until_prompt()

        processed_buf = self._process_output(
            buf=buf,
            strip_prompt=strip_prompt,
        )
        return buf, processed_buf

    @timeout_wrapper
    def send_input_and_read(
        self,
        channel_input: str,
        *,
        strip_prompt: bool = True,
        expected_outputs: Optional[List[str]] = None,
        read_duration: Optional[float] = None,
    ) -> Tuple[bytes, bytes]:
        """
        Send a command and read until expected prompt is seen, outputs are seen, or for duration

        Args:
            channel_input: string input to send to channel
            strip_prompt: strip prompt or not, defaults to True (yes, strip the prompt)
            expected_outputs: list of strings to look for in output; if any of these are seen,
                return output read up till that read
            read_duration: float duration to read for

        Returns:
            Tuple[bytes, bytes]: tuple of "raw" output and "processed" (cleaned up/stripped) output

        Raises:
            N/A

        """
        self._pre_send_input(channel_input=channel_input)

        buf = b""
        bytes_channel_input = channel_input.encode()
        bytes_channel_outputs = [
            channel_output.encode() for channel_output in expected_outputs or []
        ]

        self.logger.info(
            f"sending channel input and read: {channel_input}; strip_prompt: {strip_prompt}; "
            f"expected_outputs: {expected_outputs}; read_duration: {read_duration}"
        )

        with self._channel_lock():
            self.write(channel_input=channel_input)
            _buf_until_input = self._read_until_input(channel_input=bytes_channel_input)
            self.send_return()

            buf += self._read_until_prompt_or_time(
                channel_outputs=bytes_channel_outputs, read_duration=read_duration
            )

        processed_buf = self._process_output(
            buf=buf,
            strip_prompt=strip_prompt,
        )

        return buf, processed_buf

    @timeout_wrapper
    def send_inputs_interact(
        self,
        interact_events: List[Tuple[str, str, Optional[bool]]],
        *,
        interaction_complete_patterns: Optional[List[str]] = None,
    ) -> Tuple[bytes, bytes]:
        """
        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)
            interaction_complete_patterns: list of patterns, that if seen, indicate the interactive
                "session" has ended and we should exit the interactive session.

        Returns:
            Tuple[bytes, bytes]: output read from the channel with no whitespace trimming/cleaning,
                and the output read from the channel that has been "cleaned up"

        Raises:
            N/A

        """
        self._pre_send_inputs_interact(interact_events=interact_events)

        buf = b""
        processed_buf = b""

        with self._channel_lock():
            for interact_event in interact_events:
                channel_input = interact_event[0]
                bytes_channel_input = channel_input.encode()
                channel_response = interact_event[1]
                prompts = [channel_response]

                if interaction_complete_patterns is not None:
                    prompts.extend(interaction_complete_patterns)

                try:
                    hidden_input = interact_event[2]
                except IndexError:
                    hidden_input = False

                _channel_input = channel_input if not hidden_input else "REDACTED"
                self.logger.info(
                    f"sending interactive input: {_channel_input}; "
                    f"expecting: {channel_response}; "
                    f"hidden_input: {hidden_input}"
                )

                self.write(channel_input=channel_input, redacted=bool(hidden_input))
                if channel_response and hidden_input is not True:
                    buf += self._read_until_input(channel_input=bytes_channel_input)
                self.send_return()
                buf += self._read_until_explicit_prompt(prompts=prompts)

        processed_buf += self._process_output(
            buf=buf,
            strip_prompt=False,
        )

        return buf, processed_buf

channel_authenticate_ssh(auth_password: str, auth_private_key_passphrase: str) -> None

Handle SSH Authentication for transports that only operate "in the channel" (i.e. system)

Parameters:

Name Type Description Default
auth_password str

password to authenticate with

required
auth_private_key_passphrase str

passphrase for ssh key if necessary

required

Returns:

Type Description
None

None

Raises:

Type Description
ScrapliAuthenticationFailed

if password prompt seen more than twice

ScrapliAuthenticationFailed

if passphrase prompt seen more than twice

Source code in channel/sync_channel.py
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
@timeout_wrapper
def channel_authenticate_ssh(
    self, auth_password: str, auth_private_key_passphrase: str
) -> None:
    """
    Handle SSH Authentication for transports that only operate "in the channel" (i.e. system)

    Args:
        auth_password: password to authenticate with
        auth_private_key_passphrase: passphrase for ssh key if necessary

    Returns:
        None

    Raises:
        ScrapliAuthenticationFailed: if password prompt seen more than twice
        ScrapliAuthenticationFailed: if passphrase prompt seen more than twice

    """
    self.logger.debug("attempting in channel ssh authentication")

    password_count = 0
    passphrase_count = 0
    authenticate_buf = b""

    (
        password_pattern,
        passphrase_pattern,
        prompt_pattern,
    ) = self._pre_channel_authenticate_ssh()

    with self._channel_lock():
        while True:
            buf = self.read()
            authenticate_buf += buf.lower()

            self._ssh_message_handler(output=authenticate_buf)

            if re.search(
                pattern=password_pattern,
                string=authenticate_buf,
            ):
                # clear the authentication buffer so we don't re-read the password prompt
                authenticate_buf = b""
                password_count += 1
                if password_count > 2:
                    msg = "password prompt seen more than once, assuming auth failed"
                    self.logger.critical(msg)
                    raise ScrapliAuthenticationFailed(msg)
                self.write(channel_input=auth_password, redacted=True)
                self.send_return()

            if re.search(
                pattern=passphrase_pattern,
                string=authenticate_buf,
            ):
                # clear the authentication buffer so we don't re-read the passphrase prompt
                authenticate_buf = b""
                passphrase_count += 1
                if passphrase_count > 2:
                    msg = "passphrase prompt seen more than once, assuming auth failed"
                    self.logger.critical(msg)
                    raise ScrapliAuthenticationFailed(msg)
                self.write(channel_input=auth_private_key_passphrase, redacted=True)
                self.send_return()

            if re.search(
                pattern=prompt_pattern,
                string=authenticate_buf,
            ):
                return

channel_authenticate_telnet(auth_username: str = '', auth_password: str = '') -> None

Handle Telnet Authentication

Parameters:

Name Type Description Default
auth_username str

username to use for telnet authentication

''
auth_password str

password to use for telnet authentication

''

Returns:

Type Description
None

None

Raises:

Type Description
ScrapliAuthenticationFailed

if password prompt seen more than twice

ScrapliAuthenticationFailed

if login prompt seen more than twice

Source code in channel/sync_channel.py
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
@timeout_wrapper
def channel_authenticate_telnet(  # noqa: c901
    self, auth_username: str = "", auth_password: str = ""
) -> None:
    """
    Handle Telnet Authentication

    Args:
        auth_username: username to use for telnet authentication
        auth_password: password to use for telnet authentication

    Returns:
        None

    Raises:
        ScrapliAuthenticationFailed: if password prompt seen more than twice
        ScrapliAuthenticationFailed: if login prompt seen more than twice

    """
    self.logger.debug("attempting in channel telnet authentication")

    username_count = 0
    password_count = 0
    authenticate_buf = b""

    (
        username_pattern,
        password_pattern,
        prompt_pattern,
        auth_start_time,
        return_interval,
    ) = self._pre_channel_authenticate_telnet()

    return_attempts = 1

    with self._channel_lock():
        while True:
            try:
                buf = self.read()
            except ScrapliConnectionError:
                # telnet transport socket can send us an EOF which gets raised as a connection
                # error, if we see that we can try to send a return and go back to the top...
                # this first cropped up with telnet on asa devices in:
                # https://github.com/carlmontanari/scrapli/issues/278
                self.send_return()
                return_attempts += 1
                continue

            if not buf:
                current_iteration_time = datetime.now().timestamp()
                if (current_iteration_time - auth_start_time) > (
                    return_interval * return_attempts
                ):
                    self.send_return()
                    return_attempts += 1

            authenticate_buf += buf.lower()

            if re.search(
                pattern=username_pattern,
                string=authenticate_buf,
            ):
                # clear the authentication buffer so we don't re-read the username prompt
                authenticate_buf = b""
                username_count += 1
                if username_count > 2:
                    msg = "username/login prompt seen more than once, assuming auth failed"
                    self.logger.critical(msg)
                    raise ScrapliAuthenticationFailed(msg)
                self.write(channel_input=auth_username)
                self.send_return()

            if re.search(
                pattern=password_pattern,
                string=authenticate_buf,
            ):
                # clear the authentication buffer so we don't re-read the password prompt
                authenticate_buf = b""
                password_count += 1
                if password_count > 2:
                    msg = "password prompt seen more than once, assuming auth failed"
                    self.logger.critical(msg)
                    raise ScrapliAuthenticationFailed(msg)
                self.write(channel_input=auth_password, redacted=True)
                self.send_return()

            if re.search(
                pattern=prompt_pattern,
                string=authenticate_buf,
            ):
                return

get_prompt() -> str

Get current channel prompt

Returns:

Name Type Description
str str

string of the current prompt

Source code in channel/sync_channel.py
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
@timeout_wrapper
def get_prompt(self) -> str:
    """
    Get current channel prompt

    Args:
        N/A

    Returns:
        str: string of the current prompt

    Raises:
        N/A

    """
    buf = b""

    search_pattern = self._get_prompt_pattern(
        class_pattern=self._base_channel_args.comms_prompt_pattern
    )

    with self._channel_lock():
        self.send_return()

        while True:
            buf += self.read()

            channel_match = re.search(
                pattern=search_pattern,
                string=buf,
            )

            if channel_match:
                current_prompt = channel_match.group(0)
                return current_prompt.decode().strip()

read() -> bytes

Read chunks of output from the channel

Replaces any r" " characters that sometimes get stuffed into the output from the devices

Returns:

Name Type Description
bytes bytes

output read from channel

Source code in channel/sync_channel.py
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
def read(self) -> bytes:
    """
    Read chunks of output from the channel

    Replaces any r"\r" characters that sometimes get stuffed into the output from the devices

    Args:
        N/A

    Returns:
        bytes: output read from channel

    Raises:
        N/A

    """
    buf = self.transport.read()
    buf = buf.replace(b"\r", b"")

    self.logger.debug(f"read: {buf!r}")

    if self.channel_log:
        self.channel_log.write(buf)

    if b"\x1b" in buf.lower():
        buf = self._strip_ansi(buf=buf)

    return buf

send_input(channel_input: str, *, strip_prompt: bool = True, eager: bool = False) -> Tuple[bytes, bytes]

Primary entry point to send data to devices in shell mode; accept input and returns result

Parameters:

Name Type Description Default
channel_input str

string input to send to channel

required
strip_prompt bool

strip prompt or not, defaults to True (yes, strip the prompt)

True
eager bool

eager mode reads and returns the _read_until_input value, but does not attempt to read to the prompt pattern -- this should not be used manually! (only used by send_configs with the eager flag set)

False

Returns:

Type Description
Tuple[bytes, bytes]

Tuple[bytes, bytes]: tuple of "raw" output and "processed" (cleaned up/stripped) output

Source code in channel/sync_channel.py
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
@timeout_wrapper
def send_input(
    self,
    channel_input: str,
    *,
    strip_prompt: bool = True,
    eager: bool = False,
) -> Tuple[bytes, bytes]:
    """
    Primary entry point to send data to devices in shell mode; accept input and returns result

    Args:
        channel_input: string input to send to channel
        strip_prompt: strip prompt or not, defaults to True (yes, strip the prompt)
        eager: eager mode reads and returns the `_read_until_input` value, but does not attempt
            to read to the prompt pattern -- this should not be used manually! (only used by
            `send_configs` with the eager flag set)

    Returns:
        Tuple[bytes, bytes]: tuple of "raw" output and "processed" (cleaned up/stripped) output

    Raises:
        N/A

    """
    self._pre_send_input(channel_input=channel_input)

    buf = b""
    bytes_channel_input = channel_input.encode()

    self.logger.info(
        f"sending channel input: {channel_input}; strip_prompt: {strip_prompt}; eager: {eager}"
    )

    with self._channel_lock():
        self.write(channel_input=channel_input)
        _buf_until_input = self._read_until_input(channel_input=bytes_channel_input)
        self.send_return()

        if not eager:
            buf += self._read_until_prompt()

    processed_buf = self._process_output(
        buf=buf,
        strip_prompt=strip_prompt,
    )
    return buf, processed_buf

send_input_and_read(channel_input: str, *, strip_prompt: bool = True, expected_outputs: Optional[List[str]] = None, read_duration: Optional[float] = None) -> Tuple[bytes, bytes]

Send a command and read until expected prompt is seen, outputs are seen, or for duration

Parameters:

Name Type Description Default
channel_input str

string input to send to channel

required
strip_prompt bool

strip prompt or not, defaults to True (yes, strip the prompt)

True
expected_outputs Optional[List[str]]

list of strings to look for in output; if any of these are seen, return output read up till that read

None
read_duration Optional[float]

float duration to read for

None

Returns:

Type Description
Tuple[bytes, bytes]

Tuple[bytes, bytes]: tuple of "raw" output and "processed" (cleaned up/stripped) output

Source code in channel/sync_channel.py
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
@timeout_wrapper
def send_input_and_read(
    self,
    channel_input: str,
    *,
    strip_prompt: bool = True,
    expected_outputs: Optional[List[str]] = None,
    read_duration: Optional[float] = None,
) -> Tuple[bytes, bytes]:
    """
    Send a command and read until expected prompt is seen, outputs are seen, or for duration

    Args:
        channel_input: string input to send to channel
        strip_prompt: strip prompt or not, defaults to True (yes, strip the prompt)
        expected_outputs: list of strings to look for in output; if any of these are seen,
            return output read up till that read
        read_duration: float duration to read for

    Returns:
        Tuple[bytes, bytes]: tuple of "raw" output and "processed" (cleaned up/stripped) output

    Raises:
        N/A

    """
    self._pre_send_input(channel_input=channel_input)

    buf = b""
    bytes_channel_input = channel_input.encode()
    bytes_channel_outputs = [
        channel_output.encode() for channel_output in expected_outputs or []
    ]

    self.logger.info(
        f"sending channel input and read: {channel_input}; strip_prompt: {strip_prompt}; "
        f"expected_outputs: {expected_outputs}; read_duration: {read_duration}"
    )

    with self._channel_lock():
        self.write(channel_input=channel_input)
        _buf_until_input = self._read_until_input(channel_input=bytes_channel_input)
        self.send_return()

        buf += self._read_until_prompt_or_time(
            channel_outputs=bytes_channel_outputs, read_duration=read_duration
        )

    processed_buf = self._process_output(
        buf=buf,
        strip_prompt=strip_prompt,
    )

    return buf, processed_buf

send_inputs_interact(interact_events: List[Tuple[str, str, Optional[bool]]], *, interaction_complete_patterns: Optional[List[str]] = None) -> Tuple[bytes, bytes]

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 List[Tuple[str, str, Optional[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
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:

Type Description
Tuple[bytes, bytes]

Tuple[bytes, bytes]: output read from the channel with no whitespace trimming/cleaning, and the output read from the channel that has been "cleaned up"

Source code in channel/sync_channel.py
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
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
@timeout_wrapper
def send_inputs_interact(
    self,
    interact_events: List[Tuple[str, str, Optional[bool]]],
    *,
    interaction_complete_patterns: Optional[List[str]] = None,
) -> Tuple[bytes, bytes]:
    """
    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)
        interaction_complete_patterns: list of patterns, that if seen, indicate the interactive
            "session" has ended and we should exit the interactive session.

    Returns:
        Tuple[bytes, bytes]: output read from the channel with no whitespace trimming/cleaning,
            and the output read from the channel that has been "cleaned up"

    Raises:
        N/A

    """
    self._pre_send_inputs_interact(interact_events=interact_events)

    buf = b""
    processed_buf = b""

    with self._channel_lock():
        for interact_event in interact_events:
            channel_input = interact_event[0]
            bytes_channel_input = channel_input.encode()
            channel_response = interact_event[1]
            prompts = [channel_response]

            if interaction_complete_patterns is not None:
                prompts.extend(interaction_complete_patterns)

            try:
                hidden_input = interact_event[2]
            except IndexError:
                hidden_input = False

            _channel_input = channel_input if not hidden_input else "REDACTED"
            self.logger.info(
                f"sending interactive input: {_channel_input}; "
                f"expecting: {channel_response}; "
                f"hidden_input: {hidden_input}"
            )

            self.write(channel_input=channel_input, redacted=bool(hidden_input))
            if channel_response and hidden_input is not True:
                buf += self._read_until_input(channel_input=bytes_channel_input)
            self.send_return()
            buf += self._read_until_explicit_prompt(prompts=prompts)

    processed_buf += self._process_output(
        buf=buf,
        strip_prompt=False,
    )

    return buf, processed_buf