Workflows package

This package contains the data-structure representing the workflow graph and the modules in it.

The sub-packages are specialisations, in this package the basic graph representation is implemented.

Apperance dataclass

Value object with attributes that are only used in the visual editor.

cssClass instance-attribute

cssClass: str

CSS classes that the current node has in the visual editor.

nodeUid instance-attribute

nodeUid: str | None

Unique ID in the DOM set by the visual editor. Sometimes it's apparently None for unknown reasons.

pos instance-attribute

pos: Tuple[float, float]

Tuple with x/y coordinates in the canvas where the current node is displayed.

typenode instance-attribute

typenode: bool

Unclear what exactly that is. Hardcoded to False for now since it's like this everywhere.

BlueprintGraph dataclass

Bases: Graph

This graph represents a blueprint. Blueprints are reusable parts of a workflow graph and MUST NOT have a trigger.

Whether that's the case can be enforced with the BlueprintGraph.check method.

check

check() -> GraphValidationResult

Custom check method for blueprint graph.

ConfigurationError dataclass

Bases: GraphError

Generic "configuration error" class. Useful to e.g. reject wrong parameters from the visual editor.

context instance-attribute

context: str

Error message.

node instance-attribute

node: Node

Reference to the misconfigured node.

CyclicGraphError dataclass

Bases: GraphError

If a cycle was found in the graph it is invalid. This class represents that error.

cycle_path instance-attribute

cycle_path: List[Tuple[Node, int, Node, int]]

List of edges (from, output_port, to, input_port) that create a cycle.

Frame dataclass

Value object representing a frame in the visual editor. Only for visualization.

clazz instance-attribute

clazz: str

Additional CSS classes for the frame. class is a reserved keyword.

nodes instance-attribute

nodes: List[Node]

List of (references) to nodes that are visually encapsulated in the frame in the visual editor.

text instance-attribute

text: str

Title of the frame

uuid instance-attribute

uuid: UUID

Unique identifier of the frame. Used by the visual editor.

Graph dataclass

Bases: ABC

A representation of a workflow graph. Consists of

frames instance-attribute

frames: Iterable[Frame]

List of frame objects in the visual editor.

nodes instance-attribute

nodes: Dict[int, Trigger | Module]

A list of all nodes inside the graph. Edges are represented by nodes having references to other nodes in their inputs/outputs.

root instance-attribute

root: Trigger | Module

Reference to the root of the graph.

check abstractmethod

check() -> GraphValidationResult

Checks if the graph's structure is valid. Works as described in GraphValidationResult.

initialize_graph_modules async

initialize_graph_modules(db: AsyncSession) -> None

This method is declared in mmisp.workflows.modules, but is a method of is part of Graph. It injects db into each module. This is done to avoid circular import situations.

Note

For now, this is necessary since modules may require other objects from the database. E.g. the ModuleOrganisationIf loads the names of all organisations.

There are a few ways forward:

• Implement a more sophisticated SQLAlchemy type that allows to query other entities while placing the JSON into the Graph structure.

• Factor the :attribute:mmisp.workflows.modules.Module.params structure out and inject it into the workflow editor on its own.

For modernizing legacy MISP's workflows and retaining backwards compatibility, just injecting the DB into each node at the beginning is the simplest solution though.

Parameters:

Name	Type	Description	Default
db	AsyncSession	SQLAlchemy DB session.	required

GraphError

Bases: ABC

Abstract base-class for validation errors in the graph structure. The idea is to display error kinds via

match error:
    case CyclicGraphError(path):
        pass
    case MultipleEdgesPerOutput(affected):
        pass
    # ...

GraphValidationResult dataclass

Accumulator for all validation results. The idea is to run graph-level checks such as cycle-detection in Graph.check and then run node-level checks in Node.check. Whenever an error is encountered, it's added into this class.

errors instance-attribute

errors: List[GraphError]

Fatal errors in the workflow graph. The graph is unusable like this.

add_result

add_result(other: Self) -> None

Merges another validation result into this accumulator object.

Parameters:

Name	Type	Description	Default
other	Self	Another validation result added in here.	required

is_valid

is_valid() -> bool

If neither errors nor warnings are added, the graph is valid.

InconsistentEdgeBetweenAdjacencyLists dataclass

Bases: GraphError

In MISP the standard for storing the graph is in duplicated form, the graph is stored once as an adjacency list of outgoing edges and in parallel a second time as an adjacency list of incoming edges. In case there is an edge from node A to node B, that is registered as outgoing from node A to node B and not as incoming from node A to node B or vica-versa, this is an inconsistency error that will be returned with an instance of this class.

from_ instance-attribute

from_: Node

The starting node of the inconsistent edge.

input_port_id instance-attribute

input_port_id: int

The input port of the inconsistent edge.

output_port_id instance-attribute

output_port_id: int

The output port of the inconsistent edge.

to instance-attribute

to: Node

The ending node of the inconsistent edge.

MissingTrigger

Bases: GraphError

Each workflow that isn't a blueprint must have a trigger as starting node. A violation of that is described with this class.

Note

Incidentally, in legacy MISP this isn't part of checkGraph, but only used in the validation rules of the Workflow model.

In MMISP, both validations are unified.

Module dataclass

Bases: WorkflowNode

A module is a workflow node that is either

• an action, i.e. performs an operation with an observable effect
• a logic module, i.e. forwards to a different module based on a condition
• a filter, i.e. manipulates the WorkflowInput.

A module implementation can be provided by writing a subclass that assigns values to at least id, version, name.

blocking class-attribute instance-attribute

blocking: bool = False

Whether or not the module is blocking, i.e. can abort the workflow if it fails. If it's not blocking and fails, execution continues.

configuration instance-attribute

configuration: ModuleConfiguration

Values for the params required by this module.

description class-attribute instance-attribute

description: str = ''

Human-readable description of the node. Must be here because Modules and triggers have a description.

html_template class-attribute instance-attribute

html_template: str = ''

HTML template provided by the visual editor to use. To be declared in the subclass implementing a concrete module.

icon class-attribute instance-attribute

icon: str = 'envelope'

Frontend icon. Must be here because Modules and triggers have a description.

on_demand_filter instance-attribute

on_demand_filter: Optional[Filter]

Some modules allow filtering of data internally without modifying WorkflowInput. The filter used for that is defined by this attribute.

on_demand_filtering_enabled class-attribute instance-attribute

on_demand_filtering_enabled: bool = False

Whether internal filtering is enabled.

previous_version class-attribute instance-attribute

previous_version: str = '?'

Previously used version of the module.

template_params class-attribute instance-attribute

template_params: List[str] = field(
    default_factory=lambda: []
)

List containing all params that are jinja2 templates.

exec async

exec(
    payload: WorkflowInput, db: AsyncSession
) -> Tuple[bool, Union[Module, None]]

Executes the module using the specific payload given by the workflow that calls the execution of the module. Execution strongly depends on the type of the module that is executed.

The first component of the tuple indicates whether execution was successful. The second component of the tuple is the next node to be executed.

Since this library allows arbitrarily many inputs & outputs, we cannot infer the next module from the success of the execution of this module (relevant for e.g. if/else).

The default implementation assumes exactly one output and returns it on success.

Parameters:

Name	Type	Description	Default
payload	WorkflowInput	The workflows input for the specific module execution.	required
db	AsyncSession	Database handle for write operations.	required

initialize_for_visual_editor async

initialize_for_visual_editor(db: AsyncSession) -> None

Initializes the parameters for a module. Done in a method since that may involve further DB operations.

Only needed for the visual editor, not the execution since it defines which params are expected and which values are accepted (e.g. all attributes stored in the database).

Parameters:

Name	Type	Description	Default
db	AsyncSession	SQLAlchemy session	required

is_initialized_for_visual_editor

is_initialized_for_visual_editor() -> bool

Checks if the module was initialized for the visual editor which happens by calling [Module.initialize][mmisp.workflows.modules.Module.initialize]. It's expected that the attribute params will be set by this method.

MultipleEdgesPerOutput dataclass

Bases: GraphError

A Node has multiple inputs and outputs. Unless explicitly allowed, each output may only have a single connection to another node. An example for that is ModuleConcurrentTask.

This class represents the error if that's not allowed.

affected instance-attribute

affected: Tuple[Node, int]

All edges affected by the problem: first component of the tuple is the node with too many connections per output, the second component is the ID of the affected output.

Node dataclass

Bases: ABC

A single node in a graph. Can be a trigger - something that caused the workflow run - or a module - a behavioral component in the workflow graph.

Inputs & outputs can be thought of like a layer 4 address where the node is the address and the input/output is the port number. So an edge doesn't just connect two nodes together, but tuples of (output number, input) and (input number, output). This is needed because

• The concurrent execution requires just a single output, but with multiple edges to other modules from it.
• An If/Else module has two outputs: one node to execute if the condition is true and one if it isn't.

It might've been possible to infer outputs by the order of the connections, but this would've lead to ambiguity because

• If >1 module is connected to this module, it's not clear whether there are multiple connections leaving the same output (as in the concurrency module) or if each module has its own meaning (as in the if/else module).
• It's valid to produce an if/else module with only else connecting to another node. In that case the first output in the list should actually be encoded as second list element since the first is empty.

Because of that, infering inputs/outputs from the order seemed too error-prone and the legacy MISP structure was reused here.

enable_multiple_edges_per_output class-attribute instance-attribute

enable_multiple_edges_per_output: bool = False

Whether it's OK to have multiple edges going from a single output. See Node for more context. Used by e.g. the concurrent tasks module.

graph_id instance-attribute

graph_id: int

The id with which this node was identified in the json data, that was used to initialize its parent graph.

inputs instance-attribute

inputs: Dict[int, List[NodeConnection]]

References to Node objects connecting to this node. Grouped by the ID of the input.

n_inputs class-attribute instance-attribute

n_inputs: int = 1

Number of allowed inputs. If inputs exceeds this, the Node.check method will return an error for this.

n_outputs class-attribute instance-attribute

n_outputs: int = 1

Number of allowed outputs. If outputs exceeds this, the Node.check method will return an error for this.

outputs instance-attribute

outputs: Dict[int, List[NodeConnection]]

References to Node objects that are connected to this node. Grouped by the ID of the output.

The data-structure is recursive here (basically a doubly-linked list) using references to make walking the graph as easy as possible.

supported class-attribute instance-attribute

supported: bool = True

Indicator whether the module / trigger is supported. All items from legacy MISP exist here to allow loading any workflow, but not all are actually supported by the application.

__eq__

__eq__(other: object) -> bool

Custom comparison if two instances are equal, needed for our special hash function so that the statement eq(a, b) = True => hash(a) == hash(b) is valid for every pair of nodes a nd b, which guaranties that the Dict (HashMap) will not malfunction.

Parameters:

Name	Type	Description	Default
other	object	A reference to the other object to be compared.	required

__hash__

__hash__() -> int

Our custom node hash function which allows to store nodes in a Dict (HashMap).

check

check() -> GraphValidationResult

Checks if the module is correctly configured. At Node level 3 conditions are checked (and errors returned if not fulfilled): 1(2). The registered input(output) ports are in the range from 1 to 'n_inputs'('n_outputs') incl. It is allowed to have less registered input(output) ports than the specified maximum amount. 3. In case 'enable_multiple_edges_per_output' has a value of false, each outgoing port has maximum 1 outgoing edge. (It is allowed for an outgoing port to have no outgoing edges.)

Trigger dataclass

Bases: WorkflowNode

A trigger represents the starting point of a workflow. It can have only one output and no inputs. Each trigger is represented by this class, the attributes are loaded from the DB.

Legacy MISP provides subclasses for each triggers, but since the properties were saved into the database anyways and no behavior was added to those classes, the idea was dropped entirely in MMISP.

description class-attribute instance-attribute

description: str = ''

Human-readable description of when the trigger gets executed.

disabled class-attribute instance-attribute

disabled: bool = False

Whether or not the trigger is disabled, i.e. ignored when triggered.

icon class-attribute instance-attribute

icon: str = 'envelope'

Frontend icon.

overhead instance-attribute

overhead: Overhead

Indicates the expensiveness/overhead of the trigger as indicated by the Overhead enum.

overhead_message class-attribute instance-attribute

overhead_message: str = ''

Explanatory message describing why the overhead level was chosen for this trigger.

scope instance-attribute

scope: str

Scope the trigger operates on. E.g. event or attribute.

normalize_data async

normalize_data(
    db: AsyncSession, input: VerbatimWorkflowInput
) -> RoamingData

Allows triggers to perform custom "normalization" operations before handing over to the actual modules.

Workflow input can be an arbitrary JSON already, but it can also be a SQLAlchemy model. In the latter case it's this method's responsibility to transform it into MISP compliant data.

Parameters:

Name	Type	Description	Default
db	AsyncSession	DB handle to load more entities from the database.	required
input	VerbatimWorkflowInput	Workflow input to modify.	required

UnsupportedWorkflow dataclass

Bases: GraphError

Not all modules are implemented in modern MISP. All of them exist though to allow reading any workflow JSON here.

If a workflow relies on unsupported modules, this error class is used.

module_graph_id instance-attribute

module_graph_id: int

ID of the unsupported module

WorkflowGraph dataclass

Bases: Graph

This graph represents a workflow. That means that it MUST have a trigger as starting node and no trigger anywhere else.

WorkflowNode dataclass

Bases: Node

Dataclass with shared properties for both modules & triggers, but not relevant for the abstract Node type.

apperance instance-attribute

apperance: Apperance

Value object with miscellaneous settings for the visual editor.

blocking instance-attribute

blocking: bool

If the workflow of a blocking trigger fails, the actual operation won't be carried out. For instance, if the "event before save"-trigger fails, the event will not besaved.

For modules, blocking and failing modules will terminate the workflow execution.

id instance-attribute

id: str

Unique identifier of the module/trigger. To be declared in the subclass implementing a concrete module/trigger.

name instance-attribute

name: str

Human-readable identifier of the module/trigger. To be declared in the subclass implementing a concrete module/trigger.

version class-attribute instance-attribute

version: str = '0.1'

Current version of the module. To be declared in the subclass implementing a concrete module.

Models related to the execution of workflows.

execute_workflow async

execute_workflow(
    workflow: Workflow,
    user: User,
    input: VerbatimWorkflowInput,
    db: AsyncSession,
) -> Tuple[bool, List[str]]

Provides the functionality for executing a workflow, which consists of traversing the given workflow graph and its modules and executing these modules with their specific configurations.

Note

Legacy MISP allows non-blocking paths, i.e. multiple "roots" & no termination of the workflow execution if one of the paths fails.

This "feature" is left out entirely here: this would not only break the assumption of having a single root, but also complicate the execution code further.

This is only implemented for concurrent tasks, however

• A job will be exposed in worker that allows to resume execution of a workflow at a given node. This is triggered for each of the concurrent nodes.

• No intertwined state since all the state of an execution is carried around via the payload.

Parameters:

Name	Type	Description	Default
workflow	Workflow	The Graph representation of the workflow to be executed.	required
input	VerbatimWorkflowInput	Initial payload for the workflow.	required
db	AsyncSession	SQLAlchemy session.	required

walk_nodes async

walk_nodes(
    input: WorkflowInput,
    current_node: Module,
    workflow: Workflow,
    db: AsyncSession,
    jinja2_engine: Environment,
) -> Tuple[bool, List[str]]

Recursive graph walker implementation starting at a given node. Used by the workflow execution itself, but can also be used to resume workflow execution, e.g. for concurrent modules that schedule jobs with the successor nodes.

Parameters:

Name	Type	Description	Default
input	WorkflowInput	Workflow payload for current_node.	required
current_node	Module	Node to resume execution with.	required
workflow	Workflow	Workflow entity. Used for logging.	required
db	AsyncSession	Database session.	required
jinja2_engine	Environment	Instantiated templating engine to substitute placeholders in module configuration with values from the payload.	required

Data structure for the payload passed to the workflow and filtering mechanism associated with it.

EvaluateImplementation

Bases: Enum

Enum representing all EvaluateCondition implementations in Legacy MISP.

Filter dataclass

The data passed to a workflow can be filtered on-demand. That means, all entries from the dict may be removed that don't match a condition.

The condition that needs to match is represented by this object.

There are two ways these filters can be applied:

• via the ModuleGenericFilterData in place. Can be undone by adding ModuleGenericFilterReset later on.

• via a module with on_demand_filtering_enabled set to True. The filtering must be called by the module itself in that case.

operator instance-attribute

operator: Operator

Operator to compare the item below path against value.

path instance-attribute

path: str

Attribute path in the list where each item will be checked against value using operation operator.

selector instance-attribute

selector: str

Attribute path pointing to a list inside the WorkflowInput. In this list, each element below attribute-path path will be checked against value using operation operator.

For instance given the structure

{
    "foo": [
        {
            "bar": "lololo"
        },
        {
            "bar": "lalala"
        },
    ]
}

a filter with selector being foo, path being bar, value being lololo and operator being not_equal would result in

{
    "foo": [
        {
            "bar": "lalala"
        }
    ]
}

Path must be a CakePHP hash path since existing legacy MISP filters are using that format as well.

Note

MMSIP filter currently only support limited hash path functionality.

Supported features are the dot-separated paths consisting of keys and '{n}' indicating iterating over a list or a dictionary with numeric keys.

Additional hash path functionality such as Matchers could be added to MMISP later.

value instance-attribute

value: str | List[str]

Value to compare against. Can be a list for operations in/not_in or a string value for equals/etc..

match_value

match_value(value: Any) -> bool

Check if a value matches a filter.

Parameters:

Name	Type	Description	Default
value	Any	The value to check.	required
filter		The filter to match against.	required

Returns:

Type	Description
bool	True if the value matches the filter, False otherwise.

FilterError

Bases: Exception

Abstract class representing possible invalid Filter inputs

InvalidOperationError dataclass

Bases: FilterError

If the operation passed to the filter is invalid this error is returned. e.g.: Operation.to_str fails to return a valid operation.

InvalidPathError dataclass

Bases: FilterError

If the path passed to the filter is invalid this error is returned. e.g.: empty String

InvalidSelectionError dataclass

Bases: FilterError

If the selection passed to the filter is invalid this error is returned. Examples for invalid selections are datatypes other from String, hashpahts, that dont lead to a list or empty strings

InvalidValueError dataclass

Bases: FilterError

If the value passed to the filter is invalid this error is returned. e.g.: input not of type str or List[str]

Operator

Bases: Enum

Enum representing possible filter operations.

from_str classmethod

from_str(input: str) -> Self

Returns a member of this enum given the string representation of the operator.

Parameters:

Name	Type	Description	Default
input	str	string representation of the operator.	required

WorkflowInput

When a workflow gets executed, it gets input data. For instance, the "after event save"-workflow gets a dictionary containing the event that was saved.

Additionally, filters can be added by modules. That way, subsequent modules will only see a filtered subset of the workflow data. This operation can be undone.

Filters are expressed using Filter class.

data property

data: RoamingData | List[RoamingData] | None

Returns either all of the data given to the workflow input OR a list with filter results if a filter was added using WorkflowInput.add_filter.

add_filter

add_filter(label: str, filter: Filter) -> None

Adds another Filter to the workflow input.

Parameters:

Name	Type	Description	Default
filter	Filter	Filter to be added.	required

reset_filters

reset_filters() -> None

Removes all filters from the workflow input. WorkflowInput.data will contain all of the data it has instead of a filtered portion now.

evaluate_condition

evaluate_condition(
    left: Any | List[Any],
    operator: Operator,
    right: Any | List[Any],
    impl: EvaluateImplementation = EvaluateImplementation.LEGACY_IF_ELSE,
) -> bool

A utility method for performing comparisons between the specified data.

Parameters:

Name	Type	Description	Default
left	Any | List[Any]	The first operand.	required
operator	Operator	The operator to be used in the comparison.	required
right	Any | List[Any]	The second operand.	required
impl	EvaluateImplementation	Which legacy MISP implementation of evaluate condition to be used: the if/else or the filter.	LEGACY_IF_ELSE

Returns: Whether the comparison holds or not.

extract_path

extract_path(path: List[str], data: Any) -> List[Any]

A utility method providing the Hash::extract functionality in CakePHP.

Parameters:

Name	Type	Description	Default
path	List[str]	The hash path to extract data from.	required
data	Any	The container to extract data from.	required

Returns: A list of the extracted data, if data matching the hash path was found, an empty List otherwise.

get_path

get_path(path: List[str], data: Any) -> Any

A utility method providing the Hash::get functionality in CakePHP.

Parameters:

Name	Type	Description	Default
path	List[str]	The hash path to extract data from.	required
data	Any	The container to extract data from.	required

Returns: The extracted data, if data matching the hash path was found, a None pointer otherwise.

This module contains subclasses for different node-types, i.e. triggers and modules and implementations for modules that were bundled with legacy MISP.

ModuleAction dataclass

Bases: Module

Marker class representing an action module. Not relevant for the behavior, but for the HTTP responses to determine which kind of module this is.

ModuleConcurrentTask dataclass

Bases: ModuleLogic

Accepts multiple connecting nodes and executes all of them concurrently.

ModuleConfiguration dataclass

Parameters for a module. If a module defines a textarea with ID foobar as param, this class expects a dictionary

{
    "foobar": "mytext"
}

These params are defined in the visual editor and thus saved together with the module.

data instance-attribute

data: Dict[str, List[str] | str | bool]

The dictionary containing values for the parameters a module needs.

validate

validate(structure: ModuleParams) -> List[str]

Check if the parameters specified here are correct. For e.g. a select-param with id "foobar", it will be checked if data["foobar"] is among the options provided by the select.

Parameters:

Name	Type	Description	Default
structure	ModuleParams	The module param definitions to validate the configuration against.	required

ModuleGenericFilterData dataclass

Bases: ModuleFilter

Configure a filter on the workflow payload. Every subsequent module will only see the filtered version unless the effect gets reversed with ModuleGenericFilterReset.

ModuleGenericFilterReset dataclass

Bases: ModuleFilter

Resets all filters declared for the workflow payload.

ModuleIf dataclass

Bases: ModuleLogic

The class on which all if/else module implementations will be based.

ModuleLogic dataclass

Bases: Module

Marker class representing a logic module. Not relevant for the behavior, but for the HTTP responses to determine which kind of module this is.

ModuleOrganisationIf dataclass

Bases: ModuleIf

Module allowing to check if the organistaion property of the payload matches a condition.

ModuleParam dataclass

Each module can be configured in the visual editor, e.g. an if/else module needs an operator, a value to check against and an attribute-path to a the value to check against. All of these values are configurable.

This class represents a single parameter passed to a module.

id instance-attribute

id: str

Identifier for the parameter. Must be unique in the context of a single module.

jinja_supported class-attribute instance-attribute

jinja_supported: bool = False

If True, the input from the visual editor for this parameter is a jinja2 template. A template gets the WorkflowInput data as input.

kind instance-attribute

kind: ModuleParamType

Which type of input is expected. Denoted by ModuleParamType.

label instance-attribute

label: str

Human-readable label in the visual editor for this parameter.

options instance-attribute

options: Dict[str, Any]

Additional options passed to the visual editor. The other options are useful for e.g. validation of actual workflow inputs. All the other parameters (e.g. placeholder) are passed into this dictionary.

ModuleParamType

Bases: Enum

This enum provides supported form fields in the visual editor to configure a parameter represented by ModuleParam for a module.

Overhead

Bases: Enum

Represents overhead of a module. That means e.g. how often it will be executed on a typical MISP installation.

from_int classmethod

from_int(input: int) -> Self

Returns a member of this enum given the int representation of the overhead.

Parameters:

Name	Type	Description	Default
input	int	numeric representation of the overhead.	required

workflow_node

workflow_node(
    cls: Type[Module | Trigger],
) -> Type[Module | Trigger]

Annotation that registers the annotated class in the [NodeRegistry][mmisp.workflows.modules.NodeRegistry]. That way modules & triggers are registered in the workflow application.

MISP workflows legacy module

All other modules intended to be agnostic of old legacy MISP formats as much as possible and can be refactored gradually into a more reasonable structure.

This is the compat-layer with legacy MISP: it understands the old JSON and imports/exports that into the new data structure.

GraphFactory

This class is responsible for creating a MMSIP graph object from the legacy MISP JSON format. The class can also convert a graph in the legacy MISP JSON format from the DB to a modern graph object.

graph2jsondict classmethod

graph2jsondict(graph: Graph) -> Dict[str, Any]

Converts a modern MISP graph object to a graph in the legacy MISP JSON format used in the DB.

Parameters:

Name	Type	Description	Default
graph	Graph	Graph object to convert to JSON.	required

jsondict2graph classmethod

jsondict2graph(input: Dict[str, Any]) -> Graph

Converts a graph in the legacy MIPS JSON format to a graph object. Returns the input graph as a modern MISP graph object.

Parameters:

Name	Type	Description	Default
input	Dict[str, Any]	JSON dictionary containing the graph information.	required

GraphValidation

This class checks whether a graph is valid or not.

report classmethod

report(result: GraphValidationResult) -> CheckGraphResponse

Reports the results of the graph validation. Returns a CheckGraphResponse object containing all the results. NB! The current API endpoint implementation does not allow multiple cycles to be returned, therefore only the first found cycle will be returned. others are discarded. To be changed in the future when the API endpoint is updated. The PathWarning system is deprecated and not used in the modern MISP, therefore in the JSON will be always set that there are no PathWarnings.

Parameters:

Name	Type	Description	Default
result	GraphValidationResult	GraphValidationResult object containing the results of the graph validation.	required

report_as_str classmethod

report_as_str(
    result: GraphValidationResult, graph: WorkflowGraph
) -> str

Turns validation report into a string. Used for legacy /workflows/edit. The frontend expects a single error message here in a modal in the top left corner, so all errors are kept brief on purpose.

Parameters:

Name	Type	Description	Default
result	GraphValidationResult	Validation report.	required
graph	WorkflowGraph	Graph that got validated.	required

JSONGraphType

Bases: UserDefinedType

SQLAlchemy type used to store and retrieve graph objects in the database. This type handles the conversion between a modern MISP graph object and its JSON representation for the compatibility with legacy MISP.

bind_processor

bind_processor(dialect: Any)

Method for processing data before storing it in the database.

Parameters:

Name	Type	Description	Default
dialect	Any	SQLAlchemy dialect being used.	required

get_col_spec

get_col_spec(**kw) -> str

Returns the colum specification for the custom SQLAlchemy type in LONGTEXT.

result_processor

result_processor(dialect: Any, coltype: Any)

Defines how to process data retrieved from the database. Converts the JSON string stored in the database back into a graph object using the GraphFactory's jsondict2graph method.

Parameters:

Name	Type	Description	Default
dialect	Any	SQLAlchemy dialect being used.	required
coltype	Any	Type of the column being processed.	required