Close
    logoCloudomation Docs

    TaskSSH

    class tasks.task_ssh.TaskSSH

    Connect to a remote host using SSH and execute a script.

    Inputs

    NameTypeDefaultDescription
    connect_timeoutNumber60How long to wait for a response from the server. Only applies after a successful connection. If a connection is impossible the Task fails immediately.
    connect_triesNumber3How many times to try to connect.
    copy_filessetNone
    hostkeyString
    hostnameString
    interpreterstr/usr/bin/env bash -e
    keyStringNone
    output_filessetNone
    output_varssetNone
    passwordStringNone
    portNumber22
    remove_ansi_escapesboolFalse
    remove_crboolTrue
    scriptString
    script_timeoutint60
    temp_pathstr/tmp
    use_shellboolFalse
    usernameString

    Outputs

    NameTypeDefaultDescription
    filesdictThe names of the output files which were registered using #OUTPUT_FILE(path)
    handler_reportstr
    reportstrThe outputs your scripts produce on the remote systems
    retcodeint
    varsdictThe content of all variables which were registered using #OUTPUT_VAR(variable)
    waiter_reportstr
    execution_idintThe ID of the task execution
    messagestrThe ended message for the task. If the task ended with an error, the message will contain information about what went wrong
    statusstrThe ended status for the task. Either "success" or "error".

    Constants

    input_list = ['connect_timeout', 'connect_tries', 'copy_files', 'hostkey', 'hostname', 'interpreter', 'key', 'output_files', 'output_vars', 'password', 'port', 'remove_ansi_escapes', 'remove_cr', 'script', 'script_timeout', 'temp_path', 'use_shell', 'username']output_list = ['files', 'handler_report', 'report', 'retcode', 'vars', 'waiter_report']version = 1

    Methods

    get_file (src, dst)
    run ()

    Example

    import flow_api
    def handler(system: flow_api.System, this: flow_api.Execution):
    # Authenticate using private key
    info_task = this.task(
    'SSH',
    # public accessible name or IP
    hostname='my-ssh-server',
    # key to check host identity.
    # can be read with "$ ssh-keyscan -t rsa <my-ssh-server>"
    hostkey='ssh-rsa AAAAB3NzaC1yc2E...',
    username='kevin',
    key='-----BEGIN RSA PRIVATE KEY-----\nMII...',
    script=(
    '''
    HOSTNAME=$(hostname)
    USERNAME=$(id -un)
    CPU=$(uname -p)
    #OUTPUT_VAR(HOSTNAME)
    #OUTPUT_VAR(USERNAME)
    #OUTPUT_VAR(CPU)
    '''
    ),
    )
    outputs = info_task.get('output_value')
    hostname = outputs['var']['HOSTNAME']
    username = outputs['var']['USERNAME']
    cpu = outputs['var']['CPU']
    this.log(f'info_task was running on {hostname} using {cpu} as {username}')
    # Authenticate using password
    uptime_task = this.task(
    'SSH',
    hostname='my-ssh-server',
    hostkey='ssh-rsa AAAAB3NzaC1yc2E...',
    username='kevin',
    password='***',
    script=(
    '''
    UPTIME=$(uptime -s)
    #OUTPUT_VAR(UPTIME)
    '''
    ),
    )
    outputs = uptime_task.get('output_value')
    uptime = outputs['var']['UPTIME']
    this.log(f'{hostname} is up since {uptime}')
    return this.success('all done')

    More

    Output variables

    There are two ways how to define "output variables":

    • from the flow starting the task, in the output_vars field of the input dictionary
    • from inside the task, in the script field of the input dictionary

    Output variables in output_vars

    You can register shell variables as "output variables" in the output_vars field of the input dictionary, e.g.:

    task = this.task(
    'SSH',
    hostname='my-ssh-server',
    hostkey='ssh-rsa AAAAB3NzaC1yc2E...',
    username='kevin',
    key='-----BEGIN RSA PRIVATE KEY-----\nMII...',
    script='''
    VALUE=foo
    ''',
    name='output_var',
    output_vars=['VALUE'],
    )
    assert task.get('output_value')['var']['VALUE'] == 'foo'

    Output variables in script

    You can register shell variables as "output variables" using #OUTPUT_VAR(variable_name):

    VARIABLE="some content"
    #OUTPUT_VAR(VARIABLE)

    The value of registered variables is available to the calling flow script in the var dictionary of the task outputs:

    outputs = task(...).get('output_value')
    variable = outputs['var']['VARIABLE']
    # `variable` contains "some content"

    Output files

    There are two ways how to define "output files":

    • from the flow starting the task, in the output_files field of the input dictionary
    • from inside the task, in the script field of the input dictionary

    Output files in output_files

    You can register files as "output files" in the output_files field of the input dictionary, e.g.:

    task = this.task(
    'SSH',
    hostname='my-ssh-server',
    hostkey='ssh-rsa AAAAB3NzaC1yc2E...',
    username='kevin',
    key='-----BEGIN RSA PRIVATE KEY-----\nMII...',
    script='''
    echo -n "spam" > file.txt
    ''',
    name='output_var',
    output_files=['file.txt'],
    )
    assert 'file.txt' in task.get('output_value')['files']
    assert system.file('file.txt').get('content') == 'spam'

    Output files in script

    You can register files as "output files" using #OUTPUT_FILE(filename):

    task = this.task(
    'SSH',
    hostname='my-ssh-server',
    hostkey='ssh-rsa AAAAB3NzaC1yc2E...',
    username='kevin',
    key='-----BEGIN RSA PRIVATE KEY-----\nMII...',
    script='''
    echo -n "egg" > file2.txt
    #OUTPUT_FILE(file2.txt)
    ''',
    name='output_var',
    )
    assert 'file2.txt' in task.get('output_value')['files']
    assert system.file('file2.txt').get('content') == 'egg'
    Previous
    TaskSQLPG
    Next
    TaskVAULT