msticpy.datamodel package

Subpackages

Submodules

msticpy.datamodel.pivot module

Pivot functions main module.

class msticpy.datamodel.pivot.Pivot(namespace: Optional[Dict[str, Any]] = None, providers: Optional[Iterable[Any]] = None, timespan: Optional[msticpy.common.timespan.TimeSpan] = None)

Bases: object

Pivot environment loader.

Instantiate a Pivot environment.

Parameters

namespace (Dict[str, Any], optional) – To search for and use any current providers, specify namespace=globals(), by default None
providers (Iterable[Any], optional) – A list of query providers, TILookup or other providers to use (these will override providers of the same type read from namespace), by default None
timespan (Optional[TimeSpan], optional) – The default timespan used by providers that require start and end times. By default the time range is initialized to be 24 hours prior to the load time.

static add_pivot_function(func: Callable[[Any], Any], pivot_reg: Optional[msticpy.datamodel.pivot_register.PivotRegistration] = None, container: Optional[str] = None, **kwargs)

Add a pivot function to entities.

Parameters

func (Callable[[Any], Any]) – The function to add
pivot_reg (PivotRegistration, optional) – Pivot registration object, by default None
container (str, optional) – The name of the container into which the function should be added, by default “other”
kwargs – If pivot_reg is not supplied you can specify required pivot registration parameters via keyword arguments. You must specify input_type (str) and entity_map (dict of entity_name, entity_attribute pairs)

msticpy.datamodel.pivot_browser module

Pivot browser widget.

class msticpy.datamodel.pivot_browser.PivotBrowser

Bases: object

Pivot enitity browser class.

Create an instance of the Pivot browser.

display(): Display the widget.

msticpy.datamodel.pivot_data_queries module

Pivot query functions class.

class msticpy.datamodel.pivot_data_queries.ParamAttrs(type, query, family, required)

Bases: tuple

Create new instance of ParamAttrs(type, query, family, required)

count(value, /): Return number of occurrences of value.

property family: Alias for field number 2

index(value, start=0, stop=9223372036854775807, /)

Return first index of value.

Raises ValueError if the value is not present.

property query: Alias for field number 1

property required: Alias for field number 3

property type: Alias for field number 0

class msticpy.datamodel.pivot_data_queries.PivQuerySettings(short_name, direct_func_entities, assigned_entities)

Bases: tuple

Create new instance of PivQuerySettings(short_name, direct_func_entities, assigned_entities)

property assigned_entities: Alias for field number 2

count(value, /): Return number of occurrences of value.

property direct_func_entities: Alias for field number 1

index(value, start=0, stop=9223372036854775807, /)

Return first index of value.

Raises ValueError if the value is not present.

property short_name: Alias for field number 0

class msticpy.datamodel.pivot_data_queries.PivotQueryFunctions(query_provider: msticpy.data.data_providers.QueryProvider, ignore_reqd: Optional[List[str]] = None)

Bases: object

Class to retrieve the queries and params from a provider.

Instantiate PivotQueryFunctions class.

Parameters

query_provider ([type]) – The query provider to load
ignore_reqd (List[str], optional) – List of parameters to ignore when building the required parameters list (e.g. [‘start’, ‘end’]), by default None

current = None

get_param_attrs(param_name: str) → List[msticpy.datamodel.pivot_data_queries.ParamAttrs]

Get the attributes for a parameter name.

Parameters: param_name (str) – Parameter name
Returns: List of ParamAttrs named tuples: (type, query, family, required)
Return type: List[ParamAttrs]

Notes

Since parameters may be defined for multiple queries, the set of parameter attributes will be returned for each query.

get_params(query_func_name: str) → Optional[msticpy.datamodel.pivot_data_queries.QueryParams]

Get the parameters for a query function.

Parameters: query_func_name (str) – Query name - the name must be fully-qualified (e.g. ‘WindowsSecurity.list_processes’)
Returns: QueryParams named tuple (all, required, full_required, param_attrs, table)
Return type: QueryParams

get_queries_and_types_for_param(param: str) → Iterable[Tuple[str, str, str, Callable[[Any], Any]]]

Get queries and parameter data types for param.

Parameters: param (str) – The parameter name.
Returns: Iterable of tuples listing: query_name, param_type, query_func
Return type: Iterable[Tuple[str, str, Callable[[Any], Any]]]

get_queries_for_param(param: str) → Iterable[Tuple[str, str, Callable[[Any], Any]]]

Get the list of queries for a parameter.

Parameters: param (str) – Parameter name
Returns: Iterable of tuples listing: query_name, query_func
Return type: Iterable[Tuple[str, str, Callable[[Any], Any]]]

get_query_pivot_settings(family: str, query: str) → msticpy.datamodel.pivot_data_queries.PivQuerySettings

Get Pivot settings metadata for a query.

Parameters

family (str) – Data family
query (str) – Query name

Returns

Named tuple:

short_name - short name for the query
direct_func_entities - the entities to add a top level function to
assigned_entities - entities to assign the query to (if parameter mapping is not applicable).

Return type

PivQuerySettings

get_query_settings(family: str, query: str) → msticpy.data.query_source.QuerySource

Get the QuerySource for the named family and query.

Parameters

family (str) – Data family name
query (str) – Query name

Returns

Query settings object

Return type

QuerySource

Raises

KeyError – If family.`query` could not be found.

class msticpy.datamodel.pivot_data_queries.QueryParams(all, required, full_required, param_attrs, table)

Bases: tuple

Create new instance of QueryParams(all, required, full_required, param_attrs, table)

property all: Alias for field number 0

count(value, /): Return number of occurrences of value.

property full_required: Alias for field number 2

index(value, start=0, stop=9223372036854775807, /)

Return first index of value.

Raises ValueError if the value is not present.

property param_attrs: Alias for field number 3

property required: Alias for field number 1

property table: Alias for field number 4

msticpy.datamodel.pivot_data_queries.add_data_queries_to_entities(provider: msticpy.data.data_providers.QueryProvider, get_timespan: Callable[[], msticpy.common.timespan.TimeSpan])

Add data queries from provider to entities.

Parameters

provider (QueryProvider) – Query provider
get_timespan (Callable[[], TimeSpan]) – Callback to get time span

msticpy.datamodel.pivot_data_queries.add_queries_to_entities(prov_qry_funcs: msticpy.datamodel.pivot_data_queries.PivotQueryFunctions, container: str, get_timespan: Callable[[], msticpy.common.timespan.TimeSpan])

Add data queries to entities.

Parameters

prov_qry_funcs (PivotQueryFunctions) – Collection of wrapped query functions
container (str) – The name of the container to add query functions to
get_timespan (Callable[[], TimeSpan]) – Function to get the current timespan.

msticpy.datamodel.pivot_magic_core module

Txt2df core code.

msticpy.datamodel.pivot_magic_core.run_txt2df(line, cell, local_ns) → pandas.core.frame.DataFrame: Convert cell text to pandas DataFrame.

msticpy.datamodel.pivot_pd_accessor module

Pandas DataFrame accessor for Pivot functions.

class msticpy.datamodel.pivot_pd_accessor.PivotAccessor(pandas_obj)

Bases: object

Pandas api extension for Pivot functions.

Instantiate pivot extension class.

display(title: Optional[str] = None, cols: Optional[Iterable[str]] = None, query: Optional[str] = None, head: Optional[int] = None) → pandas.core.frame.DataFrame

Display the DataFrame in the middle of a pipeline.

Parameters

title (str, optional) – Title to display for the DataFrame, by default None
cols (Iterable[str], optional) – List of columns to display, by default None
query (str, optional) – Query to filter the displayed data, by default None This should be a string executable by the DataFrame.query function
head (int, optional) – Limit the displayed output to head rows, by default None

Returns

Passed through input DataFrame.

Return type

pd.DataFrame

filter(expr: Union[str, numbers.Number], match_case: bool = False, numeric_col: bool = False) → pandas.core.frame.DataFrame

Filter all columns of DataFrame, return rows with any matches.

Parameters

expr (Union[str, Number]) – String or regular expression to match or a (partial) number. If expr is a string it is matched against any string or object columns using pandas str.contains(..regex=True) If expr is a number or if numeric_col is True, expr is converted to a string and matched as a substring of any numeric columns.
match_case (bool, optional) – The match is not case-sensitive by default. Set to True to force case-sensitive matches.
numeric_col (bool, optional) – If expr is a numeric string or number this will force a match against only numeric columns, by default False

Returns

The filtered dataframe

Return type

pd.DataFrame

Raises

TypeError – If expr is neither a string or number.

filter_cols(cols: Union[str, Iterable[str]], match_case: bool = False, sort_cols: bool = False) → pandas.core.frame.DataFrame

Filter output columns matching names in cols expression(s).

Parameters

cols (Union[str, Iterable[str]]) – Either a string or a list of strings with filter expressions. These can be exact matches for column names, wildcard patterns (“*” matches multiple chars and “?” matches a single char), or regular expressions.
match_case (bool, optional) – Use case-sensitive matching, by default False
sort_cols (bool, optional) – Alphabetically sort column names, by default False

Returns

The input DataFrame with only columns that match the filtering expressions.

Return type

pd.DataFrame

list_to_rows(cols: Union[str, Iterable[str]]) → pandas.core.frame.DataFrame

Expand a list column to individual rows.

Parameters: cols (Union[str, Iterable[str]]) – The columns to be expanded.
Returns: The expanded DataFrame
Return type: pd.DataFrame

parse_json(cols: Union[str, Iterable[str]]) → pandas.core.frame.DataFrame

Convert JSON string columns to Python types.

Parameters: cols (Union[str, Iterable[str]]) – Column or interable of columns to process
Returns: Processed dataframe
Return type: pd.DataFrame

run(func: Callable[[...], pandas.core.frame.DataFrame], **kwargs) → pandas.core.frame.DataFrame

Run a pivot function on the current DataFrame.

Parameters

func (Callable[..., pd.DataFrame]) – Pivot function to run
kwargs – Keyword arguments to pass to func. A column specification (e.g. column=”src_col_name”) is usually the minimum needed. For data queries the column keyword must be the name of the the query parameter (e.g. host_name = “src_col_name”)

Returns

The output DataFrame from the function.

Return type

pd.DataFrame

Notes

You can pass the join keyword argument to most pivot functions. Values for join are “inner”, “left”, “right” or “outer”.

sort(cols: Union[str, Iterable[str], Dict[str, str]], ascending: Optional[bool] = None) → pandas.core.frame.DataFrame

Sort output by column expression.

Parameters

cols (Union[str, Iterable[str], Dict[str, str]]) – If this is a string, then this should be a column name expression. A column name expression is either a column name, a case-insenstive column name or a regular expression to match one or more column names. Each column name expression can be of the format col_name_expr:desc to sort descending (col_name_expr:asc is the default). The col_name can also be a regular expression or partial column name. If this is a list, then each element should be a column name expression with an optional ‘:asc’ or ‘:desc’ suffix. If this is a dict, then the keys should be column name expressions and the values bools indication ‘ascending’ (True) or ‘descending’ (False) sort.
ascending ([type], optional) – Overrides any ordering specified for individual columns and sorts ‘ascending’ if True or ‘descending’ if False. If not supplied and no column-specific ordering is supplied it sorts ascending.

Returns

The sorted DataFrame

Return type

pd.DataFrame

Raises

ValueError – One or more column expressions matched no column name in the input.

tee(var_name: str, clobber: bool = False) → pandas.core.frame.DataFrame

Save current dataframe to var_name in the IPython user namespace.

Parameters

var_name (str) – The name of the DF variable to create.
clobber (bool, optional) – Whether to overwrite an existing variable of the same name, by default False

Returns

Passed through input DataFrame.

Return type

pd.DataFrame

Notes

This function only works in an IPython/Jupyter notebook environment. It will attempt to create a variable in the user local namespace that references the current state of the DataFrame in the pipeline.

By default it will not overwrite an existing variable of the same name (specify clobber=True to overwrite)

tee_exec(df_func: str, *args, **kwargs) → pandas.core.frame.DataFrame

Run a dataframe method on the dataframe without changing it.

Parameters

df_func (str) – The name of the function to execute. Accessor methods must be of the form “accessor.method”.
args (tuple) – Positional arguments to be passed to the function
kwargs (dict) – Keyword arguments to be passed to the function.

Returns

Passed through input DataFrame.

Return type

pd.DataFrame

Notes

This function runs the DataFrame method or accessor function. It does not alter the DataFrame (unless the function does any kind of in-place modification). The function is run and the original input DataFrame is returned.

msticpy.datamodel.pivot_pipeline module

Pivot pipeline class.

class msticpy.datamodel.pivot_pipeline.Pipeline(name: str, description: Optional[str] = None, steps: Optional[Iterable[msticpy.datamodel.pivot_pipeline.PipelineStep]] = None)

Bases: object

Pivot pipeline.

Create Pipeline instance.

Parameters

name (str) – The pipeline name.
description (Optional[str]) – The pipeline description, by default None.
steps (Optional[Iterable[PipelineStep]]) – Pipeline steps, by default None.

classmethod from_yaml(yml_str: str) → Iterable[msticpy.datamodel.pivot_pipeline.Pipeline]

Parse pipelines from yaml string.

Parameters: yml_str (str) – Yaml dict of pipelines.
Yields: Pipeline – Iterable of pipeline instances

classmethod parse_pipeline(pipeline: Dict[str, Dict[str, Any]]) → msticpy.datamodel.pivot_pipeline.Pipeline

Parse single pipeline from dictionary.

Parameters: pipeline (Dict[str, Dict[str, Any]]) – Single pipeline as a dictionary: {name: {pipeline_dict…}}.
Returns: The pivot pipeline.
Return type: Pipeline
Raises: ValueError – The dictionary could not be parsed as a pipeline.

static parse_pipelines(pipelines: Dict[str, Dict[str, Any]]) → Iterable[msticpy.datamodel.pivot_pipeline.Pipeline]

Parse dict of pipelines.

Parameters: pipelines (Dict[str, Dict[str, Any]]) – Dict of pipelines.
Yields: Pipeline – Iterable of pipeline instances

print_pipeline(df_name: str = 'input_df', comments: bool = True) → str

Return the pipeline as text that can be executed in Python.

Parameters

df_name (str, optional) – Name of the input dataframe to be used in the returned code, by default “input_df”
comments (bool, optional) – If True show step comments, by default True

Returns

The executable pipeline text.

Return type

str

run(data: pandas.core.frame.DataFrame, verbose: bool = True, debug: bool = False) → Optional[Any]

Run the pipeline on the supplied DataFrame.

Parameters

data (pd.DataFrame) – Input DataFrame for pipeline
verbose (bool, optional) – If True, report progress, by default True
debug (bool, optional) – If True, report more detailed progress, by default False

Returns

The output of the last stage of the pipeline

Return type

Any

to_yaml() → str

Return yaml representation of pipeline.

Returns: Pipeline as yaml.
Return type: str

class msticpy.datamodel.pivot_pipeline.PipelineExecStep(accessor, pos_params, params, text, comment)

Bases: tuple

Create new instance of PipelineExecStep(accessor, pos_params, params, text, comment)

property accessor: Alias for field number 0

property comment: Alias for field number 4

count(value, /): Return number of occurrences of value.

index(value, start=0, stop=9223372036854775807, /)

Return first index of value.

Raises ValueError if the value is not present.

property params: Alias for field number 2

property pos_params: Alias for field number 1

property text: Alias for field number 3

class msticpy.datamodel.pivot_pipeline.PipelineStep(name: str, step_type: str, function: Optional[str] = None, entity: Optional[str] = None, comment: Optional[str] = None, pos_params: List[str] = NOTHING, params: Dict[str, Any] = NOTHING)

Bases: object

Pivot pipeline step class.

Method generated by attrs for class PipelineStep.

comment: Optional[str]

entity: Optional[str]

function: Optional[str]

get_exec_step() → msticpy.datamodel.pivot_pipeline.PipelineExecStep

Return the executable step details.

Returns: Named tuple with the following fields accessor - the name of the pandas DataFrame accessor function params - parameters to be passed to the function text - the text representation of the accessor + params comment - optional comment that can be used by the pipeline builder to add Python comments to output.
Return type: PipelineExecStep

name: str

params: Dict[str, Any]

pos_params: List[str]

step_type: str

msticpy.datamodel.pivot_register module

Pivot helper functions .

class msticpy.datamodel.pivot_register.PivotRegistration(input_type: str, entity_map: Dict[str, str], func_df_param_name: Optional[str] = None, func_out_column_name: Optional[str] = None, func_df_col_param_name: Optional[str] = None, func_new_name: Optional[str] = None, src_module: Optional[str] = None, src_class: Optional[str] = None, src_func_name: Optional[str] = None, can_iterate: bool = True, func_static_params: Optional[Dict[str, Any]] = None, func_input_value_arg: Optional[str] = None, src_config_path: Optional[str] = None, src_config_entry: Optional[str] = None, entity_container_name: Optional[str] = None, return_raw_output: bool = False, create_shortcut: bool = False)

Bases: object

Pivot registration for function.

Notes

src_modulestr: The src_module to import
src_classstr, optional: class to import and instantiate that contains the function/method (not needed if the target function is a pure Python function)
src_func_name: Callable: The function to wrap.
func_new_name: str, optional: Rename the function to this, defaults to src_func_name
input_typestr: The input data type that the function is expecting. One of ‘dataframe’, ‘iterable’, ‘value’
can_iterate: bool, optional: True if the function supports being called multiple times (for iterable input). Default is True
entity_map: Dict[str, str]: dict of entities supported (keys) and attribute to use from entity as input to the function
func_df_param_name: str: The name of the parameter that func takes the input value e.g. func(ip=my_address) => ‘ip’ == func_df_col_param_name. In the case of a DataFrame, this is usually ‘data’
func_df_col_param_name: str: The name that the target function uses to identify the column to use for input in the input DataFrame.
func_out_column_name: str, optional: The name of the column in the output DF to use as a key to join to the input. If None, use func_df_col_param_name
func_static_params: Optional[Dict[str, Any]]: static parameters (kwargs) that are always passed to the target function
func_input_value_arg: Optional[str]: The name of kwarg passed to the function that contain the input value. If function supports DF input, func_df_col_param_name will be used and this is not needed.
src_config_pathOptional[str]: The source path that the configuration was read from, default None.
src_config_entryOptional[str]: The entry name in the configuration file, default None.
entity_container_nameOptional[str]: The name of the container in the entity that will hold this pivot function.
return_raw_outputbool: Return raw output from the wrapped function, do not try to format into a DataFrame. Default is False.
create_shortcutbool: If True, create a shortcut function directly on the entity.

Method generated by attrs for class PivotRegistration.

attr_for_entity(entity: Union[msticpy.datamodel.entities.entity.Entity, str]) → Optional[str]

Return the attribute to use for the specified entity.

Parameters: entity (Union[entities.Entity, str]) – Entity instance or name
Returns: Attribute name to use.
Return type: Optional[str]

can_iterate: bool

create_shortcut: bool

entity_container_name: Optional[str]

entity_map: Dict[str, str]

func_df_col_param_name: Optional[str]

func_df_param_name: Optional[str]

func_input_value_arg: Optional[str]

func_new_name: Optional[str]

func_out_column_name: Optional[str]

func_static_params: Optional[Dict[str, Any]]

input_type: str

return_raw_output: bool

src_class: Optional[str]

src_config_entry: Optional[str]

src_config_path: Optional[str]

src_func_name: Optional[str]

src_module: Optional[str]

msticpy.datamodel.pivot_register.create_pivot_func(target_func: Callable[[Any], Any], pivot_reg: msticpy.datamodel.pivot_register.PivotRegistration) → Callable[[...], pandas.core.frame.DataFrame]

Create function wrapper for pivot function.

Parameters

target_func (Callable) – The target function to wrap.
pivot_reg (PivotRegistration) – The pivot function registration object.

Returns

The original target_func wrapped in pre-processing and post-processing code.

Return type

Callable[[Any], pd.DataFrame]

msticpy.datamodel.pivot_register.get_join_params(func_kwargs: Dict[str, Any]) → Tuple[Optional[str], Optional[str], Optional[str], bool]

Get join parameters from kwargs.

Parameters: func_kwargs (Dict[str, Any]) – Keyword arguments from caller
Returns: join_type, left_on, right_on, join_ignore_case
Return type: Tuple[str, str, str, bool]

msticpy.datamodel.pivot_register.join_result(input_df: pandas.core.frame.DataFrame, result_df: pandas.core.frame.DataFrame, how: str, left_on: str, right_on: str, ignore_case: bool) → pandas.core.frame.DataFrame

Join input and result DFs, optionally ignoring case.

Parameters

input_df (pd.DataFrame) – Input DF
result_df (pd.DataFrame) – Result DF
how (str) – Join type - “inner”, “left”, “right”, “outer”
left_on (str) – Column from input_df to use as join key
right_on (str) – Column from result_df to use as join key
ignore_case (bool) – If True and input_df column is a string

Returns

The merged DataFrame

Return type

pd.DataFrame

msticpy.datamodel.pivot_register_reader module

Reads pivot registration config files.

msticpy.datamodel.pivot_register_reader.add_unbound_pivot_function(func: Callable[[Any], Any], pivot_reg: Optional[msticpy.datamodel.pivot_register.PivotRegistration] = None, container: str = 'other', **kwargs)

Add a pivot function to entities.

Parameters

func (Callable[[Any], Any]) – The function to add
pivot_reg (PivotRegistration, optional) – Pivot registration object, by default None
container (str, optional) – The name of the container into which the function should be added, by default “other”
kwargs – If pivot_reg is not supplied you can specify required pivot registration parameters via keyword arguments. You must specify input_type (str) and entity_map (dict of entity_name, entity_attribute pairs)

msticpy.datamodel.pivot_ti_provider module

Pivot TI Provider helper functions.

msticpy.datamodel.pivot_ti_provider.add_ioc_queries_to_entities(ti_lookup: msticpy.sectools.tilookup.TILookup, container: str = 'ti', **kwargs)

Add TI functions to entities.

Parameters

ti_lookup (TILookup) – TILookup instance.
container (str) – The name of the container to add query functions to

msticpy.datamodel.pivot_ti_provider.create_ti_pivot_funcs(ti_lookup: msticpy.sectools.tilookup.TILookup): Create the TI Pivot functions.

msticpy.datamodel.pivot_ti_provider.register_ti_pivot_providers(ti_lookup: msticpy.sectools.tilookup.TILookup, pivot: Pivot): Register pivot functions from TI providers.

Module contents

Package file for datamodel subpackage.