1
2
3
4
5
6
7
8
9
10
11
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
|
#
# (c) 2016 Red Hat Inc.
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
#
from __future__ import annotations
import re
from abc import ABC, abstractmethod
from ansible.errors import AnsibleConnectionFailure
class TerminalBase(ABC):
"""
A base class for implementing cli connections
.. note:: Unlike most of Ansible, nearly all strings in
:class:`TerminalBase` plugins are byte strings. This is because of
how close to the underlying platform these plugins operate. Remember
to mark literal strings as byte string (``b"string"``) and to use
:func:`~ansible.module_utils.common.text.converters.to_bytes` and
:func:`~ansible.module_utils.common.text.converters.to_text` to avoid unexpected
problems.
"""
#: compiled bytes regular expressions as stdout
terminal_stdout_re = [] # type: list[re.Pattern]
#: compiled bytes regular expressions as stderr
terminal_stderr_re = [] # type: list[re.Pattern]
#: compiled bytes regular expressions to remove ANSI codes
ansi_re = [
re.compile(br'\x1b\[\?1h\x1b='), # CSI ? 1 h ESC =
re.compile(br'\x08.'), # [Backspace] .
re.compile(br"\x1b\[m"), # ANSI reset code
]
#: terminal initial prompt
terminal_initial_prompt = None
#: terminal initial answer
terminal_initial_answer = None
#: Send newline after prompt match
terminal_inital_prompt_newline = True
def __init__(self, connection):
self._connection = connection
def _exec_cli_command(self, cmd, check_rc=True):
"""
Executes the CLI command on the remote device and returns the output
:arg cmd: Byte string command to be executed
"""
return self._connection.exec_command(cmd)
def _get_prompt(self):
"""
Returns the current prompt from the device
:returns: A byte string of the prompt
"""
return self._connection.get_prompt()
def on_open_shell(self):
"""Called after the SSH session is established
This method is called right after the invoke_shell() is called from
the Paramiko SSHClient instance. It provides an opportunity to setup
terminal parameters such as disabling paging for instance.
"""
pass
def on_close_shell(self):
"""Called before the connection is closed
This method gets called once the connection close has been requested
but before the connection is actually closed. It provides an
opportunity to clean up any terminal resources before the shell is
actually closed
"""
pass
def on_become(self, passwd=None):
"""Called when privilege escalation is requested
:kwarg passwd: String containing the password
This method is called when the privilege is requested to be elevated
in the play context by setting become to True. It is the responsibility
of the terminal plugin to actually do the privilege escalation such
as entering `enable` mode for instance
"""
pass
def on_unbecome(self):
"""Called when privilege deescalation is requested
This method is called when the privilege changed from escalated
(become=True) to non escalated (become=False). It is the responsibility
of this method to actually perform the deauthorization procedure
"""
pass
def on_authorize(self, passwd=None):
"""Deprecated method for privilege escalation
:kwarg passwd: String containing the password
"""
return self.on_become(passwd)
def on_deauthorize(self):
"""Deprecated method for privilege deescalation
"""
return self.on_unbecome()
|
