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

Fix bug in QSS notifications

Browse source
This commit is contained in:
Joseph Schorr 2017-03-10 11:25:55 -05:00
parent 850c32ebfb
commit 48db77b521
2 changed files with 13 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
endpoints/notificationevent.py
View file

@ -112,6 +112,7 @@ def _build_summary(event_data):
class VulnerabilityFoundEvent(NotificationEvent): class VulnerabilityFoundEvent(NotificationEvent):
CONFIG_LEVEL = 'level' CONFIG_LEVEL = 'level'
PRIORITY_KEY = 'priority'
VULNERABILITY_KEY = 'vulnerability' VULNERABILITY_KEY = 'vulnerability'
MULTIPLE_VULNERABILITY_KEY = 'vulnerabilities' MULTIPLE_VULNERABILITY_KEY = 'vulnerabilities'
@ -120,7 +121,8 @@ class VulnerabilityFoundEvent(NotificationEvent):
return 'vulnerability_found' return 'vulnerability_found'
def get_level(self, event_data, notification_data): def get_level(self, event_data, notification_data):
priority = event_data[VulnerabilityFoundEvent.CONFIG_LEVEL]['priority'] vuln_data = event_data[VulnerabilityFoundEvent.VULNERABILITY_KEY]
priority = vuln_data[VulnerabilityFoundEvent.PRIORITY_KEY]
if priority == 'Defcon1' or priority == 'Critical': if priority == 'Defcon1' or priority == 'Critical':
return 'error' return 'error'
@ -138,6 +140,8 @@ class VulnerabilityFoundEvent(NotificationEvent):
'namespace_name': notification.repository.namespace_user.username, 'namespace_name': notification.repository.namespace_user.username,
'name': notification.repository.name, 'name': notification.repository.name,
}) })
level = event_config.get(VulnerabilityFoundEvent.CONFIG_LEVEL, 'Critical')
return build_event_data(repo, { return build_event_data(repo, {
'tags': ['latest', 'prod', 'foo', 'bar', 'baz'], 'tags': ['latest', 'prod', 'foo', 'bar', 'baz'],
'image': 'some-image-id', 'image': 'some-image-id',
@ -145,7 +149,7 @@ class VulnerabilityFoundEvent(NotificationEvent):
'id': 'CVE-FAKE-CVE', 'id': 'CVE-FAKE-CVE',
'description': 'A futurist vulnerability', 'description': 'A futurist vulnerability',
'link': 'https://security-tracker.debian.org/tracker/CVE-FAKE-CVE', 'link': 'https://security-tracker.debian.org/tracker/CVE-FAKE-CVE',
'priority': get_priority_for_index(event_config[VulnerabilityFoundEvent.CONFIG_LEVEL]) 'priority': get_priority_for_index(level)
}, },
}) })
@ -167,10 +171,13 @@ class VulnerabilityFoundEvent(NotificationEvent):
return actual_level_index <= filter_level_index return actual_level_index <= filter_level_index
def get_summary(self, event_data, notification_data): def get_summary(self, event_data, notification_data):
vuln_key = VulnerabilityFoundEvent.VULNERABILITY_KEY
priority_key = VulnerabilityFoundEvent.PRIORITY_KEY
multiple_vulns = event_data.get(VulnerabilityFoundEvent.MULTIPLE_VULNERABILITY_KEY) multiple_vulns = event_data.get(VulnerabilityFoundEvent.MULTIPLE_VULNERABILITY_KEY)
if multiple_vulns is not None: if multiple_vulns is not None:
top_priority = multiple_vulns[0].get('priority', 'Unknown') top_priority = multiple_vulns[0].get(priority_key, 'Unknown')
matching = [v for v in multiple_vulns if v.get('priority', 'Unknown') == top_priority] matching = [v for v in multiple_vulns if v.get(priority_key, 'Unknown') == top_priority]
msg = '%s %s' % (len(matching), top_priority) msg = '%s %s' % (len(matching), top_priority)
if len(matching) < len(multiple_vulns): if len(matching) < len(multiple_vulns):
@ -180,7 +187,7 @@ class VulnerabilityFoundEvent(NotificationEvent):
return msg % (event_data['repository'], len(event_data['tags'])) return msg % (event_data['repository'], len(event_data['tags']))
else: else:
msg = '%s vulnerability detected in repository %s in %s tags' msg = '%s vulnerability detected in repository %s in %s tags'
return msg % (event_data['vulnerability']['priority'], event_data['repository'], return msg % (event_data[vuln_key][priority_key], event_data['repository'],
len(event_data['tags'])) len(event_data['tags']))

1
test/test_secscan.py
View file

@ -355,6 +355,7 @@ class TestSecurityScanner(unittest.TestCase):
event = VulnerabilityFoundEvent() event = VulnerabilityFoundEvent()
msg = '1 Low and 1 more vulnerabilities were detected in repository devtable/simple in 2 tags' 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(msg, event.get_summary(body['event_data'], {}))
self.assertEquals('info', event.get_level(body['event_data'], {}))
else: else:
self.assertIsNone(queue_item) self.assertIsNone(queue_item)