s3_apt_wrapper.py

#!/usr/bin/env python2
"""s3_apt_wrapper

Manage an S3-backed APT repository and its packages.

This script requires the 'deb-s3' tool to be installed and executable via $PATH
https://github.com/krobertson/deb-s3

Usage:
    s3_apt_wrapper --help
    s3_apt_wrapper [options] list
    s3_apt_wrapper [options] upload <package_path>...
    s3_apt_wrapper [options] delete <package_name> <versions>...
    s3_apt_wrapper [options] verify

Options:
    --help, -h          Display this message
    --verbose, -v       Display extra information when listing packages

Commands:
    list         list packages that are in the APT repo's manifest
    upload       upload one or more .deb packages to the APT repo
    delete       delete a pkg from the APT repo
    verify       remove missing packages from the APT repo's manifest

Arguments:
    <package_path>...   path to one or more .deb package
    <package_name>      name of the package to delete
    <versions>          one or more versions to delete
"""
from __future__ import unicode_literals

from docopt import docopt
from subprocess import check_call, CalledProcessError
import os
import sys
import boto3

# Set up some global constants
########################################

# AWS-related
BUCKET_NAME = 'bucket name'
S3_REGION = 's3 region'

# APT-related. https://wiki.debian.org/DebianRepository/Format
REPO_CODENAME = 'codename'
PKG_ARCH = 'package architecture'

# Global deb-s3 options
S3_DEB_OPTS = ['--bucket', BUCKET_NAME,
               '--codename', REPO_CODENAME,
               '--s3-region', S3_REGION]


def is_exec(program):
    """
    Determine if a given "program" is an executable in the $PATH

    Args:
        program (unicode): program name to search for

    Return:
        bool
    """
    assert isinstance(program, unicode)
    assert '/' not in program

    for pathdir in os.environ["PATH"].split(os.pathsep):
        program_path = os.path.join(pathdir.strip('"'), program)
        if (os.path.isfile(program_path) and
                os.access(program_path, os.X_OK)):
            return True

    return False


def list_pkgs(VERBOSE):
    """
    List the packages in the repo's manifest

    Args:
        VERBOSE (bool): supply extra info or not (True if --verbose is given)
    """
    assert isinstance(VERBOSE, bool)

    # Create our API resource object
    s3 = boto3.resource('s3')

    # Load the APT repo's manifest
    manifest_path = 'dists/{}/main/binary-{}/Packages'.format(
        REPO_CODENAME, PKG_ARCH)
    manifest_body = s3.Object(BUCKET_NAME, manifest_path).get()['Body'].read()

    # Which package attributes do we want to display?
    wanted_fields = ['Package', 'Version']
    if VERBOSE:
        wanted_fields.extend(['Architecture',
                              'Maintainer',
                              'Filename',
                              'SHA1'])

    # Parse the the manifest body into a useful dict
    manifest = dict()
    manifest_lines = iter(manifest_body.splitlines())
    for line in manifest_lines:
        try:
            key, value = line.split(': ')
        except ValueError:  # multiline descriptions won't split at ': '
            pass
        else:
            if key == 'Package':
                # Each key should be <pkg>_<version>
                pkg_name = value
                pkg_version = next(manifest_lines).split(': ')[1]
                pkg = pkg_name + '_' + pkg_version
                manifest[pkg] = dict()
                manifest[pkg]['Version'] = pkg_version
            manifest[pkg][key] = value

    # Print out the list of packages and their metadata
    for pkg in sorted(manifest):
        if VERBOSE:
            for field in wanted_fields:
                print(field + ': ' + manifest[pkg][field])
            print('')
        else:
            print('{} ({})'.format(manifest[pkg]['Package'],
                                   manifest[pkg]['Version']))


def upload_pkg(package_path):
    """
    Upload a package to the APT repo

    Args:
        package_path (unicode): path to the .deb file to upload
    """
    assert isinstance(package_path, unicode)

    # Build our deb-s3 command
    cmd = ['deb-s3', 'upload', '--lock', '--preserve-versions']
    cmd.extend(S3_DEB_OPTS)
    cmd.extend(['--arch', PKG_ARCH])
    cmd.append(package_path)

    # Upload the package
    print('Uploading package: {}'.format(package_path))
    try:
        check_call(cmd)
    except CalledProcessError:
        sys.exit(1)


def delete_pkg(package_name, package_versions):
    """
    Delete a package from the APT repo

    Args:
        package_name (unicode): name of the package
        package_versions (list): list of versions to delete
    """
    assert isinstance(package_name, unicode)
    assert isinstance(package_versions, list)
    assert all(isinstance(version, str) for version in package_versions)

    # Build our deb-s3 command
    cmd = ['deb-s3', 'delete']
    cmd.extend(S3_DEB_OPTS)
    cmd.extend(['--arch', PKG_ARCH])
    cmd.extend([package_name, '--versions'])
    cmd.extend(package_versions)

    # Upload the package
    try:
        check_call(cmd)
    except CalledProcessError:
        sys.exit(1)
    else:
        # TODO: provide user-option to delete the package file also (separate
        # from the deb-s3 command).
        print("NOTE: This does NOT delete the .deb files!\nIt only removes " +
              "the reference to it in the 'Packages' file. You must delete " +
              "the package files manually, if you wish to remove them.")


def verify_pkgs():
    """
    Verify that the repo manifest and package files are in-sync, fixing any
    manifest issues if they exist
    """
    # Build our deb-s3 command
    cmd = ['deb-s3', 'verify', '--fix-manifests']
    cmd.extend(S3_DEB_OPTS)

    # Upload the package
    try:
        check_call(cmd)
    except CalledProcessError:
        sys.exit(1)


def main():
    """Main program"""

    # Init docopt arguments
    args = docopt(__doc__, help=True)

    # Check that our required binaries exist
    if not is_exec('deb-s3'):
        print('ERROR: could not find deb-s3 binary!')
        print('You can install it by running `gem install deb-s3`')
        os.exit(1)

    if args['list']:
        list_pkgs(args['--verbose'])

    if args['upload']:
        for package in args['<package_path>']:
            upload_pkg(unicode(package))

    if args['delete']:
        delete_pkg(unicode(args['<package_name>']),
                   args['<versions>'])

    if args['verify']:
        verify_pkgs()


if __name__ == "__main__":
    main()