Python을 사용하여 ssh를 통해 명령 수행
저는 Python에서 일부 명령줄 명령을 자동화하기 위해 스크립트를 작성하고 있습니다.현재 저는 다음과 같은 전화를 하고 있습니다.
cmd = "some unix command"
retcode = subprocess.call(cmd,shell=True)
하지만 원격 컴퓨터에서 몇 가지 명령을 실행해야 합니다.수으로로면려를 합니다.ssh그런 다음 명령을 실행합니다.Python에서 이를 어떻게 자동화합니까?할 수 없습니다.cmd = ssh user@remotehost제가 사용해야 할 모듈이 있는지 궁금합니다.
파라미코에게 소개해 드리겠습니다.
ssh = paramiko.SSHClient()
ssh.connect(server, username=username, password=password)
ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command(cmd_to_execute)
SSH 키를 사용하는 경우 다음을 수행합니다.
k = paramiko.RSAKey.from_private_key_file(keyfilename)
# OR k = paramiko.DSSKey.from_private_key_file(keyfilename)
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(hostname=host, username=user, pkey=k)
단순하게 받아들여라.라이브러리가 필요하지 않습니다.
import subprocess
# Python 2
subprocess.Popen("ssh {user}@{host} {cmd}".format(user=user, host=host, cmd='ls -l'), shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()
# Python 3
subprocess.Popen(f"ssh {user}@{host} {cmd}", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()
또는 commands.getstatusoutput:
commands.getstatusoutput("ssh machine 1 'your script'")
저는 그것을 광범위하게 사용했고 그것은 아주 잘 작동합니다.
Python 2.6+에서는 를 사용합니다.
파라미코는 너무 낮은 수준이고 패브릭은 라이브러리로 사용하기에 특별히 적합하지 않다는 것을 알게 되었습니다. 그래서 저는 파라미코를 사용하여 조금 더 나은 인터페이스를 구현하는 spur라는 제 라이브러리를 만들었습니다.
import spur
shell = spur.SshShell(hostname="localhost", username="bob", password="password1")
result = shell.run(["echo", "-n", "hello"])
print result.output # prints hello
셸 내부에서 실행해야 하는 경우:
shell.run(["sh", "-c", "echo -n hello"])
모두 파라미코를 사용하여 이미 언급(권장)했으며 저는 한 번에 여러 명령을 실행할 수 있는 파이썬 코드(API라고 할 수 있음)를 공유하고 있습니다.
명령을 사용합니다.Commands().run_cmd(host_ip, list_of_commands)
실행에 실패한 명령이 있으면 실행을 중지하기 위해 보관한 TODO가 하나 표시됩니다. 실행 방법을 모르겠습니다. 지식을 공유하십시오.
#!/usr/bin/python
import os
import sys
import select
import paramiko
import time
class Commands:
def __init__(self, retry_time=0):
self.retry_time = retry_time
pass
def run_cmd(self, host_ip, cmd_list):
i = 0
while True:
# print("Trying to connect to %s (%i/%i)" % (self.host, i, self.retry_time))
try:
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(host_ip)
break
except paramiko.AuthenticationException:
print("Authentication failed when connecting to %s" % host_ip)
sys.exit(1)
except:
print("Could not SSH to %s, waiting for it to start" % host_ip)
i += 1
time.sleep(2)
# If we could not connect within time limit
if i >= self.retry_time:
print("Could not connect to %s. Giving up" % host_ip)
sys.exit(1)
# After connection is successful
# Send the command
for command in cmd_list:
# print command
print "> " + command
# execute commands
stdin, stdout, stderr = ssh.exec_command(command)
# TODO() : if an error is thrown, stop further rules and revert back changes
# Wait for the command to terminate
while not stdout.channel.exit_status_ready():
# Only print data if there is data to read in the channel
if stdout.channel.recv_ready():
rl, wl, xl = select.select([ stdout.channel ], [ ], [ ], 0.0)
if len(rl) > 0:
tmp = stdout.channel.recv(1024)
output = tmp.decode()
print output
# Close SSH connection
ssh.close()
return
def main(args=None):
if args is None:
print "arguments expected"
else:
# args = {'<ip_address>', <list_of_commands>}
mytest = Commands()
mytest.run_cmd(host_ip=args[0], cmd_list=args[1])
return
if __name__ == "__main__":
main(sys.argv[1:])
수락된 답변은 저에게 효과가 없었습니다. 대신 사용한 내용은 다음과 같습니다.
import paramiko
import os
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# ssh.load_system_host_keys()
ssh.load_host_keys(os.path.expanduser('~/.ssh/known_hosts'))
ssh.connect("d.d.d.d", username="user", password="pass", port=22222)
ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command("ls -alrt")
exit_code = ssh_stdout.channel.recv_exit_status() # handles async exit error
for line in ssh_stdout:
print(line.strip())
total 44
-rw-r--r--. 1 root root 129 Dec 28 2013 .tcshrc
-rw-r--r--. 1 root root 100 Dec 28 2013 .cshrc
-rw-r--r--. 1 root root 176 Dec 28 2013 .bashrc
...
또는 sshpass를 사용할 수 있습니다.
import subprocess
cmd = """ sshpass -p "myPas$" ssh user@d.d.d.d -p 12345 'my command; exit' """
print( subprocess.getoutput(cmd) )
참조:
주의:
- SSH를한번(SSH 통해원시에템한연합니결다야해로으동수상이번를스격(▁just합니다▁at▁make야▁manuallyss▁via▁(▁sure▁connect▁to연▁the▁system▁ssh▁remote결해▁toh▁time
ssh root@ip) 및 공개 키를 수락합니다. 이는 다음을 사용하여 연결할 수 없는 이유의 몇 배입니다.paramiko자동화된 또는자화된동ssh대본
paramiko는 누락된 호스트 키 정책을 허용하는 추가 라인을 추가한 후에 마침내 저를 위해 일했습니다. 이는 정말 중요한 라인 5입니다.
import paramiko
p = paramiko.SSHClient()
# This script doesn't work for me unless the following line is added!
p.set_missing_host_key_policy(paramiko.AutoAddPolicy())
p.connect("server", port=22, username="username", password="password")
stdin, stdout, stderr = p.exec_command("your command")
opt = stdout.readlines()
opt = "".join(opt)
print(opt)
파라미코 패키지가 설치되어 있는지 확인합니다.솔루션의 원래 소스:출처
번째:하지 않은 것에 놀랐습니다.fabric 아직은.
번째: 그런 저는 두번: 정째그라는 했습니다.jk_simpleexec: 쉽게 합니다.명령 실행을 쉽게 합니다.
제가 그것에 대해 조금 설명해 드리겠습니다.
'로컬로 명령 실행' 문제
모듈 ㅠㅠㅠㅠㅠjk_simpleexec는 라이름함제다니라는 을 제공합니다.runCmd(..)로컬 또는 원격으로 셸(!) 명령을 실행할 수 있습니다.이것은 매우 간단합니다.다음은 명령의 로컬 실행 예입니다.
import jk_simpleexec
cmdResult = jk_simpleexec.runCmd(None, "cd / ; ls -la")
참고: STDOUT 및 STDERR에서 과도한 빈 줄을 제거하기 위해 반환된 데이터는 기본적으로 자동으로 트리밍됩니다. (물론 이 동작은 비활성화될 수 있지만, 정확히 그 목적을 위해 사용자가 원하는 동작을 염두에 두고 있습니다.)
'결과 처리' 문제
당신이 받게 될 것은 반환 코드인 STDOUT와 STDERR이 포함된 객체입니다.따라서 결과를 처리하는 것은 매우 쉽습니다.
실행하는 명령이 존재하고 실행되지만 의도한 작업을 수행하지 못할 수 있으므로 이 작업을 수행할 수 있습니다.STDOUT와 STDERR에 관심이 없는 가장 간단한 경우 코드는 다음과 같습니다.
cmdResult.raiseExceptionOnError("Something went wrong!", bDumpStatusOnError=True)
디버깅을 위해 결과를 STDOUT로 출력하여 다음 작업을 수행할 수 있습니다.
cmdResult.dump()
STDOUT를 처리하고 싶다면 그것도 간단합니다.예:
for line in cmdResult.stdOutLines:
print(line)
'원격으로 명령 실행' 문제
물론 다른 시스템에서 이 명령을 원격으로 실행해야 할 수도 있습니다.를 위해 을 사용할 수 있습니다.runCmd(..)정확히 같은 방식으로 하지만 우리는 특정할 필요가 있습니다.fabric먼저 연결 개체입니다.이 작업은 다음과 같이 수행할 수 있습니다.
from fabric import Connection
REMOTE_HOST = "myhost"
REMOTE_PORT = 22
REMOTE_LOGIN = "mylogin"
REMOTE_PASSWORD = "mypwd"
c = Connection(host=REMOTE_HOST, user=REMOTE_LOGIN, port=REMOTE_PORT, connect_kwargs={"password": REMOTE_PASSWORD})
cmdResult = jk_simpleexec.runCmd(c, "cd / ; ls -la")
# ... process the result stored in cmdResult ...
c.close()
모든 것이 그대로 유지되지만 이번에는 다른 호스트에서 이 명령을 실행합니다.이 작업은 다음과 같습니다.언젠가 당신이 로컬 호스트에서 다른 호스트로 이동하기로 결정한다면 소프트웨어에 수정 사항이 필요 없는 균일한 API를 갖고 싶었습니다.
암호 입력 문제
물론 비밀번호 문제도 있습니다.위에서 언급한 내용은 다음과 같습니다.이 파이썬 코드를 실행하는 사용자에게 암호를 요청할 수 있습니다.
이 문제를 해결하기 위해 저는 꽤 오래 전에 자체 모듈을 만들었습니다. jk_pwdinput은 일비밀입다른점은과입니다.jk_pwdinput아무것도 출력하지 않는 대신 별을 출력할 것입니다.입력하는 모든 암호 문자에 대해 별이 표시됩니다.이렇게 하면 암호를 더 쉽게 입력할 수 있습니다.
코드는 다음과 같습니다.
import jk_pwdinput
# ... define other 'constants' such as REMOTE_LOGIN, REMOTE_HOST ...
REMOTE_PASSWORD = jk_pwdinput.readpwd("Password for " + REMOTE_LOGIN + "@" + REMOTE_HOST + ": ")
(완전성을 위해:한다면readpwd(..)반환된None사용자가 Ctrl+C로 암호 입력을 취소했습니다.실제 시나리오에서는 이에 대해 적절히 조치를 취할 수 있습니다.)
전체 예
다음은 전체 예입니다.
import jk_simpleexec
import jk_pwdinput
from fabric import Connection
REMOTE_HOST = "myhost"
REMOTE_PORT = 22
REMOTE_LOGIN = "mylogin"
REMOTE_PASSWORD = jk_pwdinput.readpwd("Password for " + REMOTE_LOGIN + "@" + REMOTE_HOST + ": ")
c = Connection(host=REMOTE_HOST, user=REMOTE_LOGIN, port=REMOTE_PORT, connect_kwargs={"password": REMOTE_PASSWORD})
cmdResult = jk_simpleexec.runCmd(
c = c,
command = "cd / ; ls -la"
)
cmdResult.raiseExceptionOnError("Something went wrong!", bDumpStatusOnError=True)
c.close()
최종 노트
그래서 우리는 전체 세트를 가지고 있습니다.
- 명령 실행,
- 동일한 API를 통해 원격으로 명령을 실행합니다.
- 암호 입력을 통해 쉽고 안전한 방식으로 연결을 만듭니다.
위의 코드는 저에게 문제를 꽤 잘 해결해 줍니다(당신에게도 그러길 바랍니다).그리고 모든 것이 오픈 소스입니다.패브릭은 BSD-2-Clause이며, Apache-2에서 자체 모듈이 제공됩니다.
사용된 모듈:
- 원단 : http://www.fabfile.org/
- jk_pwdinput : https://github.com/jkpubsrc/python-module-jk-pwdinput
- jk_tftxec : https://github.com/jkpubsrc/python-module-jk-simpleexec
해피 코딩! ;-)
완벽하게 작동...
import paramiko
import time
ssh = paramiko.SSHClient()
#ssh.load_system_host_keys()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('10.106.104.24', port=22, username='admin', password='')
time.sleep(5)
print('connected')
stdin, stdout, stderr = ssh.exec_command(" ")
def execute():
stdin.write('xcommand SystemUnit Boot Action: Restart\n')
print('success')
execute()
이러한 명령을 사용할 수 있으며 암호를 지정하는 데도 도움이 됩니다.
cmd = subprocess.run(["sshpass -p 'password' ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null root@domain.com ps | grep minicom"], shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
print(cmd.stdout)
OR
cmd = subprocess.getoutput("sshpass -p 'password' ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null root@domain.com ps | grep minicom")
print(cmd)
보다spurplus우리가 개발한 포장지spur형식 주석 및 몇 가지 사소한 요령(SFTP, md5 등을 다시 연결)을 제공합니다.: https://pypi.org/project/spurplus/
사용자에게 로그인 중인 장치에 따라 명령을 입력하도록 요청합니다.
아래 코드는 PEP8online.com 에서 확인할 수 있습니다.
import paramiko
import xlrd
import time
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
loc = ('/Users/harshgow/Documents/PYTHON_WORK/labcred.xlsx')
wo = xlrd.open_workbook(loc)
sheet = wo.sheet_by_index(0)
Host = sheet.cell_value(0, 1)
Port = int(sheet.cell_value(3, 1))
User = sheet.cell_value(1, 1)
Pass = sheet.cell_value(2, 1)
def details(Host, Port, User, Pass):
time.sleep(2)
ssh.connect(Host, Port, User, Pass)
print('connected to ip ', Host)
stdin, stdout, stderr = ssh.exec_command("")
x = input('Enter the command:')
stdin.write(x)
stdin.write('\n')
print('success')
details(Host, Port, User, Pass)
#Reading the Host,username,password,port from excel file
import paramiko
import xlrd
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
loc = ('/Users/harshgow/Documents/PYTHON_WORK/labcred.xlsx')
wo = xlrd.open_workbook(loc)
sheet = wo.sheet_by_index(0)
Host = sheet.cell_value(0,1)
Port = int(sheet.cell_value(3,1))
User = sheet.cell_value(1,1)
Pass = sheet.cell_value(2,1)
def details(Host,Port,User,Pass):
ssh.connect(Host, Port, User, Pass)
print('connected to ip ',Host)
stdin, stdout, stderr = ssh.exec_command("")
stdin.write('xcommand SystemUnit Boot Action: Restart\n')
print('success')
details(Host,Port,User,Pass)
가장 현대적인 접근법은 아마도 직물을 사용하는 것일 것입니다.이 모듈을 사용하면 SSH 연결을 설정한 다음 명령을 실행하고 연결 개체를 통해 결과를 가져올 수 있습니다.
다음은 간단한 예입니다.
from fabric import Connection
with Connection("your_hostname") as connection:
result = connection.run("uname -s", hide=True)
msg = "Ran {0.command!r} on {0.connection.host}, got stdout:\n{0.stdout}"
print(msg.format(result))
원격 over 네이티브 ssh에서 명령을 실행하기 위해 간단한 클래스를 작성했습니다.subprocess모듈:
사용.
from ssh_utils import SshClient
client = SshClient(user='username', remote='remote_host', key='path/to/key.pem')
# run a list of commands
client.cmd(['mkdir ~/testdir', 'ls -la', 'echo done!'])
# copy files/dirs
client.scp('my_file.txt', '~/testdir')
클래스 소스 코드
https://gist.github.com/mamaj/a7b378a5c969e3e32a9e4f9bceb0c5eb
import subprocess
from pathlib import Path
from typing import Union
class SshClient():
""" Perform commands and copy files on ssh using subprocess
and native ssh client (OpenSSH).
"""
def __init__(self,
user: str,
remote: str,
key_path: Union[str, Path]) -> None:
"""
Args:
user (str): username for the remote
remote (str): remote host IP/DNS
key_path (str or pathlib.Path): path to .pem file
"""
self.user = user
self.remote = remote
self.key_path = str(key_path)
def cmd(self,
cmds: list[str],
strict_host_key_checking=False) -> None:
"""runs commands consecutively, ensuring success of each
after calling the next command.
Args:
cmds (list[str]): list of commands to run.
strict_host_key_checking (bool, optional): Defaults to True.
"""
strict_host_key_checking = 'yes' if strict_host_key_checking \
else 'no'
cmd = ' && '.join(cmds)
subprocess.run(
[
'ssh',
'-i', self.key_path,
'-o', f'StrictHostKeyChecking={strict_host_key_checking}',
'-o', 'UserKnownHostsFile=/dev/null',
f'{self.user}@{self.remote}',
cmd
]
)
def scp(self, source: Union[str, Path], destination: Union[str, Path]):
"""Copies `srouce` file to remote `destination` using the
native `scp` command.
Args:
source (Union[str, Path]): Source file path.
destination (Union[str, Path]): Destination path on remote.
"""
subprocess.run(
[
'scp',
'-i', self.key_path,
str(source),
f'{self.user}@{self.remote}:{str(destination)}',
]
)
아래 예는 호스트 이름, 사용자 이름, 암호 및 포트 번호에 대한 사용자 입력을 원하는 경우입니다.
import paramiko
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
def details():
Host = input("Enter the Hostname: ")
Port = input("Enter the Port: ")
User = input("Enter the Username: ")
Pass = input("Enter the Password: ")
ssh.connect(Host, Port, User, Pass, timeout=2)
print('connected')
stdin, stdout, stderr = ssh.exec_command("")
stdin.write('xcommand SystemUnit Boot Action: Restart\n')
print('success')
details()
언급URL : https://stackoverflow.com/questions/3586106/perform-commands-over-ssh-with-python
'programing' 카테고리의 다른 글
| 스프링 부트 테스트 - 테스트 속성을 찾을 수 없음 (0) | 2023.06.25 |
|---|---|
| 보안 규칙에서 화재 기준 비율 제한? (0) | 2023.06.25 |
| 스프링 데이터 탄력적 검색을 사용한 탄력적 검색 Rest 클라이언트 (0) | 2023.06.25 |
| Oracle Optimizer가 동일한 SELECT에서 여러 힌트를 사용합니까? (0) | 2023.06.25 |
| SQL Server를 사용하여 폴더 내의 파일을 나열하는 방법 (0) | 2023.06.25 |