Rotate an SSH password host(str): SSH host port(int): SSH port user(str): SSH login name old_password(str): old password new_password(str): new password timeout(int): SSH connection timeout in seconds revert(bool): True if the new_password is the original password to rev
(host, port, user, old_password, new_password, timeout=5, revert=False)
| 50 | |
| 51 | |
| 52 | def rotate_ssh(host, port, user, old_password, new_password, timeout=5, revert=False): |
| 53 | """Rotate an SSH password |
| 54 | |
| 55 | host(str): SSH host |
| 56 | port(int): SSH port |
| 57 | user(str): SSH login name |
| 58 | old_password(str): old password |
| 59 | new_password(str): new password |
| 60 | timeout(int): SSH connection timeout in seconds |
| 61 | revert(bool): True if the new_password is the original password to revert a previous rotation. |
| 62 | This is used to print log messages that make more sense. |
| 63 | """ |
| 64 | rotate_success = False |
| 65 | ssh_logger = logging.getLogger('paramiko') |
| 66 | ssh_logger.setLevel(logging.WARNING) |
| 67 | with paramiko.SSHClient() as ssh: |
| 68 | ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) |
| 69 | try: |
| 70 | ssh.connect( |
| 71 | hostname=host, port=port, username=user, password=old_password, |
| 72 | timeout=timeout, allow_agent=False, look_for_keys=False |
| 73 | ) |
| 74 | except paramiko.ssh_exception.AuthenticationException: |
| 75 | if revert: |
| 76 | logging.error('SSH authentication was unsuccessful for revert of rotation.') |
| 77 | else: |
| 78 | logging.error('SSH authentication was unsuccessful using current password.') |
| 79 | return False |
| 80 | except socket.timeout: |
| 81 | logging.error('Connection to host timed out.') |
| 82 | return False |
| 83 | except Exception as e: |
| 84 | logging.error(f'Unrecognized connection error: {e}') |
| 85 | return False |
| 86 | stdin, stdout, stderr = ssh.exec_command('ver') |
| 87 | if ''.join(stdout.readlines()).strip().startswith('Microsoft Windows'): |
| 88 | try: |
| 89 | stdin, stdout, stderr = ssh.exec_command( |
| 90 | f'net user {user} {new_password}' |
| 91 | ) |
| 92 | result = ''.join(stdout.readlines()).strip() |
| 93 | if result == 'The command completed successfully.': |
| 94 | rotate_success = True |
| 95 | logging.debug(result) |
| 96 | else: |
| 97 | logging.warning(f'Unrecognized result: "{result}"') |
| 98 | except Exception as e: |
| 99 | # Catch exception because password |
| 100 | # could have still been rotated and we need to verify |
| 101 | logging.error(str(e)) |
| 102 | else: |
| 103 | stdin, stdout, stderr = ssh.exec_command('which passwd') |
| 104 | passwd_cmd = ''.join(stdout.readlines()).strip() |
| 105 | if not passwd_cmd.endswith('passwd'): |
| 106 | logging.warning('"passwd" command not found on device') |
| 107 | return False |
| 108 | else: |
| 109 | with SSHClientInteraction(ssh, timeout=timeout, display=False) as ia: |