Source code for mentat.idea.internal

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#-------------------------------------------------------------------------------
# This file is part of Mentat system (https://mentat.cesnet.cz/).
#
# Copyright (C) since 2011 CESNET, z.s.p.o (http://www.ces.net/)
# Use of this source is governed by the MIT license, see LICENSE file.
#-------------------------------------------------------------------------------


"""
This module provides classess and tools for object representation of
`IDEA <https://idea.cesnet.cz/en/index>`__ messages in Mentat project.

This module is based on ``idea.lite`` module from `idea <https://homeproj.cesnet.cz/git/idea.git>`__
package. It is attempting to reuse already defined structures wherever
and whenever possible and only append few custom datatypes to existing
definition.

The most important part of this module, which is available to users, is
the :py:class:`mentat.idea.internal.Idea` class. Following code snippet
demonstates use case scenario:

.. code-block:: python

    >>> import mentat.idea.internal

    # IDEA messages ussually come from regular dicts or JSON
    >>> idea_raw = {...}

    # Just pass the dict as parameter to constructor
    >>> idea_msg = mentat.idea.internal.Idea(idea_raw)

"""


__author__ = "Jan Mach <jan.mach@cesnet.cz>"
__credits__ = "Pavel Kácha <pavel.kacha@cesnet.cz>, Andrea Kropáčová <andrea.kropacova@cesnet.cz>"


import json
import pprint

import typedcols
import idea.base
import idea.lite

import pynspect.jpath
import pynspect.rules
import pynspect.compilers


[docs]def cesnet_dict_typedef(flavour, list_flavour, errors_list, abuses_list, addon=None): """ Typedef generator helper for easy usage of custom definitions at multiple places. """ cesnet_def = { "InspectionErrors": { "description": "List of event pecularities found during inspection", "type": errors_list }, "StorageTime": { "description": "Unix timestamp of the moment the message was stored into database", "type": flavour["Timestamp"] }, "EventTemplate": { "description": "Template used to generate the event (deprecated)", "type": flavour["String"] }, "EventClass": { "description": "Event class determined by inspection", "type": flavour["String"] }, "EventSeverity": { "description": "Event severity determined by inspection", "type": flavour["String"] }, "Impact": { "description": "More user friendly description of event impact, used for reporting (IDMEF legacy)", "type": flavour["String"] }, "ResolvedAbuses": { "description": "Abuse contacts related to any alert source", "type": abuses_list }, "SourceResolvedASN": { "description": "AS numbers related to any alert source", "type": list_flavour["Integer"] }, "SourceResolvedCountry": { "description": "Coutry ISO codes related to any alert source", "type": list_flavour["String"] } } if addon is not None: cesnet_def.update(addon) return cesnet_def
[docs]class CESNETDict(typedcols.TypedDict): # pylint: disable=locally-disabled,too-many-ancestors """ This type definition represents a custom subdictionary under key *_CESNET* in message root dictionary. """ allow_unknown = True typedef = cesnet_dict_typedef( idea.lite.idea_types, idea.lite.idea_lists, typedcols.typed_list("InspectionErrorsList", str), typedcols.typed_list("ResolvedAbusesList", str) )
[docs]def internal_base_addon_typedef(flavour, list_flavour, cesnet_dict, addon=None): # pylint: disable=locally-disabled,unused-argument """ Typedef generator helper for easy usage of custom definitions at multiple places. """ addon_def = { "ts": { "description": "CESNET specific timestamp as NTP timestamp", "type": flavour["Timestamp"] }, "ts_u": { "description": "CESNET specific timestamp as native Unix timestamp", "type": flavour["Integer"] }, "_CESNET": { "description": "Custom CESNET/Mentat abominations to IDEA definition", "type": cesnet_dict } } if addon is not None: addon_def.update(addon) return addon_def
[docs]class Idea(idea.lite.Idea): # pylint: disable=locally-disabled,too-many-ancestors """ This class attempts to make only very contained changes to original ``idea.lite.Idea`` class, so that there is no need to update this class in case underlying definition changes. To make these changes, the *addon* feature of ``typedcols`` library is used. """ allow_unknown = True typedef = idea.base.idea_typedef( idea.lite.idea_types, idea.lite.idea_lists, idea.lite.idea_defaults, typedcols.typed_list("SourceList", idea.lite.SourceTargetDict), typedcols.typed_list("TargetList", idea.lite.SourceTargetDict), typedcols.typed_list("AttachList", idea.lite.AttachDict), typedcols.typed_list("NodeList", idea.lite.NodeDict), internal_base_addon_typedef( idea.lite.idea_types, idea.lite.idea_lists, CESNETDict ) ) json_default = staticmethod(idea.lite.Idea.json_default) """ Helper method for JSON serialization of :py:class:`mentat.idea.internal.Idea` messages. Example usage: .. code-block:: python >>>import json >>>idea_internal = ... >>>json.dumps(idea_internal, indent=4, sort_keys=True, default=idea_internal.json_default) """
[docs] @staticmethod def from_json(json_string): """ Instantinate message object directly from given JSON serialized string. """ return Idea(json.loads(json_string))
[docs] def get_id(self): """ Convenience method for returning message identifier. :return: Value of message attribute ``idea['ID']``. :rtype: str """ return self['ID']
[docs] def get_detect_time(self): """ Convenience method for returning message detection time. :return: Value of message attribute ``idea['DetectTime']``. :rtype: datetime.datetime """ return self['DetectTime']
[docs] def get_storage_time(self): """ Convenience method for returning message storage time. :return: Value of message attribute ``idea['_CESNET']['StorageTime']``. :rtype: datetime.datetime """ return self.get('_CESNET', {}).get('StorageTime', None)
[docs] def get_class(self): """ Convenience method for returning message event class. :return: Value of message attribute ``idea['_CESNET']['EventClass']``. :rtype: str """ return self.get('_CESNET', {}).get('EventClass', None)
[docs] def get_severity(self): """ Convenience method for returning message event severity. :return: Value of message attribute ``idea['_CESNET']['EventSeverity']``. :rtype: str """ return self.get('_CESNET', {}).get('EventSeverity', None)
[docs] def get_abuses(self): """ Convenience method for returning list of all resolved abuses. :return: Value of message attribute ``idea['_CESNET']['ResolvedAbuses']``. :rtype: list of strings """ return list(self.get('_CESNET', {}).get('ResolvedAbuses', list()))
[docs] def get_categories(self): """ Convenience method for returning list of all message categories. :return: Value of message attribute ``idea['Category']``. :rtype: list of strings """ return list(self['Category'])
[docs] def get_description(self): """ Convenience method for returning message description. :return: Value of message attribute ``idea['description']``. :rtype: list of strings """ return self.get('Description', None)
[docs] def get_detectors(self): """ Convenience method for returning list of all message detectors. :return: Value of message attribute ``idea['Node']['Name']``. :rtype: list of strings """ return [node['Name'] for node in self.get('Node', [])]
[docs] def get_addresses(self, node, get_v4 = True, get_v6 = True): """ Convenience method for returning list of all addresses (both v4 and v6) for given node (``Source`` and ``Target``). :param str node: Type of the node (``Source`` or ``Target``). :param bool get_v4: Fetch IPv4 addressess. :param bool get_v6: Fetch IPv6 addressess. :return: Value of message attributes ``idea[node]['IP4']`` and ``idea[node]['IP6']``. :rtype: list of ipranges """ result = [] if node in self: for src in self[node]: if get_v4 and 'IP4' in src: result.extend(list(src['IP4'])) if get_v6 and 'IP6' in src: result.extend(list(src['IP6'])) return result
[docs] def get_ports(self, node): """ Convenience method for returning list of all ports for given node (``Source`` and ``Target``). :param str node: Type of the node (``Source`` or ``Target``). :return: Value of message attributes ``idea[node]['Port']``. :rtype: list of int """ result = set() if node in self: for src in self[node]: if 'Port' in src: for item in list(src['Port']): result.add(item) return sorted(list(result))
[docs] def get_protocols(self, node): """ Convenience method for returning list of all protocols for given node (``Source`` and ``Target``). :param str node: Type of the node (``Source`` or ``Target``). :return: Value of message attributes ``idea[node]['Port']``. :rtype: list of int """ result = set() if node in self: for src in self[node]: if 'Proto' in src: for item in list(src['Proto']): result.add(str(item).lower()) return sorted(list(result))
[docs] def get_types(self, node): """ Convenience method for returning list of all types for given node (``Source``, ``Target`` and ``Node``). :param str node: Type of the node (``Source``, ``Target`` or ``Node``). :return: Value of message attributes ``idea[node]['Port']``. :rtype: list of int """ result = set() if node in self: for src in self[node]: if 'Type' in src: for item in list(src['Type']): result.add(item) return sorted(list(result))
[docs] def get_countries_src(self): """ Convenience method for returning list of all resolved source countries. :return: Value of message attribute ``idea['_CESNET']['SourceResolvedCountry']``. :rtype: list of strings """ return list( self.get('_CESNET', {}).get('SourceResolvedCountry', []) )
[docs] def get_asns_src(self): """ Convenience method for returning list of all resolved source ASNs. :return: Value of message attribute ``idea['_CESNET']['SourceResolvedASN']``. :rtype: list of strings """ return list( self.get('_CESNET', {}).get('SourceResolvedASN', []) )
[docs] def get_jpath_value(self, jpath): """ Return single (first in case o a list) value on given JPath within the IDEA message. :param str jpath: JPath as defined in :py:mod:`pynspect.jpath` module. :return: Single (or first) value on given JPath. """ return pynspect.jpath.jpath_value(self, jpath)
[docs] def get_jpath_values(self, jpath): """ Return all values on given JPath within the IDEA message. :param str jpath: JPath as defined in :py:mod:`pynspect.jpath` module. :return: List of all values on given JPath. :rtype: list """ return pynspect.jpath.jpath_values(self, jpath)
[docs]class IdeaGhost(Idea): """ This class represents simplified IDEA message objects as reconstructed from SQL records retrieved from database. These records are represented by the :py:class:`mentat.idea.sqldb.Idea` class, however the native database record object is used when the message is fetched from database. Objects of this class ARE NOT a perfect 1to1 match with original objects based on :py:class:`mentat.idea.internal.Idea`. During the conversion to database representation some of the information gets lost due to the design of the data model optimized for database search. However from the point of view of the searching these objects are same, because there is the same set of categories, source and target IPs and all other message attributes, so they can be used as representatives in use cases where the full message is not necessary. The big pro in using these ghost objects is that they are much cheaper to construct in comparison with the full IDEA message representation. """
[docs] @classmethod def from_record(cls, record): """ Construct the IDEA ghost object from the given SQL record. """ idea_raw = {} idea_raw['ID'] = record.id idea_raw['DetectTime'] = record.detecttime idea_raw['Category'] = list(record.category) idea_raw['Description'] = record.description for whatfrom, whereto in (('source', 'Source'), ('target', 'Target')): datastore = {} ip_list = getattr(record, '{}_ip'.format(whatfrom), '').replace('{', '').replace('}', '') if ip_list: ip_list = ip_list.split(',') if ip_list: for ipaddr in ip_list: if ipaddr.find(':') != -1: datastore.setdefault('IP6',[]).append(ipaddr) else: datastore.setdefault('IP4',[]).append(ipaddr) if getattr(record, '{}_type'.format(whatfrom), []): datastore['Type'] = list(getattr(record, '{}_type'.format(whatfrom), [])) if getattr(record, '{}_port'.format(whatfrom), []): datastore['Port'] = list(getattr(record, '{}_port'.format(whatfrom), [])) if record.protocol: datastore['Proto'] = list(record.protocol) if datastore: idea_raw[whereto] = [datastore] if record.node_name: for node_name in record.node_name: node = {} node['Name'] = node_name if record.node_type: node['Type'] = list(record.node_type) idea_raw.setdefault('Node', []).append(node) if record.cesnet_resolvedabuses: idea_raw.setdefault('_CESNET', {})['ResolvedAbuses'] = list(record.cesnet_resolvedabuses) if record.cesnet_storagetime: idea_raw.setdefault('_CESNET', {})['StorageTime'] = record.cesnet_storagetime if record.cesnet_eventclass: idea_raw.setdefault('_CESNET', {})['EventClass'] = record.cesnet_eventclass if record.cesnet_eventseverity: idea_raw.setdefault('_CESNET', {})['EventSeverity'] = record.cesnet_eventseverity if record.cesnet_inspectionerrors: idea_raw.setdefault('_CESNET', {})['InspectionErrors'] = list(record.cesnet_inspectionerrors) try: return cls(idea_raw) except Exception as exc: print("Record:") pprint.pprint(record) print("IDEA raw:") pprint.pprint(idea_raw) raise
[docs]class IDEAFilterCompiler(pynspect.compilers.IDEAFilterCompiler): """ Custom filter compiler tailored for extended schema of :py:class:`mentat.idea.internal.Idea` messages. The implementation is based on :py:class:`pynspect.compilers.IDEAFilterCompiler` and it simply adds additional compilation directives for the extended IDEA schema. When working with :py:class:`mentat.idea.internal.Idea` messages this compiler should be used for proper rule compilations. """ def __init__(self): super(IDEAFilterCompiler, self).__init__() self.register_variable_compilation( '_CESNET.StorageTime', pynspect.compilers.compile_timeoper, pynspect.rules.ListRule )