quay/workers/securityworker.py

import logging
import logging.config

import requests
import features
import time

from peewee import fn
from collections import defaultdict

from app import app, config_provider, storage, secscan_api
from endpoints.notificationhelper import spawn_notification
from workers.worker import Worker
from data import model
from data.database import (Image, UseThenDisconnect, ExternalNotificationEvent)
from data.model.tag import filter_tags_have_repository_event, get_tags_for_image
from data.model.image import set_secscan_status, get_image_with_storage_and_parent_base
from data.model.storage import get_storage_locations
from util.secscan.api import SecurityConfigValidator
from util.migrate.allocator import yield_random_entries

BATCH_SIZE = 50
INDEXING_INTERVAL = 30
API_METHOD_INSERT = '/v1/layers'
API_METHOD_GET_WITH_VULNERABILITIES = '/v1/layers/%s?vulnerabilities'

logger = logging.getLogger(__name__)

class SecurityWorker(Worker):
  def __init__(self):
    super(SecurityWorker, self).__init__()
    validator = SecurityConfigValidator(app, config_provider)
    if validator.valid():
      secscan_config = app.config.get('SECURITY_SCANNER')
      self._api = secscan_config['ENDPOINT']
      self._target_version = secscan_config['ENGINE_VERSION_TARGET']
      self._default_storage_locations = app.config['DISTRIBUTED_STORAGE_PREFERENCE']
      self._cert = validator.cert()
      self._keys = validator.keypair()

      self.add_operation(self._index_images, INDEXING_INTERVAL)
    else:
      logger.warning('Failed to validate security scan configuration')

  def _new_request(self, image):
    """ Create the request body to submit the given image for analysis. """
    url = self._get_image_url(image)
    if url is None:
      return None

    request = {
      'Layer': {
        'Name': '%s.%s' % (image.docker_image_id, image.storage.uuid),
        'Path': url,
        'Format': 'Docker'
      }
    }

    if image.parent.docker_image_id and image.parent.storage.uuid:
      request['Layer']['ParentName'] = '%s.%s' % (image.parent.docker_image_id,
                                                  image.parent.storage.uuid)

    return request

  def _get_image_url(self, image):
    """ Gets the download URL for an image and if the storage doesn't exist,
        marks the image as unindexed. """
    path = model.storage.get_layer_path(image.storage)
    locations = self._default_storage_locations

    if not storage.exists(locations, path):
      locations = get_storage_locations(image.storage.uuid)

      if not locations or not storage.exists(locations, path):
        logger.warning('Could not find a valid location to download layer %s.%s',
                       image.docker_image_id, image.storage.uuid)
        set_secscan_status(image, False, self._target_version)
        return None

    uri = storage.get_direct_download_url(locations, path)
    if uri is None:
      # Handle local storage
      local_storage_enabled = False
      for storage_type, _ in app.config.get('DISTRIBUTED_STORAGE_CONFIG', {}).values():
        if storage_type == 'LocalStorage':
          local_storage_enabled = True

      if local_storage_enabled:
        uri = path
      else:
        logger.warning('Could not get image URL and local storage was not enabled')
        return None

    return uri

  def _index_images(self):
    def batch_query():
      base_query = get_image_with_storage_and_parent_base()
      return base_query.where(Image.security_indexed_engine < self._target_version)

    min_id = (Image
              .select(fn.Min(Image.id))
              .where(Image.security_indexed_engine < self._target_version)
              .scalar())
    max_id = Image.select(fn.Max(Image.id)).scalar()

    with UseThenDisconnect(app.config):
      for candidate, abt in yield_random_entries(batch_query, Image.id, BATCH_SIZE, max_id, min_id):
        _, continue_batch = self._analyze_recursively(candidate)
        if not continue_batch:
          logger.info('Another worker pre-empted us for layer: %s', candidate.id)
          abt.set()

  def _analyze_recursively(self, layer):
    """ Analyzes a layer and all its parents """
    if layer.parent_id and layer.parent.security_indexed_engine < self._target_version:
      # The image has a parent that is not analyzed yet with this engine.
      # Get the parent to get it's own parent and recurse.
      try:
        base_query = get_image_with_storage_and_parent_base()
        parent_layer = base_query.where(Image.id == layer.parent_id).get()
      except Image.DoesNotExist:
        logger.warning("Image %s has Image %s as parent but doesn't exist.", layer.id,
                       layer.parent_id)

        return False, set_secscan_status(layer, False, self._target_version)

      cont, _ = self._analyze_recursively(parent_layer)
      if not cont:
        # The analysis failed for some reason and did not mark the layer as failed,
        # thus we should not try to analyze the children of that layer.
        # Interrupt the recursive analysis and return as no-one pre-empted us.
        return False, True

    # Now we know all parents are analyzed.
    return self._analyze(layer)

  def _analyze(self, layer):
    """ Analyzes a single layer.
        Return two bools, the first one tells us if we should evaluate its children, the second
        one is set to False when another worker pre-empted the candidate's analysis for us. """

    # If the parent couldn't be analyzed with the target version or higher, we can't analyze
    # this image. Mark it as failed with the current target version.
    if (layer.parent_id and not layer.parent.security_indexed and
        layer.parent.security_indexed_engine >= self._target_version):
      return True, set_secscan_status(layer, False, self._target_version)

    request = self._new_request(layer)
    if request is None:
      return False, True

    # Analyze the image.
    try:
      logger.info('Analyzing layer %s', request['Layer']['Name'])
      # Using invalid certificates doesn't return proper errors because of
      # https://github.com/shazow/urllib3/issues/556
      http_response = requests.post(self._api + API_METHOD_INSERT, json=request,
                                    cert=self._keys, verify=self._cert)
      json_response = http_response.json()
    except (requests.exceptions.RequestException, ValueError):
      logger.exception('An exception occurred when analyzing layer %s', request['Layer']['Name'])
      return False, True

    # Handle any errors from the security scanner.
    if http_response.status_code != 201:
      message = json_response.get('Error').get('Message', '')
      logger.warning('A warning event occurred when analyzing layer %s (status code %s): %s',
                     request['Layer']['Name'], http_response.status_code, message)

      # 422 means that the layer could not be analyzed:
      # - the layer could not be extracted (manifest?)
      # - the layer operating system / package manager is unsupported
      # Set the layer as failed.
      if http_response.status_code == 422:
        return True, set_secscan_status(layer, False, self._target_version)
      else:
        return False, True

    # Verify that the version matches.
    api_version = json_response['Layer']['IndexedByVersion']
    if api_version < self._target_version:
      logger.warning('An engine runs on version %d but the target version is %d', api_version,
                     self._target_version)

    # Mark the image as analyzed.
    logger.info('Analyzed layer %s successfully', request['Layer']['Name'])
    set_status = set_secscan_status(layer, True, api_version)

    # If we are the one who've done the job successfully first, get the vulnerabilities and
    # send notifications to the repos that have a tag on that layer.
    # TODO(josephschorr): Adapt this depending on the new notification format we adopt.
    # if set_status:
    #   # Get the tags of the layer we analyzed.
    #   repository_map = defaultdict(list)
    #   event = ExternalNotificationEvent.get(name='vulnerability_found')
    #   matching = list(filter_tags_have_repository_event(get_tags_for_image(layer.id), event))
    #
    #   for tag in matching:
    #     repository_map[tag.repository_id].append(tag)
    #
    #   # If there is at least one tag,
    #   # Lookup the vulnerabilities for the image, now that it is analyzed.
    #   if len(repository_map) > 0:
    #     logger.debug('Loading vulnerabilities for layer %s', layer.id)
    #     sec_data = self._get_vulnerabilities(layer)
    #
    #     if sec_data is not None:
    #       # Dispatch events for any detected vulnerabilities
    #       logger.debug('Got vulnerabilities for layer %s: %s', layer.id, sec_data)
    #
    #       for repository_id in repository_map:
    #         tags = repository_map[repository_id]
    #
    #         for vuln in sec_data['Vulnerabilities']:
    #           event_data = {
    #             'tags': [tag.name for tag in tags],
    #             'vulnerability': {
    #               'id': vuln['Name'],
    #               'description': vuln['Description'],
    #               'link': vuln['Link'],
    #               'priority': vuln['Priority'],
    #             },
    #           }
    #
    #           spawn_notification(tags[0].repository, 'vulnerability_found', event_data)

    return True, set_status

  def _get_vulnerabilities(self, layer):
    """ Returns the vulnerabilities detected (if any) or None on error. """
    try:
      response = secscan_api.call(self._api + API_METHOD_GET_WITH_VULNERABILITIES, None,
                                  '%s.%s' % (layer.docker_image_id, layer.storage.uuid))

      logger.debug('Got response %s for vulnerabilities for layer %s',
                   response.status_code, layer.id)

      if response.status_code == 404:
        return None
    except (requests.exceptions.RequestException, ValueError):
      logger.exception('Failed to get vulnerability response for %s', layer.id)
      return None

    return response.json()


if __name__ == '__main__':
  if not features.SECURITY_SCANNER:
    logger.debug('Security scanner disabled; skipping SecurityWorker')
    while True:
      time.sleep(100000)

  logging.config.fileConfig('conf/logging_debug.conf', disable_existing_loggers=False)
  worker = SecurityWorker()
  worker.start()