130 lines
3.0 KiB
Python
130 lines
3.0 KiB
Python
import paramiko
|
|
from dataclasses import dataclass
|
|
import json
|
|
from paramiko.ssh_exception import AuthenticationException
|
|
from paramiko import SSHClient
|
|
from scp import SCPClient
|
|
from os import path
|
|
|
|
|
|
|
|
class CONFIG:
|
|
scp_dest="/tmp"
|
|
files_folder = 'C:/Users/OriolFilterAnson/PycharmProjects/sshtest'
|
|
files_to_copy = ['t1']
|
|
|
|
@dataclass
|
|
class USER:
|
|
name: str
|
|
password: str
|
|
|
|
# def __init__(self, name, password):
|
|
# self.name = name
|
|
# self.password = password
|
|
|
|
|
|
@dataclass
|
|
class VALIDATIONS:
|
|
loged_in: bool = False
|
|
|
|
def __post_init__(self):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class ITEM:
|
|
hostname: str
|
|
client: paramiko.SSHClient
|
|
# login: USER
|
|
validations: VALIDATIONS = None
|
|
|
|
# def __init__(self, hostname, login):
|
|
# self.hostname = hostname
|
|
# self.login = login
|
|
def __post_init__(self):
|
|
self.validations = VALIDATIONS()
|
|
|
|
|
|
def attempt_to_login(hostname, user: USER) -> SSHClient or bool:
|
|
try:
|
|
client = paramiko.SSHClient()
|
|
client.load_system_host_keys()
|
|
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
client.connect(hostname, username=user.name, password=user.password)
|
|
# ssh_stdin, ssh_stdout, ssh_stderr = client.exec_command("uptime")
|
|
# print(ssh_stdout.read().decode())
|
|
return client
|
|
except AuthenticationException as e:
|
|
pass
|
|
# ssh = paramiko.SSHClient()
|
|
return False
|
|
|
|
|
|
config=CONFIG()
|
|
|
|
bs2 = USER(name='bs2cloud', password='xxxx')
|
|
ibmuser = USER(name='ibmuser', password='xxxx')
|
|
home = USER(name='testing', password='testing')
|
|
pi = USER(name='pifail', password='pifail')
|
|
|
|
## found= user[]
|
|
result_dic = {
|
|
"found": {},
|
|
"discards": [],
|
|
}
|
|
|
|
user_list: [USER] = [home, bs2, ibmuser, pi]
|
|
host_list = ['192.168.1.3', '192.168.1.2']
|
|
|
|
|
|
|
|
for user in user_list:
|
|
result_dic['found'][user.name] = []
|
|
|
|
for host in host_list:
|
|
for user in user_list:
|
|
result = attempt_to_login(hostname=host, user=user)
|
|
if result:
|
|
session = ITEM(hostname=host, client=result)
|
|
result_dic['found'][user.name].append(session)
|
|
host_list.remove(host)
|
|
session.validations.loged_in = True
|
|
|
|
print(result_dic['found'])
|
|
|
|
result_dic['discards'] = set(host_list)
|
|
|
|
print('discards')
|
|
print(result_dic['discards'])
|
|
|
|
print('SCP')
|
|
|
|
for user in user_list:
|
|
for session in result_dic['found'][user.name]:
|
|
session: ITEM
|
|
|
|
scp = SCPClient(session.client.get_transport())
|
|
for file in config.files_to_copy:
|
|
print(f"\t{file}")
|
|
scp.put(file, remote_path='/tmp')
|
|
|
|
|
|
def test_connections():
|
|
pass
|
|
|
|
def menu():
|
|
pass
|
|
|
|
def main():
|
|
pass
|
|
# print(json.dumps(result_dic, indent=2, sort_keys=True))
|
|
|
|
# server = "192.168.1.3"
|
|
# username = "fadm"
|
|
# password = ""
|
|
# ssh = paramiko.SSHClient()
|
|
# ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
# ssh.connect(server, username=username, password=password)
|
|
# ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command("uptime")
|
|
# print(ssh_stdout.read().decode())
|