Skip to content

Instantly share code, notes, and snippets.

@vladwa
Last active December 31, 2017 07:29
Show Gist options
  • Save vladwa/26ee97c2b99324c99c81a662a7d977eb to your computer and use it in GitHub Desktop.
Save vladwa/26ee97c2b99324c99c81a662a7d977eb to your computer and use it in GitHub Desktop.
Python code to execute command as a non sudo user over ssh connection on a remote server using "paramiko" module. On The code snippet establishes connection and executes the command and return the status and output of the executed command.
import logging
import paramiko
class SSH:
def __init__(self):
pass
def get_ssh_connection(self, ssh_machine, ssh_username, ssh_password):
"""Establishes a ssh connection to execute command.
:param ssh_machine: IP of the machine to which SSH connection to be established.
:param ssh_username: User Name of the machine to which SSH connection to be established..
:param ssh_password: Password of the machine to which SSH connection to be established..
returns connection Object
"""
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(hostname=ssh_machine, username=ssh_username, password=ssh_password, timeout=10)
return client
def run_command(self, ssh_username="root", ssh_password="abc123", ssh_machine="localhost", command="ls",
jobid="None", job_details=""):
"""Executes a command over a established SSH connectio.
:param ssh_machine: IP of the machine to which SSH connection to be established.
:param ssh_username: User Name of the machine to which SSH connection to be established..
:param ssh_password: Password of the machine to which SSH connection to be established..
returns status of the command executed and Outpuf of the command.
"""
conn = self.get_ssh_connection(ssh_machine=ssh_machine, ssh_username=ssh_username, ssh_password=ssh_password)
stdin, stdout, stderr = conn.exec_command(command="hostname")
stdoutput = [line for line in stdout]
logging.debug("Job[%s]: Executing: %s on Host: %s" % (jobid, command, stdoutput))
stdin, stdout, stderr = conn.exec_command(command=command)
stdoutput = [line for line in stdout]
stderroutput = [line for line in stderr]
for output in stdoutput:
logging.info("Job[%s]: %s" % (jobid, output.strip()))
# Check exit code.
logging.debug("Job[%s]:stdout: %s" % (jobid, stdoutput))
logging.debug("Job[%s]:Job details: %s" % (jobid, job_details))
logging.debug("Job[%s]:stderror: %s, job_details: %s" % (jobid, stderroutput, job_details))
logging.info("Job[%s]:Command status: %s" % (jobid, stdout.channel.recv_exit_status()))
if not stdout.channel.recv_exit_status():
logging.info("Job[%s]: Command executed." % jobid)
conn.close()
if not stdoutput:
stdoutput = True
return True, stdoutput
else:
logging.error("Job[%s]: Command failed." % jobid)
for output in stderroutput:
logging.error("Job[%s]: %s" % (jobid, output))
conn.close()
return False, stderroutput
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment