1162814734
This allows us to differentiate between images that are queued and those we can't analyze in constant time.
224 lines
8 KiB
Python
224 lines
8 KiB
Python
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()
|