~brian-murray/ubuntu-archive-tools/pup-raring

#!/usr/bin/python

# -*- coding: utf-8 -*-

# Copyright (C) 2008, 2009, 2010, 2011, 2012  Canonical Ltd.

# Copyright (C) 2010  Stéphane Graber <stgraber@stgraber.org>

# Copyright (C) 2010  Michael Bienia <geser@ubuntu.com>

# Copyright (C) 2011  Iain Lane <laney@ubuntu.com>

# Copyright (C) 2011  Soren Hansen <soren@linux2go.dk>

# Copyright (C) 2012  Stefano Rivera <stefanor@ubuntu.com>

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; version 3 of the License.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""Edit uploader permissions for the Ubuntu distro in Launchpad."""

from __future__ import print_function

from optparse import OptionParser

import sys

import launchpadlib.errors

from launchpadlib.launchpad import Launchpad

if sys.version < '3':

    input = raw_input

CONSUMER_KEY = "edit-acl"

def print_perms(perms, series=None):

    for perm in perms:

        if (series is not None and perm.distro_series_name is not None and

                series != perm.distro_series_name):

            continue

        desc = []

        desc.append("archive '%s'" % perm.archive.name)

        if perm.component_name:

            desc.append("component '%s'" % perm.component_name)

            if series:

                desc[-1] += ' in %s' % series

        if perm.package_set_name:

            desc.append("package set '%s' in %s" % (perm.package_set_name,

                                                    perm.distro_series_name))

        if perm.source_package_name:

            desc.append("source package '%s'" % perm.source_package_name)

        if perm.pocket:

            desc.append("pocket '%s'" % perm.pocket)

            if perm.distro_series_name is not None:

                desc[-1] += " in %s" % perm.distro_series_name

        print("%s for %s: %s" % (perm.permission, perm.person.name,

                                 ', '.join(desc)))

def multiline_input(prompt):

    print(prompt)

    print("End with a line containing only a full-stop '.'")

    buf = []

    while True:

        line = input()

        if line == '.':

            return '\n'.join(buf)

        buf.append(line)

def get_archive(options, launchpad):

    if options.archive is None:

        return options.distro.main_archive

    else:

        if '/' in options.archive:

            owner, ppa_name = options.archive.split('/')

            return launchpad.people[owner].getPPAByName(name=ppa_name)

        for archive in options.distro.archives:

            if archive.name == options.archive:

                return archive

    raise AssertionError("No such archive in Ubuntu: %s" % options.archive)

def get_source_components(options, launchpad, archive, source):

    try:

        from debian import debian_support

    except ImportError:

        from debian_bundle import debian_support

    args = {}

    if options.series:

        args['distro_series'] = options.distro.getSeries(

            name_or_version=options.series)

    newest = {}

    for spph in archive.getPublishedSources(

            source_name=source, exact_match=True, status='Published', **args):

        if not spph.distro_series.active:

            continue

        new_version = debian_support.Version(spph.source_package_version)

        if (spph.distro_series.name not in newest or

                new_version > newest[spph.distro_series.name][0]):

            newest[spph.distro_series.name] = (new_version,

                                               spph.component_name)

    for series in sorted(newest, key=lambda s: newest[s][0]):

        yield series, newest[series][1]

permission_names = dict(upload='Archive Upload Rights',

                        admin='Queue Administration Rights')

def do_query(options):

    """Query existing permissions and show on stdout."""

    if options.archive is None:

        archives = options.distro.archives

    else:

        archives = [get_archive(options, launchpad)]

    if options.person:

        for person in options.person:

            if '@' in person:

                lp_person = launchpad.people.getByEmail(email=person)

            else:

                try:

                    lp_person = launchpad.people[person]

                except KeyError:

                    print("Person '%s' doesn't exist." % person)

                    sys.exit(1)

            perms = []

            for archive in archives:

                perms.extend(archive.getPermissionsForPerson(

                    person=lp_person))

            if options.acl_type:

                perm_name = permission_names[options.acl_type]

                perms = [p for p in perms if p.permission == perm_name]

            print("== All rights for %s ==" % lp_person.name)

            print_perms(perms, options.series)

    if options.component:

        perms = []

        if not options.acl_type or options.acl_type == 'upload':

            for archive in archives:

                perms.extend(archive.getUploadersForComponent(

                    component_name=options.component))

        if not options.acl_type or options.acl_type == 'admin':

            for archive in archives:

                perms.extend(archive.getQueueAdminsForComponent(

                    component_name=options.component))

        print("== All rights for component '%s' ==" % options.component)

        print_perms(perms, options.series)

    if options.packageset:

        lp_series = options.distro.getSeries(name_or_version=options.series)

        for packageset in options.packageset:

            lp_set = launchpad.packagesets.getByName(name=packageset,

                                                     distroseries=lp_series)

            perms = []

            for archive in archives:

                perms.extend(archive.getUploadersForPackageset(

                    packageset=lp_set))

            print(("== All uploaders for package set '%s' in '%s' "

                   "(owned by '%s') ==" %

                  (packageset, options.series, lp_set.owner.display_name)))

            print_perms(perms, options.series)

            sources = sorted(lp_set.getSourcesIncluded(direct_inclusion=True))

            if sources:

                print()

                print("== All source packages in package set '%s' "

                      "in '%s' ==" % (packageset, options.series))

                for source in sources:

                    print(source)

            child_sets = list(lp_set.setsIncluded(direct_inclusion=True))

            if child_sets:

                print()

                print("== All package sets in package set '%s' in '%s' ==" %

                      (packageset, options.series))

                for child_set in child_sets:

                    print(child_set.name)

    if options.source:

        for source in options.source:

            perms = []

            perms_set = []

            for archive in archives:

                perms.extend(archive.getUploadersForPackage(

                    source_package_name=source))

                perms_set.extend(archive.getPackagesetsForSource(

                    sourcepackagename=source))

            print("== All uploaders for package '%s' ==" % source)

            print_perms(perms, options.series)

            print_perms(perms_set, options.series)

            for archive in archives:

                for series, component in get_source_components(

                        options, launchpad, archive, source):

                    perms_component = archive.getUploadersForComponent(

                        component_name=component)

                    print_perms(perms_component, series=series)

    if options.pocket:

        perms = []

        if not options.acl_type or options.acl_type == 'upload':

            for archive in archives:

                perms.extend(archive.getUploadersForPocket(

                    pocket=options.pocket))

        if not options.acl_type or options.acl_type == 'admin':

            for archive in archives:

                perms.extend(archive.getQueueAdminsForPocket(

                    pocket=options.pocket))

        print("== All rights for pocket '%s' ==" % options.pocket)

        print_perms(perms, options.series)

def validate_add_delete_options(options, requires_person=True):

    if options.packageset and options.source:

        # Special options to manage package sets, bodged into this tool

        # since they aren't entirely inconvenient here.

        if options.component or options.person:

            print("-P <packageset> -s <source> cannot be used with a "

                  "component or person as well")

            return False

        return True

    if requires_person and not options.person:

        print("You must specify at least one person to (de-)authorise.")

        return False

    count = 0

    if options.component:

        count += 1

    if options.packageset:

        count += 1

    if options.source:

        count += 1

    if options.pocket:

        count += 1

    if count > 1:

        print("You can only specify one of package set, source, component, "

              "or pocket")

        return False

    if count == 0:

        print("You must specify one of package set, source, component, or "

              "pocket")

        return False

    return True

def do_add(options):

    """Add a new permission."""

    if not validate_add_delete_options(options):

        return False

    archive = get_archive(options, launchpad)

    if options.packageset and options.source:

        lp_series = options.distro.getSeries(name_or_version=options.series)

        for packageset in options.packageset:

            lp_set = launchpad.packagesets.getByName(name=packageset,

                                                     distroseries=lp_series)

            lp_set.addSources(names=options.source)

            print("Added:")

            for source in options.source:

                print(source)

        return

    people = [launchpad.people[person] for person in options.person]

    if options.source:

        for source in options.source:

            for person in people:

                perm = archive.newPackageUploader(

                    person=person, source_package_name=source)

                print("Added:")

                print_perms([perm])

        return

    if options.packageset:

        lp_series = options.distro.getSeries(name_or_version=options.series)

        for packageset in options.packageset:

            lp_set = launchpad.packagesets.getByName(name=packageset,

                                                     distroseries=lp_series)

            for person in people:

                perm = archive.newPackagesetUploader(

                    person=person, packageset=lp_set)

                print("Added:")

                print_perms([perm])

        return

    if options.component:

        for person in people:

            if not options.acl_type or options.acl_type == 'upload':

                perm = archive.newComponentUploader(

                    person=person, component_name=options.component)

            else:

                perm = archive.newQueueAdmin(

                    person=person, component_name=options.component)

            print("Added:")

            print_perms([perm])

        return

    if options.pocket:

        admin_kwargs = {}

        if options.series:

            admin_kwargs["distroseries"] = options.distro.getSeries(

                name_or_version=options.series)

        for person in people:

            if not options.acl_type or options.acl_type == 'upload':

                perm = archive.newPocketUploader(

                    person=person, pocket=options.pocket)

            else:

                perm = archive.newPocketQueueAdmin(

                    person=person, pocket=options.pocket, **admin_kwargs)

            print("Added:")

            print_perms([perm])

        return

def do_delete(options):

    """Delete a permission."""

    # We kind of hacked packageset management into here.

    # Deleting packagesets doesn't require a person...

    requires_person = not (options.packageset and not options.source)

    if not validate_add_delete_options(options, requires_person):

        return False

    archive = get_archive(options, launchpad)

    if options.packageset and options.source:

        lp_series = options.distro.getSeries(name_or_version=options.series)

        for packageset in options.packageset:

            lp_set = launchpad.packagesets.getByName(name=packageset,

                                                     distroseries=lp_series)

            lp_set.removeSources(names=options.source)

            print("Deleted:")

            for source in options.source:

                print(source)

        return

    if options.packageset and not options.person:

        lp_series = options.distro.getSeries(name_or_version=options.series)

        for packageset in options.packageset:

            lp_set = launchpad.packagesets.getByName(name=packageset,

                                                     distroseries=lp_series)

            uploaders = archive.getUploadersForPackageset(

                direct_permissions=True, packageset=lp_set)

            if len(uploaders) > 0:

                print("Cannot delete packageset with defined uploaders")

                print("Current uploaders:")

                for permission in uploaders:

                    print(" %s" % permission.person.name)

                continue

            print("Confirm removal of packageset '%s'" % lp_set.name)

            print("Description:")

            print("  " + lp_set.description.replace("\n", "\n  "))

            print("Containing Sources:")

            for member in lp_set.getSourcesIncluded():

                print("  %s" % member)

            print("Containing packagesets:")

            for member in lp_set.setsIncluded():

                print("  %s" % member.name)

            ack = input("Remove? (y/N): ")

            if ack.lower() == 'y':

                lp_set.lp_delete()

                print("Deleted %s/%s" % (lp_set.name, lp_series.name))

        return

    lp_people = [launchpad.people[person] for person in options.person]

    if options.source:

        for source in options.source:

            for lp_person in lp_people:

                try:

                    archive.deletePackageUploader(

                        person=lp_person, source_package_name=source)

                    print("Deleted %s/%s" % (lp_person.name, source))

                except Exception:

                    print("Failed to delete %s/%s" % (lp_person.name, source))

        return

    if options.packageset:

        lp_series = options.distro.getSeries(name_or_version=options.series)

        for packageset in options.packageset:

            lp_set = launchpad.packagesets.getByName(name=packageset,

                                                     distroseries=lp_series)

            for lp_person in lp_people:

                archive.deletePackagesetUploader(person=lp_person,

                                                 packageset=lp_set)

                print("Deleted %s/%s/%s" % (lp_person.name, packageset,

                                            options.series))

        return

    if options.component:

        for lp_person in lp_people:

            if not options.acl_type or options.acl_type == 'upload':

                archive.deleteComponentUploader(

                    person=lp_person, component_name=options.component)

                print("Deleted %s/%s" % (lp_person.name, options.component))

            else:

                archive.deleteQueueAdmin(

                    person=lp_person, component_name=options.component)

                print("Deleted %s/%s (admin)" % (lp_person.name,

                                                 options.component))

        return

    if options.pocket:

        admin_kwargs = {}

        if options.series:

            admin_kwargs["distroseries"] = options.distro.getSeries(

                name_or_version=options.series)

        for lp_person in lp_people:

            if not options.acl_type or options.acl_type == 'upload':

                archive.deletePocketUploader(

                    person=lp_person, pocket=options.pocket)

                print("Deleted %s/%s" % (lp_person.name, options.pocket))

            else:

                archive.deletePocketQueueAdmin(

                    person=lp_person, pocket=options.pocket, **admin_kwargs)

                if options.series:

                    print("Deleted %s/%s/%s (admin)" %

                          (lp_person.name, options.pocket, options.series))

                else:

                    print("Deleted %s/%s (admin)" %

                          (lp_person.name, options.pocket))

        return

def do_create(options):

    if not options.packageset:

        print("You can only create a package set, not something else.")

        return False

    if not options.person or len(options.person) != 1:

        print("You must specify exactly one person to own the new package "

              "set.")

        return False

    if options.series:

        distro_series = options.distro.getSeries(

            name_or_version=options.series)

    else:

        distro_series = options.distro.getDevelopmentSeries()[0]

    lp_person = launchpad.people[options.person[0]]

    for packageset in options.packageset:

        try:

            if launchpad.packagesets.getByName(

                    name=packageset, distroseries=distro_series):

                print("Package set %s already exists" % packageset)

                continue

        except (TypeError, launchpadlib.errors.HTTPError):

            pass

        desc = multiline_input("Description for new package set %s:"

                               % packageset)

        ps = launchpad.packagesets.new(

            name=packageset, description=desc, distroseries=distro_series,

            owner=lp_person)

        print(ps)

def do_modify(options):

    if not options.packageset:

        print("You can only modify a package set, not something else.")

        return False

    if options.person and len(options.person) > 1:

        print("You can only specify one person as the new packageset owner.")

        return False

    if options.series:

        distro_series = options.distro.getSeries(

            name_or_version=options.series)

    else:

        distro_series = options.distro.getDevelopmentSeries()[0]

    lp_person = None

    if options.person:

        lp_person = launchpad.people[options.person[0]]

    for packageset in options.packageset:

        lp_set = launchpad.packagesets.getByName(

            name=packageset, distroseries=distro_series)

        if lp_person:

            print("Making %s the owner of %s/%s"

                  % (lp_person.name, lp_set.name, distro_series.name))

            lp_set.owner = lp_person

            lp_set.lp_save()

            continue

        print("Current description of %s:" % lp_set.name)

        print("  " + lp_set.description.replace("\n", "\n  "))

        desc = multiline_input("New description [blank=leave unmodified]:")

        if desc:

            print("Modifying description of %s/%s"

                  % (lp_set.name, distro_series.name))

            lp_set.description = desc

            lp_set.lp_save()

            continue

        rename = input("Rename %s to? [blank=don't rename]: " % lp_set.name)

        if rename:

            print("Renaming %s/%s -> %s"

                  % (lp_set.name, distro_series.name, rename))

            lp_set.name = rename

            lp_set.lp_save()

            continue

def do_copy(options):

    if options.archive is None:

        archives = options.distro.archives

    else:

        archives = [get_archive(options, launchpad)]

    if not options.packageset:

        print("You can only copy a package set, not something else.")

        return False

    if options.series:

        distro_series = options.distro.getSeries(

            name_or_version=options.series)

    else:

        distro_series = options.distro.getDevelopmentSeries()[0]

    dst = input("Name of the destination series: ")

    dst_series = options.distro.getSeries(name_or_version=dst)

    for packageset in options.packageset:

        src_pkgset = launchpad.packagesets.getByName(

            name=packageset, distroseries=distro_series)

        if not src_pkgset:

            print("Package set %s doesn't exist" % packageset)

        ps = launchpad.packagesets.new(

            name=packageset, description=src_pkgset.description,

            distroseries=dst_series, owner=src_pkgset.owner_link,

            related_set=src_pkgset)

        print(ps)

        ps.addSources(names=src_pkgset.getSourcesIncluded())

        perms = []

        for archive in archives:

            perms.extend(archive.getUploadersForPackageset(

                packageset=src_pkgset))

        for perm in perms:

            perm.archive.newPackagesetUploader(

                person=perm.person_link, packageset=ps)

def do_check(options):

    """Check if a person can upload a package."""

    if not options.person:

        print("A person needs to be specified to check.")

        return False

    if not options.source:

        print("A source package needs to be specified to check.")

        return False

    archive = get_archive(options, launchpad)

    people = [launchpad.people[person] for person in options.person]

    if options.series:

        distro_series = options.distro.getSeries(

            name_or_version=options.series)

    else:

        distro_series = options.distro.getDevelopmentSeries()[0]

    if options.pocket:

        pocket = options.pocket

    else:

        pocket = 'Release'

    for person in people:

        for srcpkg in options.source:

            try:

                spph = archive.getPublishedSources(

                    distro_series=distro_series,

                    exact_match=True,