import logging import time from datetime import datetime from notificationhelper import build_event_data from util.jinjautil import get_template_env template_env = get_template_env("events") logger = logging.getLogger(__name__) class InvalidNotificationEventException(Exception): pass class NotificationEvent(object): def __init__(self): pass def get_level(self, event_data, notification_data): """ Returns a 'level' representing the severity of the event. Valid values are: 'info', 'warning', 'error', 'primary', 'success' """ raise NotImplementedError def get_summary(self, event_data, notification_data): """ Returns a human readable one-line summary for the given notification data. """ raise NotImplementedError def get_message(self, event_data, notification_data): """ Returns a human readable HTML message for the given notification data. """ return template_env.get_template(self.event_name() + '.html').render({ 'event_data': event_data, 'notification_data': notification_data }) def get_sample_data(self, repository=None): """ Returns sample data for testing the raising of this notification, with an optional repository. """ raise NotImplementedError @classmethod def event_name(cls): """ Particular event implemented by subclasses. """ raise NotImplementedError @classmethod def get_event(cls, eventname): for subc in cls.__subclasses__(): if subc.event_name() == eventname: return subc() raise InvalidNotificationEventException('Unable to find event: %s' % eventname) class RepoPushEvent(NotificationEvent): @classmethod def event_name(cls): return 'repo_push' def get_level(self, event_data, notification_data): return 'primary' def get_summary(self, event_data, notification_data): return 'Repository %s updated' % (event_data['repository']) def get_sample_data(self, repository): return build_event_data(repository, { 'updated_tags': {'latest': 'someimageid', 'foo': 'anotherimage'}, 'pruned_image_count': 3 }) def _build_summary(event_data): """ Returns a summary string for the build data found in the event data block. """ summary = 'for repository %s [%s]' % (event_data['repository'], event_data['build_id'][0:7]) return summary class VulnerabilityFoundEvent(NotificationEvent): @classmethod def event_name(cls): return 'vulnerability_found' def get_level(self, event_data, notification_data): priority = event_data['vulnerability']['priority'] if priority == 'Defcon1' or priority == 'Critical': return 'error' if priority == 'Medium' or priority == 'High': return 'warning' return 'info' def get_sample_data(self, repository): return build_event_data(repository, { 'tags': ['latest', 'prod'], 'image': 'some-image-id', 'vulnerability': { 'id': 'CVE-FAKE-CVE', 'description': 'A futurist vulnerability', 'link': 'https://security-tracker.debian.org/tracker/CVE-FAKE-CVE', 'priority': 'Critical', }, }) def get_summary(self, event_data, notification_data): msg = '%s vulnerability detected in repository %s in tags %s' return msg % (event_data['vulnerability']['priority'], event_data['repository'], ', '.join(event_data['tags'])) class BuildQueueEvent(NotificationEvent): @classmethod def event_name(cls): return 'build_queued' def get_level(self, event_data, notification_data): return 'info' def get_sample_data(self, repository): build_uuid = 'fake-build-id' return build_event_data(repository, { 'is_manual': False, 'build_id': build_uuid, 'build_name': 'some-fake-build', 'docker_tags': ['latest', 'foo', 'bar'], 'trigger_id': '1245634', 'trigger_kind': 'GitHub', 'trigger_metadata': { "default_branch": "master", "ref": "refs/heads/somebranch", "commit": "42d4a62c53350993ea41069e9f2cfdefb0df097d", "commit_info": { 'url': 'http://path/to/the/commit', 'message': 'Some commit message', 'date': time.mktime(datetime.now().timetuple()), 'author': { 'username': 'fakeauthor', 'url': 'http://path/to/fake/author/in/scm', 'avatar_url': 'http://www.gravatar.com/avatar/fakehash' } } } }, subpage='/build/%s' % build_uuid) def get_summary(self, event_data, notification_data): return 'Build queued ' + _build_summary(event_data) class BuildStartEvent(NotificationEvent): @classmethod def event_name(cls): return 'build_start' def get_level(self, event_data, notification_data): return 'info' def get_sample_data(self, repository): build_uuid = 'fake-build-id' return build_event_data(repository, { 'build_id': build_uuid, 'build_name': 'some-fake-build', 'docker_tags': ['latest', 'foo', 'bar'], 'trigger_id': '1245634', 'trigger_kind': 'GitHub', 'trigger_metadata': { "default_branch": "master", "ref": "refs/heads/somebranch", "commit": "42d4a62c53350993ea41069e9f2cfdefb0df097d" } }, subpage='/build/%s' % build_uuid) def get_summary(self, event_data, notification_data): return 'Build started ' + _build_summary(event_data) class BuildSuccessEvent(NotificationEvent): @classmethod def event_name(cls): return 'build_success' def get_level(self, event_data, notification_data): return 'success' def get_sample_data(self, repository): build_uuid = 'fake-build-id' return build_event_data(repository, { 'build_id': build_uuid, 'build_name': 'some-fake-build', 'docker_tags': ['latest', 'foo', 'bar'], 'trigger_id': '1245634', 'trigger_kind': 'GitHub', 'trigger_metadata': { "default_branch": "master", "ref": "refs/heads/somebranch", "commit": "42d4a62c53350993ea41069e9f2cfdefb0df097d" }, 'image_id': '1245657346' }, subpage='/build/%s' % build_uuid) def get_summary(self, event_data, notification_data): return 'Build succeeded ' + _build_summary(event_data) class BuildFailureEvent(NotificationEvent): @classmethod def event_name(cls): return 'build_failure' def get_level(self, event_data, notification_data): return 'error' def get_sample_data(self, repository): build_uuid = 'fake-build-id' return build_event_data(repository, { 'build_id': build_uuid, 'build_name': 'some-fake-build', 'docker_tags': ['latest', 'foo', 'bar'], 'trigger_kind': 'GitHub', 'error_message': 'This is a fake error message', 'trigger_id': '1245634', 'trigger_kind': 'GitHub', 'trigger_metadata': { "default_branch": "master", "ref": "refs/heads/somebranch", "commit": "42d4a62c53350993ea41069e9f2cfdefb0df097d", "commit_info": { 'url': 'http://path/to/the/commit', 'message': 'Some commit message', 'date': time.mktime(datetime.now().timetuple()), 'author': { 'username': 'fakeauthor', 'url': 'http://path/to/fake/author/in/scm', 'avatar_url': 'http://www.gravatar.com/avatar/fakehash' } } } }, subpage='/build?current=%s' % build_uuid) def get_summary(self, event_data, notification_data): return 'Build failure ' + _build_summary(event_data)