vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Send notifications for previously unscannable layers in QSS

Browse source 
Following this change, if an image was previously indexed unsuccessfully, then we will send notifications once successfully indexed
This commit is contained in:
Joseph Schorr 2016-12-14 11:10:53 -05:00
parent 2a6632cff4
commit 6871eb95b1
2 changed files with 42 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

73
test/test_secscan.py
View file

@ -10,7 +10,7 @@ from util.secscan.api import SecurityScannerAPI, AnalyzeLayerException
from util.secscan.analyzer import LayerAnalyzer
from util.secscan.notifier import process_notification_data
from data import model
from data.database import Image
from data.database import Image, IMAGE_NOT_SCANNED_ENGINE_VERSION
from workers.security_notification_worker import SecurityNotificationWorker
from endpoints.v2 import v2_bp
@ -271,8 +271,7 @@ class TestSecurityScanner(unittest.TestCase):
self.assertEquals(False, layer.security_indexed)
self.assertEquals(1, layer.security_indexed_engine)
def test_analyze_layer_success_events(self):
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)
@ -284,47 +283,11 @@ class TestSecurityScanner(unittest.TestCase):
repo = model.repository.get_repository(ADMIN_ACCESS_USER, SIMPLE_REPO)
model.notification.create_repo_notification(repo, 'vulnerability_found', 'quay_notification', {}, {'level': 100})
with HTTMock(analyze_layer_success_mock, get_layer_success_mock, response_content):
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, True, 1)
# Ensure an event was written for the tag.
time.sleep(1)
queue_item = notification_queue.get()
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'])
# 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_no_notification(self):
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())
# Set the security_indexed_engine of the layer to 0 to ensure it is marked as having been
# indexed (in some form) before this call.
layer.security_indexed_engine = 0
# Update the layer's state before analyzing.
layer.security_indexed_engine = security_indexed_engine
layer.security_indexed = security_indexed
layer.save()
# 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})
with HTTMock(analyze_layer_success_mock, get_layer_success_mock, response_content):
analyzer = LayerAnalyzer(app.config, self.api)
analyzer.analyze_recursively(layer)
@ -332,15 +295,37 @@ class TestSecurityScanner(unittest.TestCase):
layer = model.tag.get_tag_image(ADMIN_ACCESS_USER, SIMPLE_REPO, 'latest')
self.assertAnalyzed(layer, True, 1)
# Ensure no event was written for the tag, as the layer was being re-indexed.
# Ensure an event was written for the tag (if necessary).
time.sleep(1)
self.assertIsNone(notification_queue.get())
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'])
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 _get_notification_data(self, new_layer_ids, old_layer_ids, new_severity='Low'):
return {