blob: 4fa5bbacc94a0384f8a63d51910d38b9e53b9b29 [file] [log] [blame]
# 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()
'type': 'object',
'properties': {
'cl': {'type': 'number'},
'patch': {'type': 'number'},
'build_id': {'type': 'number'},
'internal': {'type': 'boolean'},
'required': ['cl', 'patch', 'build_id', 'internal']
def VerifyPassword(username, password):
"""Checks if a username / password is valid.
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'])
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."""
jsonschema.validate(data, schema)
except jsonschema.ValidationError:
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:
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.
innocents: A sequence of innocent GerritPatchTuples
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`.
innocents: A sequence of innocent GerritPatchTuples
The list of patches which are ready, converted to human-readable strings.
changes = [c for c, _ in innocents]
ready_states = parallel.RunTasksInProcessPool(
[[c] for c in changes])
return [
for c, ready in zip(changes, ready_states)
if ready