quay/util/secscan/api.py

import logging
import requests

from data.database import CloseForLongOperation
from data import model
from data.model.storage import get_storage_locations

from urlparse import urljoin
from util.secscan.validator import SecurityConfigValidator

logger = logging.getLogger(__name__)

class AnalyzeLayerException(Exception):
  """ Exception raised when a layer fails to analyze due to a *client-side* issue. """

class APIRequestFailure(Exception):
  """ Exception raised when there is a failure to conduct an API request. """


_API_METHOD_INSERT = 'layers'
_API_METHOD_GET_LAYER = 'layers/%s'
_API_METHOD_MARK_NOTIFICATION_READ = 'notifications/%s'
_API_METHOD_GET_NOTIFICATION = 'notifications/%s'


class SecurityScannerAPI(object):
  """ Helper class for talking to the Security Scan service (Clair). """
  def __init__(self, config, config_provider, storage):
    self.config = config
    self.config_provider = config_provider

    self._storage = storage
    self._security_config = None

    config_validator = SecurityConfigValidator(config, config_provider)
    if not config_validator.valid():
      logger.warning('Invalid config provided to SecurityScannerAPI')
      return

    self._default_storage_locations = config['DISTRIBUTED_STORAGE_PREFERENCE']

    self._security_config = config.get('SECURITY_SCANNER')
    self._target_version = self._security_config['ENGINE_VERSION_TARGET']

    self._certificate = config_validator.cert()
    self._keys = config_validator.keypair()


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

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

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

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

      if local_storage_enabled:
        # TODO: fix to use the proper local storage path.
        uri = path
      else:
        logger.warning('Could not get image URL and local storage was not enabled')
        return None

    return uri


  def _new_analyze_request(self, image):
    """ Create the request body to submit the given image for analysis. If the image's URL cannot
        be found, returns None.
    """
    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 analyze_layer(self, layer):
    """ Posts the given layer to the security scanner for analysis, blocking until complete.
        Returns a tuple containing the analysis version (on success, None on failure) and
        whether the request should be retried.
    """
    request = self._new_analyze_request(layer)
    if not request:
      return None, False

    logger.info('Analyzing layer %s', request['Layer']['Name'])
    try:
      response = self._call('POST', _API_METHOD_INSERT, request)
      json_response = response.json()
    except requests.exceptions.Timeout:
      logger.exception('Timeout when trying to post layer data response for %s', layer.id)
      return None, True
    except requests.exceptions.ConnectionError:
      logger.exception('Connection error when trying to post layer data response for %s', layer.id)
      return None, True
    except (requests.exceptions.RequestException, ValueError):
      logger.exception('Failed to post layer data response for %s', layer.id)
      return None, False

    # Handle any errors from the security scanner.
    if 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'], response.status_code, message)

      # 400 means the layer could not be analyzed due to a bad request.
      if response.status_code == 400:
        logger.error('Bad request when calling security scanner for layer %s: %s',
                     response.status_code, json_response)
        raise AnalyzeLayerException('Bad request to security scanner')

      # 422 means that the layer could not be analyzed:
      # - the layer could not be extracted (manifest?)
      # - the layer operating system / package manager is unsupported
      return None, response.status_code != 422

    api_version = json_response['Layer']['IndexedByVersion']
    return api_version, False


  def check_layer_vulnerable(self, layer_id, cve_name):
    """ Checks to see if the layer with the given ID is vulnerable to the specified CVE. """
    layer_data = self._get_layer_data(layer_id, include_vulnerabilities=True)
    if layer_data is None or 'Layer' not in layer_data or 'Features' not in layer_data['Layer']:
      return False

    for feature in layer_data['Layer']['Features']:
      for vuln in feature.get('Vulnerabilities', []):
        if vuln['Name'] == cve_name:
          return True

    return False


  def get_notification(self, notification_name, layer_limit=10, page=None):
    """ Gets the data for a specific notification, with optional page token.
        Returns a tuple of the data (None on failure) and whether to retry.
    """
    try:
      params = {
        'limit': layer_limit
      }

      if page is not None:
        params['page'] = page

      response = self._call('GET', _API_METHOD_GET_NOTIFICATION % notification_name, params=params)
      json_response = response.json()
    except requests.exceptions.Timeout:
      logger.exception('Timeout when trying to get notification for %s', notification_name)
      return None, True
    except requests.exceptions.ConnectionError:
      logger.exception('Connection error when trying to get notification for %s', notification_name)
      return None, True
    except (requests.exceptions.RequestException, ValueError):
      logger.exception('Failed to get notification for %s', notification_name)
      return None, False

    if response.status_code != 200:
      return None, response.status_code != 404 and response.status_code != 400

    return json_response, False


  def mark_notification_read(self, notification_name):
    """ Marks a security scanner notification as read. """
    try:
      response = self._call('DELETE', _API_METHOD_MARK_NOTIFICATION_READ % notification_name)
      return response.status_code == 200
    except requests.exceptions.RequestException:
      logger.exception('Failed to mark notification as read: %s', notification_name)
      return False


  def get_layer_data(self, layer, include_features=False, include_vulnerabilities=False):
    """ Returns the layer data for the specified layer. On error, returns None. """
    layer_id = '%s.%s' % (layer.docker_image_id, layer.storage.uuid)
    return self._get_layer_data(layer_id, include_features, include_vulnerabilities)


  def _get_layer_data(self, layer_id, include_features=False, include_vulnerabilities=False):
    try:
      params = {}
      if include_features:
        params = {'features': True}

      if include_vulnerabilities:
        params = {'vulnerabilities': True}

      response = self._call('GET', _API_METHOD_GET_LAYER % layer_id, params=params)
      logger.debug('Got response %s for vulnerabilities for layer %s',
                   response.status_code, layer_id)
      json_response = response.json()
    except requests.exceptions.Timeout:
      raise APIRequestFailure('API call timed out')
    except requests.exceptions.ConnectionError:
      raise APIRequestFailure('Could not connect to security service')
    except (requests.exceptions.RequestException, ValueError):
      logger.exception('Failed to get layer data response for %s', layer_id)
      raise APIRequestFailure()

    if response.status_code == 404:
      return None

    return json_response


  def _call(self, method, relative_url, params=None, body=None):
    """ Issues an HTTP call to the sec API at the given relative URL.
        This function disconnects from the database while awaiting a response
        from the API server.
    """
    security_config = self._security_config
    if security_config is None:
      raise Exception('Cannot call unconfigured security system')

    api_url = urljoin(security_config['ENDPOINT'], '/' + security_config['API_VERSION']) + '/'
    url = urljoin(api_url, relative_url)

    client = self.config['HTTPCLIENT']
    headers = {'Connection': 'close'}
    timeout = security_config.get('API_TIMEOUT_SECONDS', 1)

    with CloseForLongOperation(self.config):
      if method == 'POST':
        logger.debug('POSTing security URL %s', url)
        return client.post(url, json=body, params=params, timeout=timeout, cert=self._keys,
                           verify=self._certificate, headers=headers)
      elif method == 'DELETE':
        logger.debug('DELETEing security URL %s', url)
        return client.delete(url, params=params, timeout=timeout, cert=self._keys,
                             verify=self._certificate, headers=headers)
      else:
        logger.debug('GETing security URL %s', url)
        return client.get(url, params=params, timeout=timeout, cert=self._keys,
                          verify=self._certificate, headers=headers)