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

Merge pull request #1295 from coreos-inc/fixeventseverity

Browse source 
Add another test for security notification filtering
This commit is contained in:
josephschorr 2016-03-17 13:00:59 -04:00
commit c1cceb2949
2 changed files with 23 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
endpoints/notificationevent.py
View file

@ -122,10 +122,14 @@ class VulnerabilityFoundEvent(NotificationEvent):
def should_perform(self, event_data, notification_data):
event_config = json.loads(notification_data.event_config_json)
expected_level_index = event_config['level']
priority = PRIORITY_LEVELS[event_data['vulnerability']['priority']]
actual_level_index = priority['index']
return actual_level_index <= expected_level_index
filter_level_index = int(event_config['level'])
event_severity = PRIORITY_LEVELS.get(event_data['vulnerability']['priority'])
if event_severity is None:
return False
actual_level_index = int(event_severity['index'])
return actual_level_index <= filter_level_index
def get_summary(self, event_data, notification_data):
msg = '%s vulnerability detected in repository %s in tags %s'

18
test/test_secscan.py
View file

@ -4,6 +4,7 @@ import os
from httmock import urlmatch, all_requests, HTTMock
from app import app, config_provider, storage, notification_queue
from endpoints.notificationevent import VulnerabilityFoundEvent
from initdb import setup_database_for_testing, finished_database_for_testing
from util.secscan.api import SecurityScannerAPI, AnalyzeLayerException
from util.secscan.analyzer import LayerAnalyzer
@ -405,7 +406,7 @@ class TestSecurityScanner(unittest.TestCase):
# 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})
notification = model.notification.create_repo_notification(repo, 'vulnerability_found', 'quay_notification', {}, {'level': 100})
@urlmatch(netloc=r'(.*\.)?mockclairservice', path=r'/v1/layers/(.+)')
def get_matching_layer_vulnerable(url, request):
@ -436,7 +437,7 @@ class TestSecurityScanner(unittest.TestCase):
# Fire off the notification processing.
with HTTMock(get_matching_layer_vulnerable, response_content):
notification_data = self._get_notification_data([layer_id], [layer_id], new_severity='High')
notification_data = self._get_notification_data([layer_id], [layer_id], new_severity='Critical')
self.assertTrue(process_notification_data(notification_data))
# Ensure an event was written for the tag.
@ -446,9 +447,20 @@ class TestSecurityScanner(unittest.TestCase):
body = json.loads(queue_item.body)
self.assertEquals(['prod', 'latest'], body['event_data']['tags'])
self.assertEquals('CVE-TEST', body['event_data']['vulnerability']['id'])
self.assertEquals('High', body['event_data']['vulnerability']['priority'])
self.assertEquals('Critical', body['event_data']['vulnerability']['priority'])
self.assertTrue(body['event_data']['vulnerability']['has_fix'])
# Verify that an event would be raised.
event_data = 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))
if __name__ == '__main__':
unittest.main()