# encoding: utf-8 import boto3 import json import logging import os import time import traceback import uuid from boto3.dynamodb.conditions import Attr, Key from botocore.exceptions import ClientError from functools import reduce from .exceptions import * from .utils import * SUPPORTED_ACTIVITY_TYPES = [ "Action", "Event", ] SUPPORTED_SEVERITY_ALIASES = { "debug": -4, "info": 0, "warning": 4, "critical": 8, } REQUIRED_TIMESTAMP_KEYS = [ # record_time is filled by us "calibration_time", ] SUPPORTED_TIMESTAMP_KEYS = [ # record_time is filled by us "calibration_time", "report_time", "occurrence_time", "detect_time", "expect_time", ] class Activity: """Acvitity is the basic object in Activity Hub.""" db_type = "dynamodb" def __init__(self, db_type, logger=None): if logger: self.logger = logger else: self.logger = logging.getLogger() self.logger.setLevel(logging.INFO) """init db resource""" if db_type: self.db_type = db_type self.db = boto3.resource(self.db_type) activity_table_name = os.getenv('ACTIVITYHUB_MESSAGE_TABLE_NAME') source_table_name = os.getenv('ACTIVITYHUB_SOURCE_TABLE_NAME') try: self.msg_table = self.db.Table(activity_table_name) self.msg_table.load() self.src_table = self.db.Table(source_table_name) self.src_table.load() except ClientError as e: self.logger.error("error connecting dynamodb: %s", e) raise def validate_message(self, msg): try: if msg['type'].capitalize() not in SUPPORTED_ACTIVITY_TYPES: raise ActivityValidationError( "Unsupported Activity type {}".format(msg['type'])) except KeyError: raise ActivityValidationError("Activity type must be specified") try: if msg['name'] == '': raise ActivityValidationError( "Activity name must be specified") except KeyError: raise ActivityValidationError("Activity name must be specified") try: if msg['target_id'] == '': raise ActivityValidationError("Target ID must be specified") except KeyError: raise ActivityValidationError("Target ID must be specified") # check for pre-defined timestamp values for tk in SUPPORTED_TIMESTAMP_KEYS: try: t = int(msg[tk]) msg[tk] = t if t <= 0: raise ActivityValidationError( f"{tk} must be a positive value") except KeyError: if tk in REQUIRED_TIMESTAMP_KEYS: raise ActivityValidationError(f"{tk} must be specified") except ValueError: raise ActivityValidationError(f"{tk} must be an integer") # check for custom tags try: for k, v in msg["tags"].items(): if '::' in k or '::' in v: raise ActivityValidationError( "reserved seperator '::' is not allowed in tag keys and values") if ',' in k or ',' in v: raise ActivityValidationError( "reserved seperator ',' is not allowed in tag keys and values") except KeyError: pass except AttributeError: raise ActivityValidationError( "tags must be a key-value map") def auth_check(self, headers): try: _source = headers['x-activityhub-source'] _psk = headers['x-activityhub-key'] if _source == '' or _psk == '': raise ActivityAuthenticationError( "Source and Source Key must be specified") # use a dedicate variable to store auth source, to avoid conflic # with source variable in query param list self.auth_source = _source.lower() try: _item = self.src_table.get_item(Key={'name': _source}) if _item['Item']['psk'] != _psk: raise ActivityAuthenticationError( "Source Key verification failed: PSK mismatch") except Exception as e: raise ActivityAuthenticationError( f'Source Key verification failed: {e}') except KeyError: raise ActivityAuthenticationError( "Source and Source Key must be specified") def parse_activity(self, msg): try: self.validate_message(msg) except ActivityValidationError as e: self.logger.error(json.dumps(traceback.format_exception( type(e), e, e.__traceback__))) raise e # populate necessary fields try: if msg['activity_id'] == '': msg['activity_id'] = str(uuid.uuid4()) except KeyError: msg['activity_id'] = str(uuid.uuid4()) try: if msg['context_id'] == '': msg['context_id'] = str(uuid.uuid4()) except KeyError: msg['context_id'] = str(uuid.uuid4()) try: msg['severity'] = int(msg['severity']) except ValueError: try: msg['severity'] = SUPPORTED_SEVERITY_ALIASES[msg['severity'].lower()] except KeyError: raise ActivityValidationError( "Severity value must be a number") except KeyError: msg['severity'] = 0 # type and name are already checked in validate_message() so no need to try-except again msg['display_name'] = msg['name'] msg['name'] = msg['name'].lower() msg['type'] = msg['type'].capitalize() # handle source fields msg['source'] = self.auth_source try: del msg['source_key'] self.logger.debug('Deleted source_key from input messaage') except KeyError: self.logger.debug('Ignored KeyError for source_key') pass # parse other fields try: msg['catalog'] = msg['catalog'].lower() except KeyError: self.logger.debug('Ignored KeyError for catalog') pass msg['record_time'] = int(time.time()) def split_param(self, param): if param == '': return None return param.split(',') def populate_query_params(self, params): self.logger.debug(f'populating query params: {params}') if not params: # populate default values params = {} try: self.catalog = [c.lower() for c in self.split_param(params['catalog'])] except (KeyError, TypeError): self.catalog = None try: self.type = [t.capitalize() for t in self.split_param(params['type'])] except (KeyError, TypeError): self.type = None try: self.name = [n.lower() for n in self.split_param(params['name'])] except (KeyError, TypeError): self.name = None try: _severity = params['severity'] try: self.severity = SUPPORTED_SEVERITY_ALIASES[_severity] except KeyError: try: self.severity = int(_severity) except: raise ActivityQueryError( 'Severity value must be a number.') except KeyError: self.severity = None try: self.source = [s.lower() for s in self.split_param(params['source'])] except (KeyError, TypeError): self.source = None try: self.author = [s for s in self.split_param(params['author'])] except (KeyError, TypeError): self.author = None try: self.application_id = params['application_id'] except KeyError: self.application_id = None try: self.concept_id = params['concept_id'] except KeyError: self.concept_id = None try: self.target_id = params['target_id'] except KeyError: self.target_id = None # tags try: self.tags = {} for _pair in params['tags'].split(','): if not _pair or _pair == "": continue _kv = _pair.split('::') if len(_kv) != 2: raise ActivityQueryError(f"Invalid tag pair {_pair}") self.tags[_kv[0]] = _kv[1] except KeyError: self.tags = None # controling params try: self.limit_cnt = int(params['limit']) if self.limit_cnt > 100: raise ActivityQueryError('Limit cannot be greater than 100.') except KeyError: self.limit_cnt = None except ValueError: raise ActivityQueryError('Limit value must be an integer.') try: self.desc_order = bool(params['desc']) except KeyError: self.desc_order = False # default to asce except ValueError: raise ActivityQueryError('Desc flag must be a boolean.') def build_filter_exp(self, search=False): _exp = [] if self.catalog and len(self.catalog) > 0: if search: _exp.append(Attr('catalog').contains(self.catalog[0])) else: _exp.append(Attr('catalog').is_in(self.catalog)) if self.type and len(self.type) > 0: _exp.append(Attr('type').is_in(self.type)) if self.name and len(self.name) > 0: if search: _exp.append(Attr('name').contains(self.name[0])) else: _exp.append(Attr('name').is_in(self.name)) if self.severity is not None: # 0 is false _exp.append(Attr('severity').gte(self.severity)) if self.source and len(self.source) > 0: if search: _exp.append(Attr('source').contains(self.source[0])) else: _exp.append(Attr('source').is_in(self.source)) if self.author and len(self.author) > 0: # do partial matching for all author input and ignores multiple values _exp.append(Attr('author').contains(self.author[0])) if self.application_id and self.application_id != '': _exp.append(Attr('application_id').eq(self.application_id)) if self.concept_id and self.concept_id != '': _exp.append(Attr('concept_id').eq(self.concept_id)) if self.target_id and self.target_id != '': _exp.append(Attr('target_id').eq(self.target_id)) if self.tags and len(self.tags) > 0: for _k, _v in self.tags.items(): _exp.append(Attr(f'tags.{_k}').eq(_v)) if len(_exp) < 1: # use an always-true condition to workaround for empty filter return Attr('activity_id').exists() return reduce(lambda x, y: x & y, _exp) def parse_cursor(self, headers): try: _cur = headers['x-activityhub-cursor'] if len(_cur) > 0: self.cursor = decode_obj(_cur) self.logger.debug(f'decoded cursor {self.cursor}') else: self.cursor = None self.logger.info(f'ignored empty cursor from request.') except KeyError: self.cursor = None return def build_query_kwargs(self, kwargs): # https://github.com/boto/botocore/issues/1688#issuecomment-599691978 if self.cursor: kwargs['ExclusiveStartKey'] = self.cursor if self.limit_cnt: # add an extra count to make query_to_limit behave more friendly to clients kwargs['Limit'] = self.limit_cnt + \ int(1 if self.limit_cnt < 10 else self.limit_cnt * 0.1) kwargs['ScanIndexForward'] = not self.desc_order return kwargs def populate_timestamps(self, from_ts, to_ts): # type checks if not from_ts or from_ts == '': raise ActivityQueryError('Start timestamp must be specified.') start_ts = None try: start_ts = int(from_ts) except: raise ActivityQueryError( f'Start timestamp {from_ts} is not valid.') if not to_ts or to_ts == '': to_ts = time.time() end_ts = None try: end_ts = int(to_ts) except: raise ActivityQueryError( f'End timestamp {to_ts} is not valid.') # logical checks if start_ts >= end_ts: raise ActivityQueryError( "Start timestamp must be earlier than end.") if end_ts - start_ts > 60 * 60 * 24 * 90: # 90 days raise ActivityQueryError( "Time range loger than 90 days is not supported.") return start_ts, end_ts def write(self, msg): try: self.parse_activity(msg) except: raise try: self.msg_table.put_item( Item=msg, ConditionExpression='attribute_not_exists(activity_id)') except: raise return msg['activity_id'] def get_message(self, msg_id): if not msg_id or msg_id == '': raise ActivityQueryError('Activity ID must be specified.') try: resp = self.msg_table.query( IndexName='activity_id-index', KeyConditionExpression=Key('activity_id').eq(msg_id) ) self.logger.info( f"Query finished. Scanned: {resp['ScannedCount']}, Returned: {resp['Count']}, RequestID: {resp['ResponseMetadata']['RequestId']}") except ClientError as e: self.logger.error( "error querying activity %s from db: %s", msg_id, e) raise else: items = resp['Items'] if len(items) < 1: raise ActivityNotFoundError('Activity not found.') elif len(items) > 1: self.logger.warning( f'Multiple Activity with the same ID {msg_id} found.') raise ActivityQueryError( 'Multiple Activity with the same ID found, data might be corrupted.') return items[0] def query_to_limit(self, kwargs): # query for the first time try: resp = self.msg_table.query(**kwargs) self.logger.info( f"Query finished. Scanned: {resp['ScannedCount']}, Returned: {resp['Count']}, RequestID: {resp['ResponseMetadata']['RequestId']}") self.logger.debug(json.dumps(resp, cls=DecimalEncoder)) except ClientError as e: self.logger.error("error querying activity list from db: %s", e) raise result = [] cursor = None cnt = len(result) if self.limit_cnt: for _item in resp['Items']: if cnt >= self.limit_cnt: # reached the limit return result, cursor result.append(_item) cursor = { 'target_id': _item['target_id'], 'activity_id': _item['activity_id'], 'calibration_time': _item['calibration_time'], } cnt += 1 else: result = resp['Items'] if 'LastEvaluatedKey' not in resp: # already full data cursor = None # clear cursor # dynamo limits before filter, we need to give client literal limit # while 'LastEvaluatedKey' in resp: while not self.limit_cnt or cnt < self.limit_cnt: if 'LastEvaluatedKey' not in resp: # already full data cursor = None # clear cursor break try: kwargs['ExclusiveStartKey'] = resp['LastEvaluatedKey'] resp = self.msg_table.query(**kwargs) self.logger.info( f"Query finished. Scanned: {resp['ScannedCount']}, Returned: {resp['Count']}, RequestID: {resp['ResponseMetadata']['RequestId']}") self.logger.debug(json.dumps(resp, cls=DecimalEncoder)) except ClientError as e: self.logger.error( "error querying activity list from db: %s", e) raise for _item in resp['Items']: if self.limit_cnt and cnt >= self.limit_cnt: # reached the limit return result, cursor result.append(_item) cursor = { 'target_id': _item['target_id'], 'activity_id': _item['activity_id'], 'calibration_time': _item['calibration_time'], } cnt += 1 if len(result) < 1: raise ActivityNotFoundError('Activity not found.') return result, cursor def list_by_context(self, ctx_id): if not ctx_id or ctx_id == '': raise ActivityQueryError('Context ID must be specified.') kwargs = self.build_query_kwargs({ 'IndexName': 'context_id-calibration_time-index', 'KeyConditionExpression': Key('context_id').eq(ctx_id), 'FilterExpression': self.build_filter_exp(), }) resp, cursor = self.query_to_limit(kwargs) self.logger.info( f"Returning {len(resp)} items to client with limit set.") return resp, encode_obj(cursor) def list_activities(self, target, from_ts, to_ts, list_ctx=False, search=False): if not target or target == '': raise ActivityQueryError('Target ID must be specified.') start_ts, end_ts = self.populate_timestamps(from_ts, to_ts) kwargs = self.build_query_kwargs({ 'IndexName': 'target_id-calibration_time-index', 'KeyConditionExpression': Key('target_id').eq(target) & Key('calibration_time').between(start_ts, end_ts), 'FilterExpression': self.build_filter_exp(search), }) resp, cursor = self.query_to_limit(kwargs) if list_ctx: # only return context_id list _ctx_lst = set() for _item in resp: try: _ctx_lst.add(_item['context_id']) except KeyError: self.logger.debug('Ignored KeyError for context_id') pass return list(_ctx_lst), encode_obj(cursor) self.logger.info(f"Returning {len(resp)} items to client.") return resp, encode_obj(cursor)