quay/test/test_secscan.py

import json
import time
import unittest

from app import app, storage, notification_queue
from data import model
from data.database import Image, IMAGE_NOT_SCANNED_ENGINE_VERSION
from endpoints.notificationevent import VulnerabilityFoundEvent
from endpoints.v2 import v2_bp
from initdb import setup_database_for_testing, finished_database_for_testing
from util.secscan.api import SecurityScannerAPI, APIRequestFailure
from util.secscan.analyzer import LayerAnalyzer
from util.secscan.fake import fake_security_scanner
from util.secscan.notifier import SecurityNotificationHandler, ProcessNotificationPageResult
from workers.security_notification_worker import SecurityNotificationWorker


ADMIN_ACCESS_USER = 'devtable'
SIMPLE_REPO = 'simple'
COMPLEX_REPO = 'complex'


def process_notification_data(notification_data):
  handler = SecurityNotificationHandler(100)
  result = handler.process_notification_page_data(notification_data)
  handler.send_notifications()
  return result == ProcessNotificationPageResult.FINISHED_PROCESSING


class TestSecurityScanner(unittest.TestCase):
  def setUp(self):
    # Enable direct download in fake storage.
    storage.put_content(['local_us'], 'supports_direct_download', 'true')

    # Have fake storage say all files exist for the duration of the test.
    storage.put_content(['local_us'], 'all_files_exist', 'true')

    # Setup the database with fake storage.
    setup_database_for_testing(self)
    self.app = app.test_client()
    self.ctx = app.test_request_context()
    self.ctx.__enter__()

    self.api = SecurityScannerAPI(app, app.config, storage)

  def tearDown(self):
    storage.remove(['local_us'], 'supports_direct_download')
    storage.remove(['local_us'], 'all_files_exist')

    finished_database_for_testing(self)
    self.ctx.__exit__(True, None, None)

  def assertAnalyzed(self, layer, security_scanner, isAnalyzed, engineVersion):
    self.assertEquals(isAnalyzed, layer.security_indexed)
    self.assertEquals(engineVersion, layer.security_indexed_engine)

    if isAnalyzed:
      self.assertTrue(security_scanner.has_layer(security_scanner.layer_id(layer)))

      # Ensure all parent layers are marked as analyzed.
      parents = model.image.get_parent_images(ADMIN_ACCESS_USER, SIMPLE_REPO, layer)
      for parent in parents:
        self.assertTrue(parent.security_indexed)
        self.assertEquals(engineVersion, parent.security_indexed_engine)
        self.assertTrue(security_scanner.has_layer(security_scanner.layer_id(parent)))

  def test_get_layer(self):
    """ Test for basic retrieval of layers from the security scanner. """
    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)

    with fake_security_scanner() as security_scanner:
      # Ensure the layer doesn't exist yet.
      self.assertFalse(security_scanner.has_layer(security_scanner.layer_id(layer)))
      self.assertIsNone(self.api.get_layer_data(layer))

      # Add the layer.
      security_scanner.add_layer(security_scanner.layer_id(layer))

      # Retrieve the results.
      result = self.api.get_layer_data(layer, include_vulnerabilities=True)
      self.assertIsNotNone(result)
      self.assertEquals(result['Layer']['Name'], security_scanner.layer_id(layer))

  def test_analyze_layer_nodirectdownload_success(self):
    """ Tests analyzing a layer when direct download is disabled. """

    # Disable direct download in fake storage.
    storage.put_content(['local_us'], 'supports_direct_download', 'false')

    try:
      app.register_blueprint(v2_bp, url_prefix='/v2')
    except:
      # Already registered.
      pass

    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    self.assertFalse(layer.security_indexed)
    self.assertEquals(-1, layer.security_indexed_engine)

    # Ensure that the download is a registry+JWT download.
    uri, auth_header = self.api._get_image_url_and_auth(layer)
    self.assertIsNotNone(uri)
    self.assertIsNotNone(auth_header)

    # Ensure the download doesn't work without the header.
    rv = self.app.head(uri)
    self.assertEquals(rv.status_code, 401)

    # Ensure the download works with the header. Note we use a HEAD here, as GET causes DB
    # access which messes with the test runner's rollback.
    rv = self.app.head(uri, headers=[('authorization', auth_header)])
    self.assertEquals(rv.status_code, 200)

    # Ensure the code works when called via analyze.
    with fake_security_scanner() as security_scanner:
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, True, 1)

  def test_analyze_layer_success(self):
    """ Tests that analyzing a layer successfully marks it as analyzed. """

    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    self.assertFalse(layer.security_indexed)
    self.assertEquals(-1, layer.security_indexed_engine)

    with fake_security_scanner() as security_scanner:
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, True, 1)

  def test_analyze_layer_failure(self):
    """ Tests that failing to analyze a layer (because it 422s) marks it as analyzed but failed. """

    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    self.assertFalse(layer.security_indexed)
    self.assertEquals(-1, layer.security_indexed_engine)

    with fake_security_scanner() as security_scanner:
      security_scanner.set_fail_layer_id(security_scanner.layer_id(layer))

      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, False, 1)

  def test_analyze_layer_internal_error(self):
    """ Tests that failing to analyze a layer (because it 500s) marks it as not analyzed. """

    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    self.assertFalse(layer.security_indexed)
    self.assertEquals(-1, layer.security_indexed_engine)

    with fake_security_scanner() as security_scanner:
      security_scanner.set_internal_error_layer_id(security_scanner.layer_id(layer))

      analyzer = LayerAnalyzer(app.config, self.api)
      with self.assertRaises(APIRequestFailure) as ctx:
        analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, False, -1)

  def test_analyze_layer_error(self):
    """ Tests that failing to analyze a layer (because it 400s) marks it as analyzed but failed. """

    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    self.assertFalse(layer.security_indexed)
    self.assertEquals(-1, layer.security_indexed_engine)

    with fake_security_scanner() as security_scanner:
      # Make is so trying to analyze the parent will fail with an error.
      security_scanner.set_error_layer_id(security_scanner.layer_id(layer.parent))

      # Try to the layer and its parents, but with one request causing an error.
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      # Make sure it is marked as analyzed, but in a failed state.
      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, False, 1)

  def test_analyze_layer_missing_parent_handled(self):
    """ Tests that a missing parent causes an automatic reanalysis, which succeeds. """

    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    self.assertFalse(layer.security_indexed)
    self.assertEquals(-1, layer.security_indexed_engine)

    with fake_security_scanner() as security_scanner:
      # Analyze the layer and its parents.
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      # Make sure it was analyzed.
      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, True, 1)

      # Mark the layer as not yet scanned.
      layer.security_indexed_engine = IMAGE_NOT_SCANNED_ENGINE_VERSION
      layer.security_indexed = False
      layer.save()

      # Remove the layer's parent entirely from the security scanner.
      security_scanner.remove_layer(security_scanner.layer_id(layer.parent))

      # Analyze again, which should properly re-analyze the missing parent and this layer.
      analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, True, 1)

  def test_analyze_layer_invalid_parent(self):
    """ Tests that trying to reanalyze a parent that is invalid causes the layer to be marked
        as analyzed, but failed.
    """

    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    self.assertFalse(layer.security_indexed)
    self.assertEquals(-1, layer.security_indexed_engine)

    with fake_security_scanner() as security_scanner:
      # Analyze the layer and its parents.
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      # Make sure it was analyzed.
      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, True, 1)

      # Mark the layer as not yet scanned.
      layer.security_indexed_engine = IMAGE_NOT_SCANNED_ENGINE_VERSION
      layer.security_indexed = False
      layer.save()

      # Remove the layer's parent entirely from the security scanner.
      security_scanner.remove_layer(security_scanner.layer_id(layer.parent))

      # Make is so trying to analyze the parent will fail.
      security_scanner.set_error_layer_id(security_scanner.layer_id(layer.parent))

      # Try to analyze again, which should try to reindex the parent and fail.
      analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, False, 1)

  def test_analyze_layer_unsupported_parent(self):
    """ Tests that attempting to analyze a layer whose parent is unanalyzable, results in the layer
        being marked as analyzed, but failed.
    """

    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    self.assertFalse(layer.security_indexed)
    self.assertEquals(-1, layer.security_indexed_engine)

    with fake_security_scanner() as security_scanner:
      # Make is so trying to analyze the parent will fail.
      security_scanner.set_fail_layer_id(security_scanner.layer_id(layer.parent))

      # Attempt to the layer and its parents. This should mark the layer itself as unanalyzable.
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, False, 1)

  def test_analyze_layer_missing_storage(self):
    """ Tests trying to analyze a layer with missing storage. """

    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    self.assertFalse(layer.security_indexed)
    self.assertEquals(-1, layer.security_indexed_engine)

    # Delete the storage for the layer.
    path = model.storage.get_layer_path(layer.storage)
    locations = app.config['DISTRIBUTED_STORAGE_PREFERENCE']
    storage.remove(locations, path)
    storage.remove(locations, 'all_files_exist')

    with fake_security_scanner() as security_scanner:
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, False, 1)

  def assert_analyze_layer_notify(self, security_indexed_engine, security_indexed,
                                  expect_notification):
    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    self.assertFalse(layer.security_indexed)
    self.assertEquals(-1, layer.security_indexed_engine)

    # Ensure there are no existing events.
    self.assertIsNone(notification_queue.get())

    # Add a repo event for the layer.
    repo = model.repository.get_repository(ADMIN_ACCESS_USER, SIMPLE_REPO)
    model.notification.create_repo_notification(repo, 'vulnerability_found',
                                                'quay_notification', {}, {'level': 100})

    # Update the layer's state before analyzing.
    layer.security_indexed_engine = security_indexed_engine
    layer.security_indexed = security_indexed
    layer.save()

    with fake_security_scanner() as security_scanner:
      security_scanner.set_vulns(security_scanner.layer_id(layer), [
        {
          "Name": "CVE-2014-9471",
          "Namespace": "debian:8",
          "Description": "Some service",
          "Link": "https://security-tracker.debian.org/tracker/CVE-2014-9471",
          "Severity": "Low",
          "FixedBy": "9.23-5"
        },

        {
          "Name": "CVE-2016-7530",
          "Namespace": "debian:8",
          "Description": "Some other service",
          "Link": "https://security-tracker.debian.org/tracker/CVE-2016-7530",
          "Severity": "Unknown",
          "FixedBy": "19.343-2"
        }
      ])

      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, True, 1)

    # Ensure an event was written for the tag (if necessary).
    time.sleep(1)
    queue_item = notification_queue.get()

    if expect_notification:
      self.assertIsNotNone(queue_item)

      body = json.loads(queue_item.body)
      self.assertEquals(set(['latest', 'prod']), set(body['event_data']['tags']))
      self.assertEquals('CVE-2014-9471', body['event_data']['vulnerability']['id'])
      self.assertEquals('Low', body['event_data']['vulnerability']['priority'])
      self.assertTrue(body['event_data']['vulnerability']['has_fix'])

      self.assertEquals('CVE-2014-9471', body['event_data']['vulnerabilities'][0]['id'])
      self.assertEquals(2, len(body['event_data']['vulnerabilities']))

      # Ensure we get the correct event message out as well.
      event = VulnerabilityFoundEvent()
      msg = '1 Low and 1 more vulnerabilities were detected in repository devtable/simple in 2 tags'
      self.assertEquals(msg, event.get_summary(body['event_data'], {}))
      self.assertEquals('info', event.get_level(body['event_data'], {}))
    else:
      self.assertIsNone(queue_item)

    # Ensure its security indexed engine was updated.
    updated_layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
    self.assertEquals(updated_layer.id, layer.id)
    self.assertTrue(updated_layer.security_indexed_engine > 0)

  def test_analyze_layer_success_events(self):
    # Not previously indexed at all => Notification
    self.assert_analyze_layer_notify(IMAGE_NOT_SCANNED_ENGINE_VERSION, False, True)

  def test_analyze_layer_success_no_notification(self):
    # Previously successfully indexed => No notification
    self.assert_analyze_layer_notify(0, True, False)

  def test_analyze_layer_failed_then_success_notification(self):
    # Previously failed to index => Notification
    self.assert_analyze_layer_notify(0, False, True)

  def test_notification_new_layers_not_vulnerable(self):
    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    layer_id = '%s.%s' % (layer.docker_image_id, layer.storage.uuid)

    # Add a repo event for the layer.
    repo = model.repository.get_repository(ADMIN_ACCESS_USER, SIMPLE_REPO)
    model.notification.create_repo_notification(repo, 'vulnerability_found', 'quay_notification',
                                                {}, {'level': 100})

    # Ensure that there are no event queue items for the layer.
    self.assertIsNone(notification_queue.get())

    # Fire off the notification processing.
    with fake_security_scanner() as security_scanner:
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, True, 1)

      # Add a notification for the layer.
      notification_data = security_scanner.add_notification([layer_id], [], {}, {})

      # Process the notification.
      self.assertTrue(process_notification_data(notification_data))

      # Ensure that there are no event queue items for the layer.
      self.assertIsNone(notification_queue.get())

  def test_notification_delete(self):
    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    layer_id = '%s.%s' % (layer.docker_image_id, layer.storage.uuid)

    # Add a repo event for the layer.
    repo = model.repository.get_repository(ADMIN_ACCESS_USER, SIMPLE_REPO)
    model.notification.create_repo_notification(repo, 'vulnerability_found', 'quay_notification',
                                                {}, {'level': 100})

    # Ensure that there are no event queue items for the layer.
    self.assertIsNone(notification_queue.get())

    # Fire off the notification processing.
    with fake_security_scanner() as security_scanner:
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, True, 1)

      # Add a notification for the layer.
      notification_data = security_scanner.add_notification([layer_id], None, {}, None)

      # Process the notification.
      self.assertTrue(process_notification_data(notification_data))

      # Ensure that there are no event queue items for the layer.
      self.assertIsNone(notification_queue.get())

  def test_notification_new_layers(self):
    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    layer_id = '%s.%s' % (layer.docker_image_id, layer.storage.uuid)

    # Add a repo event for the layer.
    repo = model.repository.get_repository(ADMIN_ACCESS_USER, SIMPLE_REPO)
    model.notification.create_repo_notification(repo, 'vulnerability_found', 'quay_notification',
                                                {}, {'level': 100})

    # Ensure that there are no event queue items for the layer.
    self.assertIsNone(notification_queue.get())

    # Fire off the notification processing.
    with fake_security_scanner() as security_scanner:
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, True, 1)

      vuln_info = {
        "Name": "CVE-TEST",
        "Namespace": "debian:8",
        "Description": "Some service",
        "Link": "https://security-tracker.debian.org/tracker/CVE-2014-9471",
        "Severity": "Low",
        "FixedIn": {"Version": "9.23-5"},
      }
      security_scanner.set_vulns(layer_id, [vuln_info])

      # Add a notification for the layer.
      notification_data = security_scanner.add_notification([], [layer_id], vuln_info, vuln_info)

      # Process the notification.
      self.assertTrue(process_notification_data(notification_data))

      # Ensure an event was written for the tag.
      time.sleep(1)
      queue_item = notification_queue.get()
      self.assertIsNotNone(queue_item)

      item_body = json.loads(queue_item.body)
      self.assertEquals(sorted(['prod', 'latest']), sorted(item_body['event_data']['tags']))
      self.assertEquals('CVE-TEST', item_body['event_data']['vulnerability']['id'])
      self.assertEquals('Low', item_body['event_data']['vulnerability']['priority'])
      self.assertTrue(item_body['event_data']['vulnerability']['has_fix'])

  def test_notification_no_new_layers(self):
    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)

    # Add a repo event for the layer.
    repo = model.repository.get_repository(ADMIN_ACCESS_USER, SIMPLE_REPO)
    model.notification.create_repo_notification(repo, 'vulnerability_found', 'quay_notification',
                                                {}, {'level': 100})

    # Ensure that there are no event queue items for the layer.
    self.assertIsNone(notification_queue.get())

    # Fire off the notification processing.
    with fake_security_scanner() as security_scanner:
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, True, 1)

      # Add a notification for the layer.
      notification_data = security_scanner.add_notification([], [], {}, {})

      # Process the notification.
      self.assertTrue(process_notification_data(notification_data))

      # Ensure that there are no event queue items for the layer.
      self.assertIsNone(notification_queue.get())

  def test_notification_no_new_layers_increased_severity(self):
    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    layer_id = '%s.%s' % (layer.docker_image_id, layer.storage.uuid)

    # Add a repo event for the layer.
    repo = model.repository.get_repository(ADMIN_ACCESS_USER, SIMPLE_REPO)
    notification = model.notification.create_repo_notification(repo, 'vulnerability_found',
                                                               'quay_notification', {},
                                                               {'level': 100})

    # Ensure that there are no event queue items for the layer.
    self.assertIsNone(notification_queue.get())

    # Fire off the notification processing.
    with fake_security_scanner() as security_scanner:
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, True, 1)

      old_vuln_info = {
        "Name": "CVE-TEST",
        "Namespace": "debian:8",
        "Description": "Some service",
        "Link": "https://security-tracker.debian.org/tracker/CVE-2014-9471",
        "Severity": "Low",
      }

      new_vuln_info = {
        "Name": "CVE-TEST",
        "Namespace": "debian:8",
        "Description": "Some service",
        "Link": "https://security-tracker.debian.org/tracker/CVE-2014-9471",
        "Severity": "Critical",
        "FixedIn": {'Version': "9.23-5"},
      }

      security_scanner.set_vulns(layer_id, [new_vuln_info])

      # Add a notification for the layer.
      notification_data = security_scanner.add_notification([layer_id], [layer_id],
                                                            old_vuln_info, new_vuln_info)

      # Process the notification.
      self.assertTrue(process_notification_data(notification_data))

      # Ensure an event was written for the tag.
      time.sleep(1)
      queue_item = notification_queue.get()
      self.assertIsNotNone(queue_item)

      item_body = json.loads(queue_item.body)
      self.assertEquals(sorted(['prod', 'latest']), sorted(item_body['event_data']['tags']))
      self.assertEquals('CVE-TEST', item_body['event_data']['vulnerability']['id'])
      self.assertEquals('Critical', item_body['event_data']['vulnerability']['priority'])
      self.assertTrue(item_body['event_data']['vulnerability']['has_fix'])

      # Verify that an event would be raised.
      event_data = item_body['event_data']
      self.assertTrue(VulnerabilityFoundEvent().should_perform(event_data, notification))

      # Create another notification with a matching level and verify it will be raised.
      notification = model.notification.create_repo_notification(repo, 'vulnerability_found',
                                                                 'quay_notification', {},
                                                                 {'level': 1})
      self.assertTrue(VulnerabilityFoundEvent().should_perform(event_data, notification))

      # Create another notification with a higher level and verify it won't be raised.
      notification = model.notification.create_repo_notification(repo, 'vulnerability_found',
                                                                 'quay_notification', {},
                                                                 {'level': 0})
      self.assertFalse(VulnerabilityFoundEvent().should_perform(event_data, notification))

  def test_select_images_to_scan(self):
    # Set all images to have a security index of a version to that of the config.
    expected_version = app.config['SECURITY_SCANNER_ENGINE_VERSION_TARGET']
    Image.update(security_indexed_engine=expected_version).execute()

    # Ensure no images are available for scanning.
    self.assertIsNone(model.image.get_min_id_for_sec_scan(expected_version))
    self.assertTrue(len(model.image.get_images_eligible_for_scan(expected_version)) == 0)

    # Check for a higher version.
    self.assertIsNotNone(model.image.get_min_id_for_sec_scan(expected_version + 1))
    self.assertTrue(len(model.image.get_images_eligible_for_scan(expected_version + 1)) > 0)

  def test_notification_worker(self):
    layer1 = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    layer2 = model.tag.get_tag_image(ADMIN_ACCESS_USER, COMPLEX_REPO, 'prod', include_storage=True)

    # Add a repo events for the layers.
    simple_repo = model.repository.get_repository(ADMIN_ACCESS_USER, SIMPLE_REPO)
    complex_repo = model.repository.get_repository(ADMIN_ACCESS_USER, COMPLEX_REPO)

    model.notification.create_repo_notification(simple_repo, 'vulnerability_found',
                                                'quay_notification', {}, {'level': 100})
    model.notification.create_repo_notification(complex_repo, 'vulnerability_found',
                                                'quay_notification', {}, {'level': 100})

    # Ensure that there are no event queue items for the layer.
    self.assertIsNone(notification_queue.get())

    with fake_security_scanner() as security_scanner:
      # Test with an unknown notification.
      worker = SecurityNotificationWorker(None)
      self.assertFalse(worker.perform_notification_work({
        'Name': 'unknownnotification'
      }))

      # Add some analyzed layers.
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer1)
      analyzer.analyze_recursively(layer2)

      # Add a notification with pages of data.
      new_vuln_info = {
        "Name": "CVE-TEST",
        "Namespace": "debian:8",
        "Description": "Some service",
        "Link": "https://security-tracker.debian.org/tracker/CVE-2014-9471",
        "Severity": "Critical",
        "FixedIn": {'Version': "9.23-5"},
      }

      security_scanner.set_vulns(security_scanner.layer_id(layer1), [new_vuln_info])
      security_scanner.set_vulns(security_scanner.layer_id(layer2), [new_vuln_info])

      layer_ids = [security_scanner.layer_id(layer1), security_scanner.layer_id(layer2)]
      notification_data = security_scanner.add_notification([], layer_ids, None, new_vuln_info)

      # Test with a known notification with pages.
      data = {
        'Name': notification_data['Name'],
      }

      worker = SecurityNotificationWorker(None)
      self.assertTrue(worker.perform_notification_work(data, layer_limit=2))

      # Make sure all pages were processed by ensuring we have two notifications.
      time.sleep(1)
      self.assertIsNotNone(notification_queue.get())
      self.assertIsNotNone(notification_queue.get())

  def test_notification_worker_offset_pages_not_indexed(self):
    # Try without indexes.
    self.assert_notification_worker_offset_pages(indexed=False)

  def test_notification_worker_offset_pages_indexed(self):
    # Try with indexes.
    self.assert_notification_worker_offset_pages(indexed=True)

  def assert_notification_worker_offset_pages(self, indexed=False):
    layer1 = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)
    layer2 = model.tag.get_tag_image(ADMIN_ACCESS_USER, COMPLEX_REPO, 'prod', include_storage=True)

    # Add a repo events for the layers.
    simple_repo = model.repository.get_repository(ADMIN_ACCESS_USER, SIMPLE_REPO)
    complex_repo = model.repository.get_repository(ADMIN_ACCESS_USER, COMPLEX_REPO)

    model.notification.create_repo_notification(simple_repo, 'vulnerability_found',
                                                'quay_notification', {}, {'level': 100})
    model.notification.create_repo_notification(complex_repo, 'vulnerability_found',
                                                'quay_notification', {}, {'level': 100})

    # Ensure that there are no event queue items for the layer.
    self.assertIsNone(notification_queue.get())

    with fake_security_scanner() as security_scanner:
      # Test with an unknown notification.
      worker = SecurityNotificationWorker(None)
      self.assertFalse(worker.perform_notification_work({
        'Name': 'unknownnotification'
      }))

      # Add some analyzed layers.
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer1)
      analyzer.analyze_recursively(layer2)

      # Add a notification with pages of data.
      new_vuln_info = {
        "Name": "CVE-TEST",
        "Namespace": "debian:8",
        "Description": "Some service",
        "Link": "https://security-tracker.debian.org/tracker/CVE-2014-9471",
        "Severity": "Critical",
        "FixedIn": {'Version': "9.23-5"},
      }

      security_scanner.set_vulns(security_scanner.layer_id(layer1), [new_vuln_info])
      security_scanner.set_vulns(security_scanner.layer_id(layer2), [new_vuln_info])

      # Define offsetting sets of layer IDs, to test cross-pagination support. In this test, we
      # will only serve 2 layer IDs per page: the first page will serve both of the 'New' layer IDs,
      # but since the first 2 'Old' layer IDs are "earlier" than the shared ID of
      # `devtable/simple:latest`, they won't get served in the 'New' list until the *second* page.
      # The notification handling system should correctly not notify for this layer, even though it
      # is marked 'New' on page 1 and marked 'Old' on page 2. Clair will served these
      # IDs sorted in the same manner.
      idx_old_layer_ids = [{'LayerName': 'old1', 'Index': 1},
                           {'LayerName': 'old2', 'Index': 2},
                           {'LayerName': security_scanner.layer_id(layer1), 'Index': 3}]

      idx_new_layer_ids = [{'LayerName': security_scanner.layer_id(layer1), 'Index': 3},
                           {'LayerName': security_scanner.layer_id(layer2), 'Index': 4}]

      old_layer_ids = [t['LayerName'] for t in idx_old_layer_ids]
      new_layer_ids = [t['LayerName'] for t in idx_new_layer_ids]

      if not indexed:
        idx_old_layer_ids = None
        idx_new_layer_ids = None

      notification_data = security_scanner.add_notification(old_layer_ids, new_layer_ids, None,
                                                            new_vuln_info, max_per_page=2,
                                                            indexed_old_layer_ids=idx_old_layer_ids,
                                                            indexed_new_layer_ids=idx_new_layer_ids)

      # Test with a known notification with pages.
      data = {
        'Name': notification_data['Name'],
      }

      worker = SecurityNotificationWorker(None)
      self.assertTrue(worker.perform_notification_work(data, layer_limit=2))

      # Make sure all pages were processed by ensuring we have only one notification. If the second
      # page was not processed, then the `Old` entry for layer1 will not be found, and we'd get two
      # notifications.
      time.sleep(1)
      self.assertIsNotNone(notification_queue.get())
      self.assertIsNone(notification_queue.get())


  def test_layer_gc(self):
    layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest', include_storage=True)

    # Delete the prod tag so that only the `latest` tag remains.
    model.tag.delete_tag(ADMIN_ACCESS_USER, SIMPLE_REPO, 'prod')

    with fake_security_scanner() as security_scanner:
      # Analyze the layer.
      analyzer = LayerAnalyzer(app.config, self.api)
      analyzer.analyze_recursively(layer)

      layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      self.assertAnalyzed(layer, security_scanner, True, 1)
      self.assertTrue(security_scanner.has_layer(security_scanner.layer_id(layer)))

      namespace_user = model.user.get_user(ADMIN_ACCESS_USER)
      model.user.change_user_tag_expiration(namespace_user, 0)

      # Delete the tag in the repository and GC.
      model.tag.delete_tag(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
      time.sleep(1)

      repo = model.repository.get_repository(ADMIN_ACCESS_USER, SIMPLE_REPO)
      model.repository.garbage_collect_repo(repo)

      # Ensure that the security scanner no longer has the image.
      self.assertFalse(security_scanner.has_layer(security_scanner.layer_id(layer)))


if __name__ == '__main__':
  unittest.main()