0
# SSH Automation
1
2
Specialized SSH connection wrapper that extends spawn with SSH-specific functionality. The pxssh module provides high-level SSH automation with built-in login, logout, and prompt management.
3
4
## Capabilities
5
6
### SSH Connection Class
7
8
Enhanced spawn class specifically designed for SSH automation with built-in SSH protocol handling.
9
10
```python { .api }
11
class pxssh(spawn):
12
"""
13
Specialized SSH connection class that extends spawn with SSH-specific methods.
14
Handles SSH login, logout, prompt detection, and common SSH scenarios.
15
"""
16
17
def __init__(self, timeout=30, maxread=2000, searchwindowsize=None,
18
logfile=None, cwd=None, env=None, ignore_sighup=False,
19
echo=True, options={}, spawn_local_ssh=True, sync_multiplier=1,
20
check_local_ip=True):
21
"""
22
Initialize SSH connection object.
23
24
Parameters:
25
- timeout (int): Default timeout for operations
26
- maxread (int): Maximum bytes to read at once
27
- searchwindowsize (int): Search window size for pattern matching
28
- logfile (file-like): File object for logging
29
- cwd (str): Working directory (inherited from spawn)
30
- env (dict): Environment variables (inherited from spawn)
31
- ignore_sighup (bool): Ignore SIGHUP signal
32
- echo (bool): Terminal echo setting
33
- options (dict): SSH client options
34
- spawn_local_ssh (bool): Use local SSH client
35
- sync_multiplier (int): Synchronization timing multiplier
36
- check_local_ip (bool): Verify local IP for security
37
"""
38
```
39
40
### SSH Connection Management
41
42
```python { .api }
43
def login(self, server, username, password='', terminal_type='ansi',
44
original_prompt=r"[#$]", login_timeout=10, port=None,
45
auto_prompt_reset=True, ssh_key=None, quiet=True,
46
sync_multiplier=1, check_local_ip=True, password_regex=None,
47
ssh_tunnels=None, spawn_local_ssh=True, sync_original_prompt=True,
48
ssh_config=None, cmd="ssh"):
49
"""
50
Establish SSH connection and login.
51
52
Parameters:
53
- server (str): Hostname or IP address to connect to
54
- username (str): Username for SSH login
55
- password (str): Password for authentication (if not using key)
56
- terminal_type (str): Terminal type to request
57
- original_prompt (str): Regex pattern for shell prompt
58
- login_timeout (int): Timeout for login process
59
- port (int): SSH port number (default: 22)
60
- auto_prompt_reset (bool): Automatically set unique prompt
61
- ssh_key (str): Path to SSH private key file
62
- quiet (bool): Suppress SSH client messages
63
- sync_multiplier (int): Timing adjustment for slow connections
64
- check_local_ip (bool): Verify local IP address
65
- password_regex (str): Custom regex for password prompt
66
- ssh_tunnels (dict): SSH tunnel configurations
67
- spawn_local_ssh (bool): Use local SSH client
68
- sync_original_prompt (bool): Synchronize with original prompt
69
- ssh_config (str): Path to SSH config file
70
- cmd (str): SSH command to execute (default: "ssh")
71
72
Returns:
73
bool: True if login successful, False otherwise
74
75
Raises:
76
- ExceptionPxssh: If login fails or connection issues occur
77
"""
78
79
def logout(self):
80
"""
81
Logout from SSH session and close connection.
82
83
Sends 'exit' command and closes the connection cleanly.
84
"""
85
86
def prompt(self, timeout=-1):
87
"""
88
Wait for shell prompt to appear.
89
90
Parameters:
91
- timeout (int): Timeout in seconds (-1 for default)
92
93
Returns:
94
bool: True if prompt found, False on timeout
95
96
Raises:
97
- TIMEOUT: If timeout exceeded
98
- EOF: If connection closed
99
"""
100
```
101
102
### Prompt Management
103
104
```python { .api }
105
def set_unique_prompt(self):
106
"""
107
Set a unique shell prompt for reliable pattern matching.
108
109
Changes the shell prompt to a unique string that's unlikely to
110
appear in command output, making expect operations more reliable.
111
112
Returns:
113
bool: True if prompt was set successfully
114
"""
115
116
def sync_original_prompt(self, sync_multiplier=1):
117
"""
118
Synchronize with the original shell prompt.
119
120
Parameters:
121
- sync_multiplier (int): Timing adjustment multiplier
122
123
Returns:
124
bool: True if synchronized successfully
125
"""
126
```
127
128
### SSH Exception Class
129
130
```python { .api }
131
class ExceptionPxssh(ExceptionPexpect):
132
"""
133
Exception class for pxssh-specific errors.
134
135
Raised for SSH connection failures, authentication errors,
136
and other SSH-specific problems.
137
"""
138
```
139
140
## SSH Configuration and Options
141
142
### SSH Client Options
143
144
```python
145
# Common SSH options that can be passed to pxssh
146
ssh_options = {
147
'StrictHostKeyChecking': 'no', # Skip host key verification
148
'UserKnownHostsFile': '/dev/null', # Don't save host keys
149
'ConnectTimeout': '10', # Connection timeout
150
'ServerAliveInterval': '60', # Keep-alive interval
151
'ServerAliveCountMax': '3', # Keep-alive retry count
152
'PasswordAuthentication': 'yes', # Allow password auth
153
'PubkeyAuthentication': 'yes', # Allow key auth
154
}
155
156
ssh = pxssh.pxssh(options=ssh_options)
157
```
158
159
### Authentication Methods
160
161
```python
162
# Password authentication
163
ssh = pxssh.pxssh()
164
ssh.login('server.example.com', 'username', 'password')
165
166
# SSH key authentication
167
ssh = pxssh.pxssh()
168
ssh.login('server.example.com', 'username', ssh_key='/path/to/private_key')
169
170
# Password authentication with custom prompt
171
ssh = pxssh.pxssh()
172
ssh.login('server.example.com', 'username', 'password',
173
original_prompt=r'[#$%>] ')
174
```
175
176
## Usage Examples
177
178
### Basic SSH Connection
179
180
```python
181
from pexpect import pxssh
182
import getpass
183
184
# Create SSH connection
185
ssh = pxssh.pxssh()
186
187
try:
188
# Login with password
189
hostname = 'server.example.com'
190
username = 'myuser'
191
password = getpass.getpass('Password: ')
192
193
ssh.login(hostname, username, password)
194
195
# Execute commands
196
ssh.sendline('ls -la')
197
ssh.prompt()
198
print(ssh.before.decode())
199
200
ssh.sendline('uptime')
201
ssh.prompt()
202
print(ssh.before.decode())
203
204
# Logout cleanly
205
ssh.logout()
206
207
except pxssh.ExceptionPxssh as e:
208
print(f"SSH connection failed: {e}")
209
```
210
211
### SSH with Key Authentication
212
213
```python
214
from pexpect import pxssh
215
216
ssh = pxssh.pxssh()
217
218
try:
219
# Login with SSH key
220
ssh.login('server.example.com', 'username',
221
ssh_key='/home/user/.ssh/id_rsa')
222
223
# Run multiple commands
224
commands = ['whoami', 'pwd', 'date', 'df -h']
225
226
for cmd in commands:
227
ssh.sendline(cmd)
228
ssh.prompt()
229
output = ssh.before.decode().strip()
230
print(f"{cmd}: {output}")
231
232
ssh.logout()
233
234
except pxssh.ExceptionPxssh as e:
235
print(f"SSH error: {e}")
236
```
237
238
### Handling Connection Issues
239
240
```python
241
from pexpect import pxssh
242
import time
243
244
def robust_ssh_connect(hostname, username, password, max_retries=3):
245
"""Establish SSH connection with retry logic."""
246
247
for attempt in range(max_retries):
248
ssh = pxssh.pxssh()
249
250
try:
251
ssh.login(hostname, username, password, login_timeout=30)
252
print(f"Connected to {hostname} on attempt {attempt + 1}")
253
return ssh
254
255
except pxssh.ExceptionPxssh as e:
256
print(f"Attempt {attempt + 1} failed: {e}")
257
if attempt < max_retries - 1:
258
time.sleep(5) # Wait before retry
259
else:
260
raise
261
262
return None
263
264
# Use robust connection
265
try:
266
ssh = robust_ssh_connect('server.example.com', 'user', 'pass')
267
268
# Your SSH operations here
269
ssh.sendline('hostname')
270
ssh.prompt()
271
print(f"Connected to: {ssh.before.decode().strip()}")
272
273
ssh.logout()
274
275
except Exception as e:
276
print(f"Failed to establish SSH connection: {e}")
277
```
278
279
### SSH Tunneling and Port Forwarding
280
281
```python
282
from pexpect import pxssh
283
284
# SSH with port forwarding
285
ssh_tunnels = {
286
'local': [
287
{'local_port': 8080, 'remote_host': 'localhost', 'remote_port': 80}
288
]
289
}
290
291
ssh = pxssh.pxssh()
292
293
try:
294
ssh.login('jumphost.example.com', 'username', 'password',
295
ssh_tunnels=ssh_tunnels)
296
297
# Now port 8080 on local machine forwards to port 80 on remote
298
print("SSH tunnel established")
299
300
# Keep connection alive while using tunnel
301
ssh.sendline('echo "Tunnel active"')
302
ssh.prompt()
303
304
# Your application can now use localhost:8080
305
# to access the remote service
306
307
finally:
308
ssh.logout()
309
```
310
311
### Custom SSH Configuration
312
313
```python
314
from pexpect import pxssh
315
316
# Custom SSH client configuration
317
ssh = pxssh.pxssh(
318
options={
319
'StrictHostKeyChecking': 'no',
320
'UserKnownHostsFile': '/dev/null',
321
'ConnectTimeout': '30',
322
'ServerAliveInterval': '120'
323
},
324
timeout=60,
325
sync_multiplier=2 # For slow connections
326
)
327
328
try:
329
# Login with custom configuration
330
ssh.login('slow-server.example.com', 'username', 'password',
331
login_timeout=60,
332
terminal_type='xterm-256color')
333
334
# Set custom prompt for better reliability
335
ssh.set_unique_prompt()
336
337
# Execute commands with custom prompt
338
ssh.sendline('export TERM=xterm-256color')
339
ssh.prompt()
340
341
ssh.sendline('ls --color=auto')
342
ssh.prompt()
343
print(ssh.before.decode())
344
345
ssh.logout()
346
347
except pxssh.ExceptionPxssh as e:
348
print(f"SSH error: {e}")
349
```
350
351
### SSH Session Management
352
353
```python
354
from pexpect import pxssh
355
import contextlib
356
357
@contextlib.contextmanager
358
def ssh_session(hostname, username, password):
359
"""Context manager for SSH sessions."""
360
ssh = pxssh.pxssh()
361
try:
362
ssh.login(hostname, username, password)
363
yield ssh
364
except pxssh.ExceptionPxssh as e:
365
print(f"SSH error: {e}")
366
raise
367
finally:
368
try:
369
ssh.logout()
370
except:
371
pass # Ignore logout errors
372
373
# Use context manager
374
with ssh_session('server.example.com', 'user', 'pass') as ssh:
375
ssh.sendline('whoami')
376
ssh.prompt()
377
user = ssh.before.decode().strip()
378
print(f"Logged in as: {user}")
379
380
ssh.sendline('uname -a')
381
ssh.prompt()
382
system_info = ssh.before.decode().strip()
383
print(f"System: {system_info}")
384
# SSH connection automatically closed
385
```