debug_routes.py

# Copyright 2017 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Routes for inspecting CL-exonerator's state and debugging it."""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import httplib
import itertools

from chromite.lib import cidb
from chromite.lib import clactions
from chromite.lib import parallel
import flask
import flask_httpauth
import jsonschema
from werkzeug import security

import config
from exonerator import cidblib
from exonerator import gerrit_cq
from exonerator import innocent_cls_cq


app = flask.Blueprint('debug_routes', __name__)
auth = flask_httpauth.HTTPBasicAuth()
config = config.Config()
_EXONERATE_SCHEMA = {
    'type': 'object',
    'properties': {
        'cl': {'type': 'number'},
        'patch': {'type': 'number'},
        'build_id': {'type': 'number'},
        'internal': {'type': 'boolean'},
    },
    'required': ['cl', 'patch', 'build_id', 'internal']
}


@auth.verify_password
def VerifyPassword(username, password):
  """Checks if a username / password is valid.

  Args:
    username: The username to check.
    password: The given password.
  """
  # Unknown user, or the password was not set.
  if username not in config.users or config.users[username] is None:
    # Note: shortcircuiting here allows attackers to find out which usernames
    # are valid by measuring timings; in this case, that's irrelevant since the
    # set of users is public.
    return False
  return security.check_password_hash(config.users[username], password)


@app.route('/exonerate', methods=['POST'])
@auth.login_required
def Exonerate():
  """Exonerates a given cl"""
  body = flask.request.get_json(force=True)
  _ValidateSchema(_EXONERATE_SCHEMA, body)

  change = clactions.GerritPatchTuple(
      body['cl'], body['patch'], body['internal'])

  conn = cidb.CIDBConnection(
      config.cidb_cred_dir, for_service=config.is_service)
  exonerated = gerrit_cq.MaybeExonerate(change, body['build_id'])
  if exonerated:
    cidblib.InsertExoneratedCLActions(conn, [change], 'cq')
  return flask.jsonify({'exonerated': exonerated})


def _ValidateSchema(schema, data):
  """Validates that |data| follows a JSON |schema|, aborting if not."""
  try:
    jsonschema.validate(data, schema)
  except jsonschema.ValidationError:
    flask.abort(httplib.BAD_REQUEST)


@app.route('/cls')
@auth.login_required
def CLs():
  """A route for inspecting CLs to be examined by exonerator."""
  valid_cl_types = {
      'pending': PendingCLs,
      'ready': ReadyCLs,
  }
  cl_type = flask.request.args.get('type')
  if cl_type not in valid_cl_types:
    flask.abort(httplib.BAD_REQUEST)

  conn = cidb.CIDBConnection(
      config.cidb_cred_dir, for_service=config.is_service)
  innocent_batches = innocent_cls_cq.NewInnocentCLs(conn, checkpoint=False)
  innocents = itertools.chain.from_iterable(innocent_batches)

  if 'limit' in flask.request.args:
    limit = flask.request.args.get('limit', type=int)
    innocents = itertools.islice(innocents, limit)

  return flask.jsonify(cls=valid_cl_types[cl_type](innocents))


def PendingCLs(innocents):
  """Pending innocent CLs.

  Args:
    innocents: A sequence of innocent GerritPatchTuples

  Returns:
    The list of patches converted to human-readable strings.
  """
  return [str(c) for c, _ in innocents]


def ReadyCLs(innocents):
  """Filters pending CLs which can be exonerated.

  See `exonerator.gerrit.CanBeMarkedReady`.

  Args:
    innocents: A sequence of innocent GerritPatchTuples

  Returns:
    The list of patches which are ready, converted to human-readable strings.
  """
  changes = [c for c, _ in innocents]

  ready_states = parallel.RunTasksInProcessPool(
      gerrit_cq.CanBeMarkedReady,
      [[c] for c in changes])

  return [
      str(c)
      for c, ready in zip(changes, ready_states)
      if ready
  ]