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

Have security scanner analyze only send notifications for *new* layers

Browse source 
Following this change, anytime a layer is indexed by the security scanner, we only send notifications out if the layer previously had a security_indexed_engine value of `-1`, thus ensuring it has *never* been indexed previously. This will allow us to change to version of the security scanner upwards, and have all the images be re-indexed, without firing off notifications in a spammy manner.
This commit is contained in:
Joseph Schorr 2016-12-13 22:51:29 -05:00
parent 5686c80af1
commit 624b2a8385
3 changed files with 53 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

39
test/test_secscan.py
View file

@ -301,6 +301,45 @@ class TestSecurityScanner(unittest.TestCase):
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
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)
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.
time.sleep(1)
self.assertIsNone(notification_queue.get())
# 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 _get_notification_data(self, new_layer_ids, old_layer_ids, new_severity='Low'):
return {