Basic automation of SSH tasks in a restricted environment

Question

I'm trying to implement a simple automation solution in a restricted environment. I have actually implemented something but it took a bit longer than I wanted. I want to improve the solution and I'll probably have to extend it in the future so I'm asking for code review - my boss will probably be happier if I can deliver things faster! :)

Anyway, the current setup:

Python 2.6 - installed as a portable package since I don't have administrative rights on my machine.

Fabric - simplest (only Paramiko, pyasn1 and cyrptography as pypi dependencies; a strong bonus in such a restrictive environment) and most robust solution I could find for running on Python 2.6

Git Bash - for smaller scripts that launch Fabric.

• ./launcher - as the name says, the entry point for the tool, a shell script
• ./commands/ - shell scripts that are just parametrized launches of Fabric
• ./configurations/ - shell scripts that contain just environment variables split per environment and application
• ./bash/completion - bash completion script for use with the launcher

And the actual code, starting with the launcher:

./launcher

#!/bin/bash

set -o errexit   # set -e
set -o nounset  # set -u
set -o pipefail

script_usage="Script usage: ${0} <environment> <application> <command>."

if [[ -z "${1-}" ]]
then echo "Environment name parameter not provided. $script_usage"
    exit 1
fi
if [[ -z "${2-}" ]]
then echo "Application parameter not provided. $script_usage"
    exit 1
fi
if [[ -z "${3-}" ]]
then echo "Command parameter not provided. $script_usage"
    exit 1
fi

export FAB_ENVIRONMENT="${1}"

cd fabfile
python -c "from common import *; environment_prompt(\"${FAB_ENVIRONMENT}\")";
cd ..

source "configurations/${1}/${2}"

./commands/"${3}"

./fabfile/common.py

##- Imports. 
from fabric.api import *

from getpass import getpass

import os
#-##

##- Environment, user, password.
def get_base_info():
    try:
        environment = os.environ['FAB_ENVIRONMENT']
    except KeyError:
        abort('FAB_ENVIRONMENT not defined, cannot connect to servers.')

    try:
        user = os.environ['SSH_USER']
    except KeyError:
        abort('SSH_USER not defined, cannot connect to servers.')

    try:
        password = os.environ['SSH_PASSWORD']
    except KeyError:
        password = getpass()

    return (environment, user, password)
#-##

##- Prompt answers.
def get_prompt_answers():
    return {
        # Initial SSH connection.
        'Are you sure you want to continue connecting (yes/no)? ': 'yes',
        # SSH password requests.
        'Password: ': env.password,
        'password: ': env.password,
        'Login password for \'{0}\': '.format(env.user): env.password,
        # Sudo password request.
        '[sudo] password for {0}: '.format(env.user): env.password,
        '{0}\'s password:'.format(env.user): env.password
    }
#-##

##- Base functions.
##- check_empty_arguments.
def check_empty_arguments(**arguments):
    for argument in arguments:
        if arguments[argument] == '':
            abort('Parameter {0} was empty!'.format(argument))
#-##

##- fix_absolute_path.
def fix_absolute_path(path):
    # Absolute paths are mangled by the Fabric + Git Cygwin interaction.
    # As a result the absolute paths gets the path to the Git installation
    # prefixed and the resulting path is a mess.
    # To fix this we get the Git installation path, delete it from the
    # absolute path and then return the fixed path.
    git_root = ''
    with hide('running'):
        git_root = local('cygpath -w /', capture=True).replace('\\', '/')
    return path.replace(git_root, '')
#-##

##- info.
def info(message, environment):
    if environment == 'local' or environment == 'dev' or environment == 'test':
        with hide('running'):
            local('bash -c \'echo -e "{0}"\''.format(message))
    elif environment == 'hom':
        with hide('running'):
            yellow = '\e[33m'
            clear = '\e[0m'
            local('bash -c \'echo -e "{0}{1}{2}"\''.format(yellow, message, clear))
    elif environment == 'prod':
        with hide('running'):
            red = '\e[31m'
            clear = '\e[0m'
            local('bash -c \'echo -e "{0}{1}{2}"\''.format(red, message, clear))
#-##

##- environment_prompt.
def environment_prompt(environment):
    line_length = 80
    marker = '#'
    line_padding = '  '
    line_start = '##'
    line_end = '##'
    separator_line = '\\\\n'
    highlight_line =  line_padding + marker * line_length
    padding_length = line_length - (len(line_start) + len(line_end))
    empty_line = line_padding + line_start + ' ' * padding_length + line_end
    whitespace = ' ' * ((padding_length - len(environment)) / 2)
    environment_line = line_padding + line_start  + whitespace + environment.upper() + whitespace
    if (len(environment) % 2) == 0:
        environment_line += line_end
    else:
        environment_line += ' ' + line_end
    prompt_start = [ separator_line, highlight_line, highlight_line, empty_line, empty_line ]
    prompt_end = list(reversed(prompt_start))
    prompt_elements = prompt_start + [ environment_line ] + prompt_end
    prompt = '\\\\n'.join(prompt_elements)
    info(prompt, environment)
#-##
#-##

##- Task functions.
@task
def scp_file(source_server, file_to_copy, source_folder, target_folder):
    with settings(prompts=get_prompt_answers()):
        return run('scp {0}:{1} {2}'.format(
            source_server,
            "{0}/{1}".format(source_folder, file_to_copy),
            target_folder))
@task
def run_as_user(command):
    with settings(prompts=get_prompt_answers()):
        return run(command)
@task
def run_as_application_user(command, application_user):
    with settings(prompts=get_prompt_answers()):
        heredoc = '<<END\n'
        heredoc += '{0}\n'.format(command)
        heredoc += 'END'
        return run('sudo su - {0}'.format(application_user) + heredoc)
#-##

./fabfile/package.py

##- Imports.
from common import *

from fabric.api import *
from fabric.utils import *

from datetime import datetime

import os
import re
import sys
#-##

##- Initial setup.
environment, env.user, env.password = get_base_info()
#-##

##- Scp.
@task
def scp(file_to_copy, source_folder, target_folder, source_server, target_server, target_user):
    check_empty_arguments(**locals())
    source_servers = [ source_server ]
    intermediary_servers = [ 'intermediary.dev.com' ]
    target_servers = [ target_server ]

    source_folder = fix_absolute_path(source_folder)

    # Short server name, used as a subfolder in /data/packages on ldps101.
    # Example: lcob, lbat, etc.
    server_folder = re.sub(r'[0-9]*', '', target_server.split('.')[0])

    copy_time = datetime.now().strftime('%Y%m%d')
    intermediary_folder =  '/data/packages/{0}/{1}'.format(server_folder, target_user)
    target_tmp_folder = '/tmp/{0}/{1}/{2}'.format(server_folder, target_user, copy_time)

    target_folder = fix_absolute_path(target_folder)

    execute(run_as_user, 'mkdir -pv {0}'.format(intermediary_folder), hosts=intermediary_servers)
    execute(scp_file, source_servers[0], file_to_copy, source_folder, intermediary_folder,
            hosts=intermediary_servers)

    execute(run_as_user, 'mkdir -pv {0}'.format(target_tmp_folder), hosts=target_servers)
    execute(scp_file, intermediary_servers[0], file_to_copy, intermediary_folder, target_tmp_folder,
            hosts=target_servers)
    execute(run_as_application_user, 'mkdir -pv {0}'.format(target_folder), target_user,
            hosts=target_servers)
    execute(run_as_application_user,
            'cp -v {0}/{1} {2}'.format(
                target_tmp_folder,
                file_to_copy,
                target_folder
                ),
            target_user,
            hosts=target_servers)
    execute(run_as_application_user, 'ls {0}'.format(target_folder), target_user,
            hosts=target_servers)
    execute(run_as_user, 'rm -rvf {0}'.format(target_tmp_folder), hosts=target_servers)
#-##

##- Archive.
@task
def archive(installation_folder, new_archive, exclusions, application_user, target_server):
    check_empty_arguments(**locals())
    target_servers = [ target_server ]

    installation_folder = fix_absolute_path(installation_folder)

    tar_list_command = 'tar --exclude="*/*" -tvf {0}/package-repo/{1}'
    tar_list_command = tar_list_command.format(installation_folder, new_archive)

    zip_list_command = 'zipinfo -1 {0}/package-repo/{1} -x "*/**"'.format(
            installation_folder, new_archive)

    if new_archive.endswith('.tar'):
        archive_list_command = tar_list_command
        new_archive_basename = new_archive[:-len('.tar')]
        archive_format = '.tar'
        archive_command = 'tar -cvf'
    elif new_archive.endswith('.tar.gz'):
        archive_list_command = tar_list_command
        new_archive_basename = new_archive[:-len('.tar.gz')]
        archive_format = '.tar.gz'
        archive_command = 'tar -czvf'
    elif new_archive.endswith('.tar.bz2'):
        archive_list_command = tar_list_command
        new_archive_basename = new_archive[:-len('.tar.bz2')]
        archive_format = '.tar.bz2'
        archive_command = 'tar -cjvf'
    elif new_archive.endswith('.zip'):
        archive_list_command = zip_list_command
        new_archive_basename = new_archive[:-len('.zip')]
        archive_format = '.zip'
        archive_command = 'zip -r'
    else:
        print('Unsupported archive format. Supported formats: tar, tar.gz, tar.bz2, zip.')
        sys.exit(1)

    archive_time = datetime.now().strftime('%Y%m%d')
    archive_name = '{0}_{1}.bak{2}'.format(new_archive_basename, archive_time, archive_format)

    # Get the list of top-level files and folders in the archive.
    execute_output = execute(run_as_application_user,
            archive_list_command,
            application_user,
            hosts=target_servers)
    new_archive_list = []
    for new_archive_file in execute_output[target_server].split('\r\n'):
        new_archive_list.append(new_archive_file.split(' ')[-1])
    new_archive_list = set(new_archive_list)

    # Get the list of top-level files and folders in the installation folder.
    execute_output = execute(run_as_application_user,
            'ls -p1 {0}'.format(installation_folder),
            application_user,
            hosts=target_servers)
    installation_folder_list = set(execute_output[target_server].split('\r\n'))

    exclusions = set(exclusions.split(','))

    # The files to archive are files that are both in the new tar and in the current installation.
    # We don't archive excluded files (temporary files, input files, output files, in general).
    to_archive_list = new_archive_list.intersection(installation_folder_list).difference(exclusions)
    to_archive_list = ' '.join(to_archive_list)


    if len(to_archive_list):
        archive_command = '{0} {1}/package-repo/{2} {3}'.format(
            archive_command, installation_folder, archive_name, to_archive_list)
        execute(run_as_application_user, archive_command, application_user, hosts=target_servers)
    else:
        warn('Not archiving anything as the list of files to archive was empty.')
#-##

##- Clean.
@task
def clean(target_folder, application_user, target_server):
    check_empty_arguments(**locals())
    target_servers = [ target_server ]

    target_folder = fix_absolute_path(target_folder)

    clean_command = 'rm -rfv {0}'.format(target_folder)
    execute(run_as_application_user, clean_command, application_user, hosts=target_servers)
#-##

##- Unpack.
@task
def unpack(archive_name, installation_folder, application_user, target_server):
    check_empty_arguments(**locals())
    target_servers = [ target_server ]

    installation_folder = fix_absolute_path(installation_folder)

    if archive_name.endswith('.tar'):
        unpack_command = 'tar -xvpf'
        unpack_option = '-C'
    elif archive_name.endswith('.tar.gz'):
        unpack_command = 'tar -xzvpf'
        unpack_option = '-C'
    elif archive_name.endswith('.tar.bz2'):
        unpack_command = 'tar -xjvpf'
        unpack_option = '-C'
    elif archive_name.endswith('.zip'):
        unpack_command = 'unzip -o'
        unpack_option = '-d'
    else:
        print('Unsupported archive format. Supported formats: tar, tar.gz, tar.bz2, zip.')
        sys.exit(1)

    unpack_command = '{0} {1}/package-repo/{2} {3} {1}'.format(
            unpack_command, installation_folder, archive_name, unpack_option)
    execute(run_as_application_user, unpack_command, application_user, hosts=target_servers)
#-##

./configurations/local/test-app

# Configuration values for the scp command.
export SCP_FILE_TO_COPY=test-app-1.0.0.tar.gz
export SCP_SOURCE_FOLDER=/home/appuser/todeploy
export SCP_TARGET_FOLDER=/home/appuser/package-repo
export SCP_SOURCE_SERVER=app.dev.com
export SCP_TARGET_SERVER=app.dev.com
export SCP_TARGET_USER=appuser

# Configuration values for the archive command.
export ARCHIVE_INSTALLATION_FOLDER=/home/appuser
export ARCHIVE_NEW_ARCHIVE=test-app-1.0.0.tar.gz
export ARCHIVE_EXCLUSIONS=RELEASE.txt
export ARCHIVE_APPLICATION_USER=appuser
export ARCHIVE_TARGET_SERVER=app.dev.com

# Configuration values for the clean command.
export CLEAN_TARGET_FOLDER=/home/appuser/test-app
export CLEAN_APPLICATION_USER=appuser
export CLEAN_TARGET_SERVER=app.dev.com

# Configuration values for the deploy command.
export UNPACK_INSTALLATION_FOLDER=/home/appuser
export UNPACK_ARCHIVE_NAME=test-app-1.0.0.tar.gz
export UNPACK_APPLICATION_USER=appuser
export UNPACK_TARGET_SERVER=app.dev.com

./commands/scp

#!/bin/bash

set -o errexit   # set -e
set -o nounset  # set -u
set -o pipefail

fab package.scp:file_to_copy=${SCP_FILE_TO_COPY},source_folder=${SCP_SOURCE_FOLDER},target_folder=${SCP_TARGET_FOLDER},source_server=${SCP_SOURCE_SERVER},target_server=${SCP_TARGET_SERVER},target_user=${SCP_TARGET_USER}

The usage is:

./launcher local test-app scp

The scripts are supposed to automate various things done manually right now, using SSH and SCP.

Answer 1

./launcher

I'm no Bash person, so I have little to say about your launcher except where you call Python:

python -c "from common import *; environment_prompt(\"${FAB_ENVIRONMENT}\")";

You're inserting the FAB_ENVIRONMENT variable in what seems to me to be an unsafe manner. What if it has a quotation mark in it? Maybe FAB_ENVIRONMENT is some valid argument"); import os; os.remove("some important file. Your closing quotation mark and parenthesis finish the spelling of disaster. A much better way to do it is to let Python get the environment variable:

python -c "from os import getenv; import common; common.environment_prompt(os.getenv(\"FAB_ENVIRONMENT\")"

That's a rather long command, so maybe split it up a bit:

python - <<EOF
import common
from os import getenv
common.environment_prompt(getenv("FAB_ENVIRONMENT"))
EOF

You may notice that I removed your from common import *. I'll talk more about that below.

./fabfile/common.py

def check_empty_arguments(**arguments):
    ...
    abort(...)

I'm not sure it's a good idea for something named check_... to do something when it's wrong. Either make it clear in the name of the function that it takes action or define it so that the calling function decides what to do:

def check_empty_arguments(**arguments):
    return all(arguments.values())

(all() sees if every item given to it is "truthy". An empty string is not, so it will return True only if every string is non-empty.)

def fix_absolute_path(path):
    ...
    return path.replace(git_root, '')

I don't like that replacement. A more reliable method would be to use a built-in function:

if git_root:
    return os.path.relpath(path, git_root)
return path

if environment == ... or environment == ... or environment == ...:

Instead of chaining ors, just use in:

if environment in {'local', 'dev', 'test'}:

I'm also wondering about why you use bash -c ... instead of just .... Maybe you have a good reason, but it seems a little strange to me.

def environment_prompt(environment):
    line_length = 80
    marker = '#'
    line_padding = ' '
    ...

All those variables are really constants. Even though some of them depend on others, they never change once they're first defined. Python's naming convention uses ALL_CAPS for constants. I might also put them at the top of the file (as suggested in PEP 8).

Another advantage to putting them outside the function is that they aren't then re-defined every time the function is called.

./fabfile/package.py

server_folder = re.sub(r'[0-9]*', '', ...)

All you're doing here is removing the digits from the string. A regex is probably not the most efficient way to do that. A solution that I prefer comes from a Stack Overflow answer:

from string import digits

server_folder = target_server.split('.')[0].translate(None, digits)

if new_archive.endswith('.tar'):
elif new_archive.endswith('.tar.gz'):
elif new_archive.endswith('.tar.bz2'):
elif new_archive.endswith('.zip'):

That seems too repetitive to me. Perhaps a better way would be this:

archive_formats = (
    ('.tar', tar_list_command, 'tar -cvf'),
    ('.tar.gz', tar_list_command, 'tar -czvf'),
    ('.tar.bz2', tar_list_command, 'tar -cjvf'),
    ('.zip', zip_list_command, 'zip -r'),
)

for archive_format, archive_list_command, archive_command in archive_formats:
    if new_archive.endswith(archive_format):
        new_archive_basename = new_archive[:-len(archive_format)]
        break
else:
    format_string = ', '.join(format.strip('.') for format,_,_ in archive_formats)
    print('Unsupported archive format. Supported formats: {}'.format(format_string)
    sys.exit(1)

It's shorter and a lot easier to add more formats. You may be surprised that I'm using a tuple of tuples instead of a dictionary. This is because order may matter. For example, you might later handle bz2 even without tar. Therefore, you would want to make sure that the .tar.bz2 extension is checked before .bz2. When order matters, a dictionary becomes infeasible. I might then use collections.OrderedDict, but I'd need to create it from a bunch of tuples, anyway, so I just left it like that.

new_archive_list = []
for ...:
    new_archive_list.append(...)
new_archive_list = set(new_archive_list)

I agree that a set is probably better than a list, but why do you create a list first and then convert that to a set? It would be better to start out a set and stay like that: new_archive_list = set() for ...: new_archive_list.add(...)

exclusions = set(exclusions.split(','))

The first use of exclusions in this function is changing it. The function is assuming that this argument is an unsuitable format, so why is that the format that is used? Why not let the calling functions give it a format it understands instead of making it do the conversion?

if len(to_archive_list):

You don't need to check the length. A string is truthy when it is non-empty, so you can do simply if to_archive_list:.

def unpack(...):
    if archive_name.endswith('.tar'):
    elif archive_name.endswith('.tar.gz'):
    ...

Again, you would probably be better off using a tuple like I showed above.

Common to all

Wildcard imports

Here's what PEP 8 says:

Wildcard imports ( from import * ) should be avoided, as they make it unclear which names are present in the namespace, confusing both readers and many automated tools. There is one defensible use case for a wildcard import, which is to republish an internal interface as part of a public API (for example, overwriting a pure Python implementation of an interface with the definitions from an optional accelerator module and exactly which definitions will be overwritten isn't known in advance).

Other imports

Again, PEP 8 says it:

Imports should be grouped in the following order:

1. standard library imports
2. related third party imports
3. local application/library specific imports

You should put a blank line between each group of imports.

Un-escaped shell arguments

You're calling external programs throughout all of your fabfile modules. Most of these include some sort of variable like a path name or an environment variable. This reminds me of the famous Little Bobby Tables. I would suggest finding a way of sanitizing your input, perhaps using the shlex module.

Comments

##- ...
...
#-##

These comment lines remind me of a section in Python's "official" humor page: Python Block Delimited Notation Parsing Explained. Python sets its functions in blocks by forcing you to indent them. Adding a special-syntax comment at the beginning and end is just extra.

##- check_empty_arguments
...
##- fix_absolute_path
...
...

Those comments are also useless. The name of the function is easy to see; putting a comment there does not make it any clearer.