tools/repo_pull/gerrit.py

#!/usr/bin/env python3

#
# Copyright (C) 2018 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

"""Gerrit Restful API client library."""

from __future__ import print_function

import argparse
import base64
import json
import os
import sys
import xml.dom.minidom

try:
    # PY3
    from urllib.error import HTTPError
    from urllib.parse import urlencode, urlparse
    from urllib.request import (
        HTTPBasicAuthHandler, Request, build_opener
    )
except ImportError:
    # PY2
    from urllib import urlencode
    from urllib2 import (
        HTTPBasicAuthHandler, HTTPError, Request, build_opener
    )
    from urlparse import urlparse

try:
    # PY3.5
    from subprocess import PIPE, run
except ImportError:
    from subprocess import CalledProcessError, PIPE, Popen

    class CompletedProcess(object):
        """Process execution result returned by subprocess.run()."""
        # pylint: disable=too-few-public-methods

        def __init__(self, args, returncode, stdout, stderr):
            self.args = args
            self.returncode = returncode
            self.stdout = stdout
            self.stderr = stderr

    def run(*args, **kwargs):
        """Run a command with subprocess.Popen() and redirect input/output."""

        check = kwargs.pop('check', False)

        try:
            stdin = kwargs.pop('input')
            assert 'stdin' not in kwargs
            kwargs['stdin'] = PIPE
        except KeyError:
            stdin = None

        proc = Popen(*args, **kwargs)
        try:
            stdout, stderr = proc.communicate(stdin)
        except:
            proc.kill()
            proc.wait()
            raise
        returncode = proc.wait()

        if check and returncode:
            raise CalledProcessError(returncode, args, stdout)
        return CompletedProcess(args, returncode, stdout, stderr)


def load_auth_credentials_from_file(cookie_file):
    """Load credentials from an opened .gitcookies file."""
    credentials = {}
    for line in cookie_file:
        if line.startswith('#HttpOnly_'):
            line = line[len('#HttpOnly_'):]

        if not line or line[0] == '#':
            continue

        row = line.split('\t')
        if len(row) != 7:
            continue

        domain = row[0]
        cookie = row[6]

        sep = cookie.find('=')
        if sep == -1:
            continue
        username = cookie[0:sep]
        password = cookie[sep + 1:]

        credentials[domain] = (username, password)
    return credentials


def load_auth_credentials(cookie_file_path):
    """Load credentials from a .gitcookies file path."""
    with open(cookie_file_path, 'r') as cookie_file:
        return load_auth_credentials_from_file(cookie_file)


def create_url_opener(cookie_file_path, domain):
    """Load username and password from .gitcookies and return a URL opener with
    an authentication handler."""

    # Load authentication credentials
    credentials = load_auth_credentials(cookie_file_path)
    username, password = credentials[domain]

    # Create URL opener with authentication handler
    auth_handler = HTTPBasicAuthHandler()
    auth_handler.add_password(domain, domain, username, password)
    return build_opener(auth_handler)


def create_url_opener_from_args(args):
    """Create URL opener from command line arguments."""

    domain = urlparse(args.gerrit).netloc

    try:
        return create_url_opener(args.gitcookies, domain)
    except KeyError:
        print('error: Cannot find the domain "{}" in "{}". '
              .format(domain, args.gitcookies), file=sys.stderr)
        print('error: Please check the Gerrit Code Review URL or follow the '
              'instructions in '
              'https://android.googlesource.com/platform/development/'
              '+/master/tools/repo_pull#installation', file=sys.stderr)
        sys.exit(1)


def _decode_xssi_json(data):
    """Trim XSSI protector and decode JSON objects.

    Returns:
        An object returned by json.loads().

    Raises:
        ValueError: If data doesn't start with a XSSI token.
        json.JSONDecodeError: If data failed to decode.
    """

    # Decode UTF-8
    data = data.decode('utf-8')

    # Trim cross site script inclusion (XSSI) protector
    if data[0:4] != ')]}\'':
        raise ValueError('unexpected responsed content: ' + data)
    data = data[4:]

    # Parse JSON objects
    return json.loads(data)


def query_change_lists(url_opener, gerrit, query_string, limits):
    """Query change lists."""
    data = [
        ('q', query_string),
        ('o', 'CURRENT_REVISION'),
        ('o', 'CURRENT_COMMIT'),
        ('n', str(limits)),
    ]
    url = gerrit + '/a/changes/?' + urlencode(data)

    response_file = url_opener.open(url)
    try:
        return _decode_xssi_json(response_file.read())
    finally:
        response_file.close()


def _make_json_post_request(url_opener, url, data, method='POST'):
    """Open an URL request and decode its response.

    Returns a 3-tuple of (code, body, json).
        code: A numerical value, the HTTP status code of the response.
        body: A bytes, the response body.
        json: An object, the parsed JSON response.
    """

    data = json.dumps(data).encode('utf-8')
    headers = {
        'Content-Type': 'application/json; charset=UTF-8',
    }

    request = Request(url, data, headers)
    request.get_method = lambda: method

    try:
        response_file = url_opener.open(request)
    except HTTPError as error:
        response_file = error

    with response_file:
        res_code = response_file.getcode()
        res_body = response_file.read()
        try:
            res_json = _decode_xssi_json(res_body)
        except ValueError:
            # The response isn't JSON if it doesn't start with a XSSI token.
            # Possibly a plain text error message or empty body.
            res_json = None
        return (res_code, res_body, res_json)


def set_review(url_opener, gerrit_url, change_id, labels, message):
    """Set review votes to a change list."""

    url = '{}/a/changes/{}/revisions/current/review'.format(
        gerrit_url, change_id)

    data = {}
    if labels:
        data['labels'] = labels
    if message:
        data['message'] = message

    return _make_json_post_request(url_opener, url, data)


def submit(url_opener, gerrit_url, change_id):
    """Submit a change list."""

    url = '{}/a/changes/{}/submit'.format(gerrit_url, change_id)

    return _make_json_post_request(url_opener, url, {})


def abandon(url_opener, gerrit_url, change_id, message):
    """Abandon a change list."""

    url = '{}/a/changes/{}/abandon'.format(gerrit_url, change_id)

    data = {}
    if message:
        data['message'] = message

    return _make_json_post_request(url_opener, url, data)


def restore(url_opener, gerrit_url, change_id):
    """Restore a change list."""

    url = '{}/a/changes/{}/restore'.format(gerrit_url, change_id)

    return _make_json_post_request(url_opener, url, {})


def set_topic(url_opener, gerrit_url, change_id, name):
    """Set the topic name."""

    url = '{}/a/changes/{}/topic'.format(gerrit_url, change_id)
    data = {'topic': name}
    return _make_json_post_request(url_opener, url, data, method='PUT')


def delete_topic(url_opener, gerrit_url, change_id):
    """Delete the topic name."""

    url = '{}/a/changes/{}/topic'.format(gerrit_url, change_id)

    return _make_json_post_request(url_opener, url, {}, method='DELETE')


def set_hashtags(url_opener, gerrit_url, change_id, add_tags=None,
                 remove_tags=None):
    """Add or remove hash tags."""

    url = '{}/a/changes/{}/hashtags'.format(gerrit_url, change_id)

    data = {}
    if add_tags:
        data['add'] = add_tags
    if remove_tags:
        data['remove'] = remove_tags

    return _make_json_post_request(url_opener, url, data)


def add_reviewers(url_opener, gerrit_url, change_id, reviewers):
    """Add reviewers."""

    url = '{}/a/changes/{}/revisions/current/review'.format(
        gerrit_url, change_id)

    data = {}
    if reviewers:
        data['reviewers'] = reviewers

    return _make_json_post_request(url_opener, url, data)


def delete_reviewer(url_opener, gerrit_url, change_id, name):
    """Delete reviewer."""

    url = '{}/a/changes/{}/reviewers/{}/delete'.format(
        gerrit_url, change_id, name)

    return _make_json_post_request(url_opener, url, {})


def get_patch(url_opener, gerrit_url, change_id, revision_id='current'):
    """Download the patch file."""

    url = '{}/a/changes/{}/revisions/{}/patch'.format(
        gerrit_url, change_id, revision_id)

    response_file = url_opener.open(url)
    try:
        return base64.b64decode(response_file.read())
    finally:
        response_file.close()

def find_gerrit_name():
    """Find the gerrit instance specified in the default remote."""
    manifest_cmd = ['repo', 'manifest']
    raw_manifest_xml = run(manifest_cmd, stdout=PIPE, check=True).stdout

    manifest_xml = xml.dom.minidom.parseString(raw_manifest_xml)
    default_remote = manifest_xml.getElementsByTagName('default')[0]
    default_remote_name = default_remote.getAttribute('remote')
    for remote in manifest_xml.getElementsByTagName('remote'):
        name = remote.getAttribute('name')
        review = remote.getAttribute('review')
        if review and name == default_remote_name:
            return review.rstrip('/')

    raise ValueError('cannot find gerrit URL from manifest')

def _parse_args():
    """Parse command line options."""
    parser = argparse.ArgumentParser()

    parser.add_argument('query', help='Change list query string')
    parser.add_argument('-g', '--gerrit', help='Gerrit review URL')

    parser.add_argument('--gitcookies',
                        default=os.path.expanduser('~/.gitcookies'),
                        help='Gerrit cookie file')
    parser.add_argument('--limits', default=1000,
                        help='Max number of change lists')

    return parser.parse_args()


def main():
    """Main function"""
    args = _parse_args()

    if not args.gerrit:
        try:
            args.gerrit = find_gerrit_name()
        # pylint: disable=bare-except
        except:
            print('gerrit instance not found, use [-g GERRIT]')
            sys.exit(1)

    # Query change lists
    url_opener = create_url_opener_from_args(args)
    change_lists = query_change_lists(
        url_opener, args.gerrit, args.query, args.limits)

    # Print the result
    json.dump(change_lists, sys.stdout, indent=4, separators=(', ', ': '))
    print()  # Print the end-of-line

if __name__ == '__main__':
    main()