import logging import logging.config import requests import features import time from endpoints.notificationhelper import spawn_notification from collections import defaultdict from app import app, config_provider, storage, secscan_api from workers.worker import Worker from data import model from data.model.tag import filter_tags_have_repository_event, get_tags_for_image from data.model.image import get_secscan_candidates, set_secscan_status from data.model.storage import get_storage_locations from data.database import (UseThenDisconnect, ExternalNotificationEvent) from util.secscan.api import SecurityConfigValidator logger = logging.getLogger(__name__) BATCH_SIZE = 5 INDEXING_INTERVAL = 30 API_METHOD_INSERT = '/v1/layers' API_METHOD_VERSION = '/v1/versions/engine' 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 _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 _new_request(self, image): url = self._get_image_url(image) if url is None: return None request = { 'ID': '%s.%s' % (image.docker_image_id, image.storage.uuid), 'Path': url, } if image.parent is not None: request['ParentID'] = '%s.%s' % (image.parent.docker_image_id, image.parent.storage.uuid) return request def _analyze_image(self, image): """ Analyzes an image by passing it to Clair. """ request = self._new_request(image) if request is None: return False # Analyze the image. try: logger.info('Analyzing %s', request['ID']) # Using invalid certificates doesn't return proper errors because of # https://github.com/shazow/urllib3/issues/556 httpResponse = requests.post(self._api + API_METHOD_INSERT, json=request, cert=self._keys, verify=self._cert) jsonResponse = httpResponse.json() except (requests.exceptions.RequestException, ValueError): logger.exception('An exception occurred when analyzing layer ID %s', request['ID']) return False # Handle any errors from the security scanner. if httpResponse.status_code != 201: if 'OS and/or package manager are not supported' in jsonResponse.get('Message', ''): # The current engine could not index this layer logger.warning('A warning event occurred when analyzing layer ID %s : %s', request['ID'], jsonResponse['Message']) # Hopefully, there is no version lower than the target one running set_secscan_status(image, False, self._target_version) return True else: logger.warning('Got non-201 when analyzing layer ID %s: %s', request['ID'], jsonResponse) return False # Verify that the version matches. api_version = jsonResponse['Version'] if api_version < self._target_version: logger.warning('An engine runs on version %d but the target version is %d') # Mark the image as analyzed. logger.debug('Layer %s analyzed successfully', image.id) set_secscan_status(image, True, api_version) return True def _get_vulnerabilities(self, image): """ Returns the vulnerabilities detected (if any) or None on error. """ try: response = secscan_api.call('layers/%s/vulnerabilities', None, '%s.%s' % (image.docker_image_id, image.storage.uuid)) logger.debug('Got response %s for vulnerabilities for layer %s', response.status_code, image.id) if response.status_code == 404: return None except (requests.exceptions.RequestException, ValueError): logger.exception('Failed to get vulnerability response for %s', image.id) return None return response.json() def _index_images(self): logger.debug('Started indexing') event = ExternalNotificationEvent.get(name='vulnerability_found') with UseThenDisconnect(app.config): while True: # Lookup the images to index. images = [] logger.debug('Looking up images to index') images = get_secscan_candidates(self._target_version, BATCH_SIZE) if not images: logger.debug('No more images left to analyze') return logger.debug('Found %d images to index', len(images)) for image in images: # If we couldn't analyze the parent, we can't analyze this image. if (image.parent and not image.parent.security_indexed and image.parent.security_indexed_engine >= self._target_version): set_secscan_status(image, False, self._target_version) continue # Analyze the image. analyzed = self._analyze_image(image) if not analyzed: return # Get the tags of the image we analyzed matching = list(filter_tags_have_repository_event(get_tags_for_image(image.id), event)) repository_map = defaultdict(list) 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', image.id) sec_data = self._get_vulnerabilities(image) if sec_data is None: continue if not sec_data.get('Vulnerabilities'): continue # Dispatch events for any detected vulnerabilities logger.debug('Got vulnerabilities for layer %s: %s', image.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['ID'], 'description': vuln['Description'], 'link': vuln['Link'], 'priority': vuln['Priority'], }, } spawn_notification(tags[0].repository, 'vulnerability_found', event_data) 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()