diff --git a/maltoolbox/language/lexer_parser/__main__.py b/maltoolbox/language/lexer_parser/__main__.py index 9086db5c..5f76ebdd 100644 --- a/maltoolbox/language/lexer_parser/__main__.py +++ b/maltoolbox/language/lexer_parser/__main__.py @@ -4,6 +4,7 @@ from .mal_lexer import malLexer from .mal_parser import malParser from .mal_visitor import malVisitor +from .mal_analyzer import malAnalyzer import sys import os @@ -14,6 +15,7 @@ class MalCompiler: def __init__(self): self.path = None self.current_file = None + self._analyzer = malAnalyzer() def compile(self, malfile: str = None): if not self.path: @@ -28,8 +30,7 @@ def compile(self, malfile: str = None): stream = CommonTokenStream(lexer) parser = malParser(stream) tree = parser.mal() - - return malVisitor(compiler=self).visit(tree) + return malVisitor(compiler=self, analyzer=self._analyzer).visit(tree) if __name__ == "__main__": diff --git a/maltoolbox/language/lexer_parser/mal_analyzer.py b/maltoolbox/language/lexer_parser/mal_analyzer.py new file mode 100644 index 00000000..55ab6bc9 --- /dev/null +++ b/maltoolbox/language/lexer_parser/mal_analyzer.py @@ -0,0 +1,560 @@ +from .mal_parser import malParser + +import logging +import re + +from typing import Any, Tuple, List + +class malAnalyzerInterface: + def checkMal(self, ctx: malParser.MalContext) -> None: + pass + def checkDefine(self, ctx: malParser.DefineContext, data: Tuple[str, dict]) -> None: + pass + def checkInclude(self, ctx: malParser.IncludeContext, data: Tuple[str, str]) -> None: + pass + def checkCategory(self, ctx: malParser.CategoryContext, data: Tuple[str, Tuple[List, Any]]) -> None: + pass + def checkAsset(self, ctx: malParser.AssetContext, asset: dict) -> None: + pass + def checkMeta(self, ctx: malParser.MetaContext, data: Tuple[Tuple[str, str],]) -> None: + pass + def checkStep(self, ctx: malParser.StepContext, step: dict) -> None: + pass + def checkVariable(self, ctx: malParser.VariableContext, var: dict) -> None: + pass + def checkAssociation(self, ctx: malParser.AssociationContext, association: dict) -> None: + pass + def checkReaches(self, ctx: malParser.ReachesContext, data: dict) -> None: + pass + +class malAnalyzer(malAnalyzerInterface): + ''' + A class to preform syntax-checks for MAL. + ''' + + def __init__(self, *args, **kwargs) -> None: + self._error: bool = False + self._preform_post_analysis = True + + self._defines: dict = {} + self._assets: dict = {} + self._category: dict = {} + self._metas: dict = {} + self._steps: dict = {} + self._vars: dict = {} + + self._associations = [] + + super().__init__(*args, **kwargs) + + def has_error(self) -> bool: + return self._error + + def _post_analysis(self) -> None: + ''' + Perform a post-analysis to confirm that the + mandatory fields and relations are met. + ''' + self._analyse_defines() + self._analyse_extends() + self._analyse_abstract() + self._analyse_parents() + self._analyse_reaches() + self._analyse_association() + + def _analyse_defines(self) -> None: + ''' + Check for mandatory defines: ID & Version + ''' + + if 'id' in self._defines.keys(): + define_value: str = self._defines['id']['value'] + if (len(define_value) == 0): + logging.error('Define \'id\' cannot be empty') + self._error = True + else: + logging.error('Missing required define \'#id: ""\'') + self._error = True + + if 'version' in self._defines: + version: str = self._defines['version']['value'] + if not re.match(r"\d+\.\d+\.\d+", version): + logging.error(f'Define \'version\' must be valid semantic versioning without pre-release identifier and build metadata') + self._error = True + else: + logging.error('Missing required define \'#version: ""\'') + self._error = True + + def _analyse_extends(self) -> None: + raise_error: bool = False + extend_asset_name: str = '' + for asset in self._assets: + asset_context: malParser.AssetContext = self._assets[asset]['ctx'] + if(asset_context.EXTENDS()): + extend_asset_name = asset_context.ID()[1].getText() + if(not extend_asset_name in self._assets): + ''' + Do we need to check if the extended asset is + in the same category? If so we can load the asset + and check it's parent + ''' + logging.error(f'Asset \'{extend_asset_name}\' not defined') + raise_error = True + if raise_error: + self._error = True # Maybe unnecessary if we raise + raise SyntaxError(f'Asset \'{extend_asset_name}\' not defined') + + def _analyse_abstract(self) -> None: + for parent in self._assets: + parent_ctx: malParser.AssetContext = self._assets[parent]['ctx'] + if(parent_ctx.ABSTRACT()): + found: bool = False + for extendee in self._assets: + ''' + Add same parent check? + ''' + extendee_ctx: malParser.AssetContext = self._assets[extendee]['ctx'] + if(extendee_ctx.EXTENDS() and extendee_ctx.ID()[1].getText() == parent_ctx.ID()[0].getText()): + found = True + break + if not found: + logging.warn(f'Asset \'{parent_ctx.ID()[0].getText()}\' is abstract but never extended to') + + def _analyse_parents(self) -> None: + error: bool = False + for asset in self._assets: + parents: list[str] = [] + parent_ctx: malParser.AssetContext = self._assets[asset]['ctx'] + while (isinstance(parent_ctx, malParser.AssetContext)): + parent_name: str = parent_ctx.ID()[0].getText() + if (parent_name in parents): + err_msg: str = ' -> '.join(parents) + err_msg += f' -> {parent_name}' + logging.error(f'Asset \'{parent_name}\' extends in loop \'{err_msg}\'') + error = True + break + parents.append(parent_name) + parent_ctx = self._get_assets_extendee(parent_ctx) + if error: + self._error = True + raise + + def _analyse_reaches(self) -> None: + for asset in self._assets.keys(): + attack_steps = self._assets[asset]['obj']['attackSteps'] + for attack_step in attack_steps: + if (attack_step['type'] in ['exist', 'notExist']): + if (attack_step['ttc']): + logging.error(f'Attack step of type \'{attack_step["type"]}\' must not have TTC') + self._error = True + continue + if (attack_step['requires']): + for expr in attack_step['requires']['stepExpressions']: + self._check_to_asset(asset, expr) + else: + logging.error(f'Attack step of type \'{attack_step["type"]}\' must have require \'<-\'') + self._error = True + continue + elif (attack_step['requires']): + logging.error('Require \'<-\' may only be defined for attack step type exist \'E\' or not-exist \'!E\'') + self._error = True + continue + + if (attack_step['reaches']): + for expr in attack_step['reaches']['stepExpressions']: + self._check_to_step(asset, expr) + + if (False): + raise '' + + def _check_to_step(self, asset, expr) -> None: + match (expr['type']): + case 'attackStep': + if (asset in self._assets.keys()): + for attackStep in self._assets[asset]['obj']['attackSteps']: + if (attackStep['name'] == expr['name']): + return attackStep + + logging.error(f'Attack step \'{expr["name"]}\' not defined for asset \'{asset}\'') + self._error = True + return None + case 'collect' | 'union' | 'intersection' | 'difference': #'AST.StepExpr': + if (left_target := self._check_to_asset(asset, expr['lhs'])): + return self._check_to_step(left_target, expr['rhs']) + return None + case 'subType': + if (sub_type := self._get_asset_name(expr['subType'])): + return self._check_association_expr(sub_type, expr['stepExpression']) + return None + case _: + logging.error('Last step is not attack step') + self._error = True + return None + + def _check_to_asset(self, asset, expr) -> None: + match (expr['type']): + case 'field' | 'attackStep': + # This is an implementation of IDExpr & CallExpr + return self._check_association_expr(asset, expr) + case 'collect': #'StepExpr': + return self._check_step_expr(asset, expr) + case 'IDExpr' | 'CallExpr': + raise + case 'union' | 'intersection' | 'difference': + return self._check_set_expr(asset, expr) + case 'transitive': + return self._check_transitive_expr(asset, expr) + case 'subType': + return self._check_sub_type_expr(asset, expr) + case _: + logging.error(f'Unexpected expression \'{expr["type"]}\'') + self._error = True + # exit(1) + return None + + def _check_association_expr(self, asset, expr): + ''' + Check field for association or variable reference to asset. + ''' + for association in self._associations: + if (expr['name'] == association['leftField']): + if (self._get_asset_name(association['leftAsset'])): + return association['leftAsset'] + if (expr['name'] == association['rightField']): + if (self._get_asset_name(association['rightAsset'])): + return association['rightAsset'] + + if (asset in self._vars.keys() and expr['name'] in self._vars[asset].keys()): + return self._check_to_asset(asset, self._vars[asset][expr['name']]['var']['stepExpression']) + + # logging.error(f'Variable \'{expr["name"]}\' is not defined') + logging.error(f'Field \'{expr["name"]}\' not defined for asset \'{asset}\'') + self._error = True + return None + + def _check_step_expr(self, asset, expr): + if (left_target := self._check_to_asset(asset, expr['lhs'])): + return self._check_to_asset(left_target, expr['rhs']) + return None + + def _check_set_expr(self, asset, expr) -> None: + lhs_target = self._check_to_asset(asset, expr['lhs']) + rhs_target = self._check_to_asset(asset, expr['rhs']) + if (not lhs_target or not rhs_target): + return None + + if (target := self._get_LCA(lhs_target, rhs_target)): + return target + + logging.error(f'Types \'{lhs_target["name"]}\' and \'{rhs_target["name"]}\' have no common ancestor') + self._error = True + return None + + def _get_LCA(self, lhs_target, rhs_target): + if (self._is_child(lhs_target, rhs_target)): + return lhs_target + elif (self._is_child(rhs_target, lhs_target)): + return rhs_target + else: + lhs_ctx = self._assets[lhs_target]['ctx'] + rhs_ctx = self._assets[rhs_target]['ctx'] + lhs_parent_ctx = self._get_assets_extendee(lhs_ctx) + rhs_parent_ctx = self._get_assets_extendee(rhs_ctx) + if (not lhs_parent_ctx or not rhs_parent_ctx): + return None + return self._get_LCA(lhs_parent_ctx.ID()[0].getText(), rhs_parent_ctx.ID()[0].getText()) + + def _check_sub_type_expr(self, asset, expr) -> None: + target = self._check_to_asset(asset, expr['stepExpression']) + if (not target): + return None + + if (asset_type := self._get_asset_name(expr['subType'])): + if (self._is_child(target, asset_type)): + return asset_type + + logging.error(f'Asset \'{target}\' cannot be of type \'{asset_type}\'') + self._error = True + return None + + def _check_transitive_expr(self, asset, expr) -> None: + if (res := self._check_to_asset(asset, expr['stepExpression'])): + if (self._is_child(res, asset)): + return res + + logging.error(f'Previous asset \'{asset}\' is not of type \'{res}\'') + self._error = True + return None + + def _is_child(self, parent_name, child_name): + if (parent_name == child_name): + return True + + if (valid_asset := self._get_asset_name(child_name)): + asset_context: malParser.AssetContext = self._assets[valid_asset]['ctx'] + if (parent_ctx := self._get_assets_extendee(asset_context)): + child_parent_name = self._get_asset_name(parent_ctx.ID()[0].getText()) + return self._is_child(parent_name, child_parent_name) + + return False + + def _get_asset_name(self, name): + if (name in self._assets.keys()): + return name + + logging.error(f'Asset \'{name}\' not defined') + self._error = True + return None + + def _analyse_association(self) -> None: + for association in self._associations: + leftAsset = association['leftAsset'] + rightAsset = association['rightAsset'] + + if (not leftAsset in self._assets.keys()): + logging.error(f'Left asset \'{leftAsset}\' is not defined') + self._error = True + if (not rightAsset in self._assets.keys()): + logging.error(f'Right asset \'{leftAsset}\' is not defined') + self._error = True + if (self._error): + raise + + def _get_assets_extendee(self, ctx: malParser.AssetContext) -> malParser.AssetContext: + if (ctx.EXTENDS()): + return self._assets[ctx.ID()[1].getText()]['ctx'] + return None + + def checkMal(self, ctx: malParser.MalContext) -> None: + ''' + We only want to preform _post_analysis as the very last step. + ''' + if (self._preform_post_analysis): + self._post_analysis() + self._preform_post_analysis = True + + def checkInclude(self, ctx: malParser.MalContext, data: Tuple[str, str]) -> None: + self._preform_post_analysis = False + + def checkDefine(self, ctx: malParser.DefineContext, data: Tuple[str, dict]) -> None: + _, obj = data + key, value = list(obj.items())[0] + + if(key in self._defines.keys()): + prev_define_line = self._defines[key]['ctx'].start.line + logging.error(f'Define \'{key}\' previously defined at line {prev_define_line}') + self._error = True + return + + self._defines[key] = {'ctx': ctx, 'value': value} + + def checkCategory(self, ctx: malParser.CategoryContext, data: Tuple[str, Tuple[List, Any]]) -> None: + _, [[category], assets] = data + + if(str(category['name']) == '>'): + category_line = ctx.start.line + logging.error(f'Category has no name at line {category_line}') + self._error = True + return + + if len(category['meta']) == 0 and len(assets) == 0: + logging.warning(f'Category \'{category["name"]}\' contains no assets or metadata') + + self._category[category['name']] = {'ctx': ctx, 'obj': {'category': category, 'assets': assets}} + + def checkAsset(self, ctx: malParser.AssetContext, asset: dict) -> None: + asset_name = asset['name'] + category_name = ctx.parentCtx.ID().getText() + + if (not asset_name or asset_name == '>'): + logging.error(f"Asset was defined without a name at line {ctx.start.line}") + self._error = True + return + # Check if asset was previously defined in same category. + if asset_name in self._assets.keys() and str(self._assets[asset_name]['parent']['name']) == str(category_name): + prev_asset_line = self._assets[asset_name]['ctx'].start.line + logging.error(f"Asset '{asset_name}' previously defined at {prev_asset_line}") + self._error = True + return + else: + self._assets[asset_name] = {'ctx': ctx, 'obj': asset, 'parent': {'name': ctx.parentCtx.ID().getText() ,'ctx': ctx.parentCtx}} + + def checkMeta(self, ctx: malParser.MetaContext, data: Tuple[Tuple[str, str],]) -> None: + ((meta_name, _),) = data + parent_name: str = '' + location_name: str = '' + + # Finding metadata type + if isinstance(ctx.parentCtx, malParser.CategoryContext): + parent_name = str(ctx.parentCtx.ID().getText()) + location_name = 'category' + elif isinstance(ctx.parentCtx, malParser.AssetContext): + parent_name = str(ctx.parentCtx.ID()[0].getText()) + location_name = 'asset' + elif isinstance(ctx.parentCtx, malParser.StepContext): + parent_name = str(ctx.parentCtx.ID().getText()) + location_name = 'step' + elif isinstance(ctx.parentCtx, malParser.AssociationContext): + parent_name = str(ctx.parentCtx.ID()[0].getText()) + location_name = 'association' + + # Validate that the metadata is unique + if not location_name in self._metas.keys(): + self._metas[location_name] = {parent_name: {meta_name: ctx}} + elif not parent_name in self._metas[location_name].keys(): + self._metas[location_name][parent_name] = {meta_name: ctx} + elif not meta_name in self._metas[location_name][parent_name].keys(): + self._metas[location_name][parent_name][meta_name] = ctx + else: + prev_ctx = self._metas[location_name][parent_name][meta_name] + logging.error(f'Metadata {meta_name} previously defined at {prev_ctx.start.line}') + self._error = True + + def checkStep(self, ctx: malParser.StepContext, step: dict) -> None: + step_name = step['name'] + + if isinstance(ctx.parentCtx, malParser.AssetContext): + asset_name = ctx.parentCtx.ID()[0].getText() + # Check if the step is defined in other assets. + for other_asset_name in self._steps.keys(): + if (asset_name == other_asset_name): + continue + if not (self._steps[other_asset_name] and step_name in self._steps[other_asset_name].keys()): + continue + + other_step = self._steps[other_asset_name][step_name] + other_type = other_step['step']['type'] + current_type = step['type'] + if (other_type == current_type): + self._steps[asset_name] = {step_name: {'ctx': ctx, 'step': step}} + return + + prev_ctx = other_step['ctx'] + logging.error(f'Cannot override attack step \'{step_name}\' previously defined at {prev_ctx.start.line} with different type \'{current_type}\' =/= \'{other_type}\'') + self._error = True + return + + if ((step['reaches'] and not step['reaches']['overrides'])): + logging.error(f'Cannot inherit attack step \'{step_name}\' without previous definition') + self._error = True + return + + # Check if the step is already defined in the parent asset. + if not asset_name in self._steps.keys(): + self._steps[asset_name] = {step_name: {'ctx': ctx, 'step': step}} + elif not step_name in self._steps[asset_name].keys(): + self._steps[asset_name][step_name] = {'ctx': ctx, 'step': step} + else: + prev_ctx = self._steps[asset_name][step_name]['ctx'] + logging.error(f'Attack step \'{step_name}\' previously defined at {prev_ctx.start.line}') + self._error = True + + self._validate_CIA(ctx, step) + self._validate_TTC(ctx, asset_name, step) + + def checkReaches(self, ctx: malParser.ReachesContext, data: dict) -> None: + pass + + def _validate_TTC(self, ctx: malParser.StepContext, asset_name, step: dict) -> None: + if not step['ttc']: + return + match step['type']: + case 'defense': + if (step['ttc']['type'] != 'function'): + logging.error(f'Defense {asset_name}.{step["name"]} may not have advanced TTC expressions') + self._error = True + return + + match step['ttc']['name']: + case 'Enabled' | 'Disabled' | 'Bernoulli': + # try/catch Distributions.validate(name, params) + return + case _: + logging.error(f'Defense {asset_name}.{step["name"]} may only have \'Enabled\', \'Disabled\', or \'Bernoulli(p)\' as TTC') + self._error = True + return + case 'exist' | 'notExist': + pass + case _: + self._check_TTC_expr(step['ttc']) + + def _check_TTC_expr(self, expr, isSubDivExp = False): + match expr['type']: + case 'function': + if (expr['name'] == 'Enabled' or expr['name'] == 'Disabled'): + logging.error('Distributions \'Enabled\' or \'Disabled\' may not be used as TTC values in \'&\' and \'|\' attack steps') + self._error = True + return + if (isSubDivExp and expr['name'] in ['Bernoulli', 'EasyAndUncertain']): + logging.error(f'TTC distribution \'{expr["name"]}\' is not available in subtraction, division or exponential expressions.') + self._error = True + return + # try/catch Distributions.validate(name, params) + case 'subtraction' | 'exponentiation' | 'division': + self._check_TTC_expr(expr['lhs'], True) + self._check_TTC_expr(expr['rhs'], True) + case 'multiplication' | 'addition': + self._check_TTC_expr(expr['lhs'], False) + self._check_TTC_expr(expr['rhs'], False) + case 'number': + pass + case _: + logging.error(f'Unexpected expression {expr}') + self._error = True + # exit(1) + + + def _validate_CIA(self, ctx: malParser.StepContext, step: dict) -> None: + if not ctx.cias(): + return + + step_name = step['name'] + asset_name = ctx.parentCtx.ID()[0].getText() + + if (step['type'] == 'defense' or step['type'] == 'exist' or step['type'] == 'notExist'): + logging.error(f'{step_name}: Defenses cannot have CIA classifications') + self._error = True + return + + index = 0 + cias = [] + while cia := ctx.cias().getChild(index): + if(isinstance(cia, malParser.CiaContext)): + letter = '' + letter = 'C' if cia.C() else letter + letter = 'I' if cia.I() else letter + letter = 'A' if cia.A() else letter + + if (letter in cias): + logging.error(f'Attack step {asset_name}.{step_name} contains duplicate classification {letter}') + self._error = True + return + cias.append(letter) + index += 1 + + def checkVariable(self, ctx: malParser.VariableContext, var: dict) -> None: + ''' + self._vars = { + : { + : + } + } + ''' + parent = ctx.parentCtx + if (isinstance(parent, malParser.AssetContext)): + asset_name: str = str(parent.ID()[0].getText()) + var_name: str = var['name'] + if (asset_name not in self._vars.keys()): + self._vars[asset_name] = {var_name: {'ctx': ctx, 'var': var}} + elif (var_name not in self._vars[asset_name]): + self._vars[asset_name][var_name] = {'ctx': ctx, 'var': var} + else: + prev_define_line = self._vars[asset_name][var_name]['ctx'].start.line + logging.error(f'Variable \'{var_name}\' previously defined at line {prev_define_line}') + self._error = True + else: + # TODO + raise + + def checkAssociation(self, ctx: malParser.AssociationContext, association: dict): + self._associations.append(association) \ No newline at end of file diff --git a/maltoolbox/language/lexer_parser/mal_visitor.py b/maltoolbox/language/lexer_parser/mal_visitor.py index f9ea74b4..27fecbb1 100644 --- a/maltoolbox/language/lexer_parser/mal_visitor.py +++ b/maltoolbox/language/lexer_parser/mal_visitor.py @@ -1,6 +1,7 @@ from antlr4 import ParseTreeVisitor from .mal_parser import malParser +from .mal_analyzer import malAnalyzerInterface from collections.abc import MutableMapping, MutableSequence @@ -10,12 +11,35 @@ class malVisitor(ParseTreeVisitor): - def __init__(self, compiler, *args, **kwargs): + def __init__(self, compiler, analyzer: malAnalyzerInterface, *args, **kwargs): self.compiler = compiler + self.analyzer = analyzer self.current_file = compiler.current_file # for debug purposes super().__init__(*args, **kwargs) + def visit(self, ctx): + data = super().visit(ctx) + + if (not ctx): + return data + + ctx_class_name: str = type(ctx).__name__ + analyzer_method_name: str = f'check{ctx_class_name.replace("Context", "")}' + analyzer_method: function | None = getattr(self.analyzer, analyzer_method_name, None) + + if analyzer_method: + arguments = analyzer_method.__code__.co_argcount + if arguments in [2, 3]: + { + 3: lambda: analyzer_method(ctx, data), + 2: lambda: analyzer_method(ctx) + }[arguments]() + else: + raise ValueError(f'Unexpected number of arguments: {arguments}') + + return data + def visitMal(self, ctx): langspec = { "formatVersion": "1.0.0", @@ -255,7 +279,7 @@ def visitExpr(self, ctx): ret["lhs"] = lhs ret["rhs"] = self.visit(ctx.parts()[i]) lhs = ret.copy() - + return ret def visitParts(self, ctx): diff --git a/maltoolbox/language/tests/__init__.py b/maltoolbox/language/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/maltoolbox/language/tests/mal_analyzer_test_wrapper.py b/maltoolbox/language/tests/mal_analyzer_test_wrapper.py new file mode 100644 index 00000000..0af3957b --- /dev/null +++ b/maltoolbox/language/tests/mal_analyzer_test_wrapper.py @@ -0,0 +1,63 @@ + +from antlr4 import FileStream, InputStream, CommonTokenStream + +from lexer_parser.mal_lexer import malLexer +from lexer_parser.mal_parser import malParser +from lexer_parser.mal_visitor import malVisitor +from lexer_parser.mal_analyzer import malAnalyzer + +import os + +class MockCompiler(): + def __init__(self, analyzer: malAnalyzer): + self._analyzer = analyzer + self.path = None + self.current_file = None + + def compile(self, malfile: str = None): + if not self.path: + self.path = os.path.dirname(malfile) + + self.current_file = os.path.basename(malfile) + + input_stream = FileStream( + os.path.join(self.path, self.current_file), encoding="utf-8" + ) + lexer = malLexer(input_stream) + stream = CommonTokenStream(lexer) + parser = malParser(stream) + tree = parser.mal() + return malVisitor(compiler=self, analyzer=self._analyzer).visit(tree) + +class AnalyzerTestWrapper(malAnalyzer): + def __init__(self, input_string: str) -> None: + super().__init__() + input_stream = InputStream(input_string) + lexer = malLexer(input_stream) + stream = CommonTokenStream(lexer) + parser = malParser(stream) + tree = parser.mal() + compiler = MockCompiler(self) + try: + self._result = malVisitor(compiler=compiler, analyzer=self).visit(tree) + except SyntaxError: + self._error = True + except RuntimeError: + self._error = True + + def test(self, error:bool=False, defines:list=[], categories:list=[], assets:list=[], lets:list=[]): + assert(self.has_error() == error) + if (defines): + assert(set(defines) == set(self._defines.keys())) + if (categories): + assert(set(categories) == set(self._category.keys())) + if (assets): + assert(set(assets) == set(self._assets.keys())) + if (lets): + for let in lets: + where, name = let + if (not (self._vars[where] and self._vars[where][name])): + assert(False) + + + diff --git a/maltoolbox/language/tests/test_mal_abstract_asset.py b/maltoolbox/language/tests/test_mal_abstract_asset.py new file mode 100644 index 00000000..cd88bd1b --- /dev/null +++ b/maltoolbox/language/tests/test_mal_abstract_asset.py @@ -0,0 +1,47 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + +''' +A file to test different cases of the `abstract` instruction in MAL. +''' + +def test_abstract_assets_1() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines abstract asset without name. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + abstract asset {} + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'] + ) + +def test_abstract_assets_2() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines abstract asset with name. + Extends asset with abstract asset. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + abstract asset Foo {} + asset Bar extends Foo {} + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Foo', 'Bar'] + ) diff --git a/maltoolbox/language/tests/test_mal_assets.py b/maltoolbox/language/tests/test_mal_assets.py new file mode 100644 index 00000000..eea4796f --- /dev/null +++ b/maltoolbox/language/tests/test_mal_assets.py @@ -0,0 +1,91 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + +''' +A file to test different cases of the `asset` instruction in MAL. +''' + +def test_assets_1() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset without name. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset {} + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'] + ) + +def test_assets_2() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Test {} + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Test'] + ) + +def test_assets_3() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset twice. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Test {} + asset Test {} + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Test'] + ) + +def test_assets_4() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset twice. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Test {} + } + category System { + asset Test {} + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Test'] + ) \ No newline at end of file diff --git a/maltoolbox/language/tests/test_mal_association.py b/maltoolbox/language/tests/test_mal_association.py new file mode 100644 index 00000000..186a72a4 --- /dev/null +++ b/maltoolbox/language/tests/test_mal_association.py @@ -0,0 +1,139 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + +''' +A file to test different cases of the `association` instruction in MAL. +''' + +def test_association_1() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines two asset with name. + Defines an association between the assets. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Asset1 {} + asset Asset2 {} + } + associations { + Asset1 [foo] * <-- connects --> * [bar] Asset2 + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Asset1', 'Asset2'] + ) + +def test_association_2() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines one asset with name. + Defines an association between the asset + and one undefined right asset. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Asset1 {} + } + + associations { + Asset1 [foo] * <-- connects --> * [bar] Asset2 + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Asset1'] + ) + +def test_association_3() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines one asset with name. + Defines an association between the asset + and one undefined left asset. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Asset2 {} + } + associations { + Asset1 [foo] * <-- connects --> * [bar] Asset2 + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Asset2'] + ) + + +def test_association_4() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category Example { + asset Asset1 + { + | compromise + -> b.compromise + } + asset Asset2 + { + | compromise + } + } + associations + { + Asset1 [a] * <-- L --> * [b] Asset2 + } + + ''').test( + defines=['id', 'version'], + categories=['Example'], + assets=['Asset1', 'Asset2'] + ) + +def test_association_5() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category Example { + asset Asset1 + { + | compromise + -> b.compromise + } + asset Asset2 + { + | compromise + } + } + associations + { + Asset1 [a] * <-- L --> * [c] Asset2 + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['Example'], + assets=['Asset1', 'Asset2'] + ) \ No newline at end of file diff --git a/maltoolbox/language/tests/test_mal_category.py b/maltoolbox/language/tests/test_mal_category.py new file mode 100644 index 00000000..bc067b4a --- /dev/null +++ b/maltoolbox/language/tests/test_mal_category.py @@ -0,0 +1,54 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + +''' +A file to test different cases of the `category` instruction in MAL. +''' + +def test_category_1() -> None: + ''' + Defines correct version and ID. + Defines category without name. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category {} + + ''').test( + error=True, + defines=['id', 'version'] + ) + +def test_category_2() -> None: + ''' + Defines correct version and ID. + Defines category with name. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category Test {} + + ''').test( + defines=['id', 'version'], + categories=['Test'] + ) + +def test_category_3() -> None: + ''' + Defines correct version and ID. + Defines category with same name twice. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category Test {} + category Test {} + + ''').test( + defines=['id', 'version'], + categories=['Test'] + ) \ No newline at end of file diff --git a/maltoolbox/language/tests/test_mal_define.py b/maltoolbox/language/tests/test_mal_define.py new file mode 100644 index 00000000..79f6a229 --- /dev/null +++ b/maltoolbox/language/tests/test_mal_define.py @@ -0,0 +1,91 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + +''' +A file to test different cases of the `define` instruction in MAL. +''' + +def test_define_1() -> None: + ''' + Defines only version. + ''' + AnalyzerTestWrapper(''' + #version:"0.0.0" + ''').test( + error=True, + defines=['version'] + ) + +def test_define_2() -> None: + ''' + Defines only ID. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + ''').test( + error=True, + defines=['id'] + ) + +def test_define_3() -> None: + ''' + Defines correct version but wrong ID. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"version1" + ''').test( + error=True, + defines=['id', 'version'] + ) + +def test_define_4() -> None: + ''' + Defines correct version and ID. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + ''').test( + defines=['id', 'version'] + ) + +def test_define_5() -> None: + ''' + Defines correct version and ID, but ID twice. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + #version:"1.0.0" + ''').test( + error=True, + defines=['id', 'version'] + ) + +def test_define_6() -> None: + ''' + Defines correct version, ID. + Defines Key with value. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + #key:"value" + ''').test( + defines=['id', 'version', 'key'] + ) + +def test_define_7() -> None: + ''' + Defines correct version, ID. + Defines Key with value twice. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + #key:"value" + #key:"value" + ''').test( + error=True, + defines=['id', 'version', 'key'] + ) \ No newline at end of file diff --git a/maltoolbox/language/tests/test_mal_developer_info.py b/maltoolbox/language/tests/test_mal_developer_info.py new file mode 100644 index 00000000..3bf1ebfe --- /dev/null +++ b/maltoolbox/language/tests/test_mal_developer_info.py @@ -0,0 +1,53 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + +''' +A file to test different cases of the `developer info` instruction in MAL. +''' + +def test_developer_info_1() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Defines developer info. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Foo + developer info: "Hello" + {} + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Foo'] + ) + +def test_developer_info_2() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Defines developer info twice. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Foo + developer info: "Hello" + developer info: "Hello" + {} + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Foo'] + ) \ No newline at end of file diff --git a/maltoolbox/language/tests/test_mal_extends.py b/maltoolbox/language/tests/test_mal_extends.py new file mode 100644 index 00000000..03bfa9d0 --- /dev/null +++ b/maltoolbox/language/tests/test_mal_extends.py @@ -0,0 +1,78 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + +''' +A file to test different cases of the `extends` instruction in MAL. +''' + +def test_extends_1() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Defines asset with extends. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem {} + asset Linux extends OperatingSystem {} + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem', 'Linux'] + ) + +def test_extends_2() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Extends asset with undefined asset. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Foo1 extends Foo2 {} + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Foo1'] + ) + +def test_extends_3() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Tests circular dependency with extends + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Foo1 extends Foo2 {} + asset Foo2 extends Foo3 {} + asset Foo3 extends Foo4 {} + asset Foo4 extends Foo5 {} + asset Foo5 extends Foo1 {} + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Foo1', 'Foo2', 'Foo3', 'Foo4', 'Foo5'] + ) + + + \ No newline at end of file diff --git a/maltoolbox/language/tests/test_mal_include.py b/maltoolbox/language/tests/test_mal_include.py new file mode 100644 index 00000000..daa1c8a2 --- /dev/null +++ b/maltoolbox/language/tests/test_mal_include.py @@ -0,0 +1,153 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + +import os + +''' +A file to test different cases of the `include` instruction in MAL. +''' + +def test_include_1() -> None: + ''' + Missing keys ID and version. + ''' + path = "./generated_test_mal.mal" + with open(path, 'w') as file: + file.write('') + AnalyzerTestWrapper(f''' + include "{path}" + ''').test( + error=True + ) + + if os.path.exists(path): + os.remove(path) + +def test_include_2() -> None: + ''' + Including file with ID and version. + ''' + path = "./generated_test_mal.mal" + with open(path, 'w') as file: + file.write(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + ''') + AnalyzerTestWrapper(f''' + include "{path}" + ''').test( + defines=['id', 'version'] + ) + + if os.path.exists(path): + os.remove(path) + +def test_include_3() -> None: + ''' + Including file with ID and version. + Defining ID and version both files. + ''' + path = "./generated_test_mal.mal" + with open(path, 'w') as file: + file.write(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + ''') + AnalyzerTestWrapper(f''' + include "{path}" + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + ''').test( + error=True, + defines=['id', 'version'] + ) + + if os.path.exists(path): + os.remove(path) + +def test_include_4() -> None: + ''' + Including file with ID and version. + Defining key with value. + ''' + path = "./generated_test_mal.mal" + with open(path, 'w') as file: + file.write(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + ''') + AnalyzerTestWrapper(f''' + include "{path}" + #key: "test" + ''').test( + defines=['id', 'version', 'key'] + ) + + if os.path.exists(path): + os.remove(path) + +def test_include_5() -> None: + ''' + Including one file with ID and another with version. + Defining key with value. + ''' + path_1 = "./generated_test_mal_1.mal" + path_2 = "./generated_test_mal_2.mal" + with open(path_1, 'w') as file: + file.write(''' + #version:"0.0.0" + ''') + with open(path_2, 'w') as file: + file.write(''' + #id: "org.mal-lang.testAnalyzer" + ''') + AnalyzerTestWrapper(f''' + include "{path_1}" + include "{path_2}" + #key: "test" + ''').test( + defines=['id', 'version', 'key'] + ) + + if os.path.exists(path_1): + os.remove(path_1) + if os.path.exists(path_2): + os.remove(path_2) + +def test_include_6() -> None: + ''' + Include same file twice. + ''' + path = "./generated_test_mal.mal" + with open(path, 'w') as file: + file.write('') + AnalyzerTestWrapper(f''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + include "{path}" + include "{path}" + ''').test( + defines=['id', 'version'] + ) + + if os.path.exists(path): + os.remove(path) + +def test_include_7() -> None: + ''' + Defining keys ID and version after include. + ''' + path = "./generated_test_mal.mal" + with open(path, 'w') as file: + file.write('') + AnalyzerTestWrapper(f''' + include "{path}" + + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + ''').test( + defines=['id', 'version'] + ) + + if os.path.exists(path): + os.remove(path) \ No newline at end of file diff --git a/maltoolbox/language/tests/test_mal_let.py b/maltoolbox/language/tests/test_mal_let.py new file mode 100644 index 00000000..34781ab9 --- /dev/null +++ b/maltoolbox/language/tests/test_mal_let.py @@ -0,0 +1,118 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + +''' +A file to test different cases of the `let` instruction in MAL. +''' + +def test_let_1() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Defines let. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Computer { + let components = software \\/ hardware + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Computer'], + lets=[('Computer', 'components')] + ) + +def test_let_2() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines two assets with name. + Defines same let twice. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Computer { + let components = hardware + } + } + category System { + asset Computer { + let components = software + } + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Computer'], + lets=[('Computer', 'components')] + ) + +def test_let_3() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Defines same let twice. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Computer { + let components = hardware + let components = software + } + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Computer'], + lets=[('Computer', 'components')] + ) + +def test_let_4() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Defines same let twice. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Asset1 { + | compromise + } + asset Computer { + let var1 = asset1 + | compromise + -> var1.compromise + } + } + + associations + { + Computer [computer] * <-- L --> * [asset1] Asset1 + } + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Computer', 'Asset1'], + lets=[('Computer', 'var1')] + ) \ No newline at end of file diff --git a/maltoolbox/language/tests/test_mal_modeler_info.py b/maltoolbox/language/tests/test_mal_modeler_info.py new file mode 100644 index 00000000..17ca1ad3 --- /dev/null +++ b/maltoolbox/language/tests/test_mal_modeler_info.py @@ -0,0 +1,53 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + +''' +A file to test different cases of the `modeler info` instruction in MAL. +''' + +def test_modeler_info_1() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Defines modeler info. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Foo + modeler info: "Hello" + {} + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Foo'] + ) + +def test_modeler_info_2() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Defines modeler info twice. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Foo + modeler info: "Hello" + modeler info: "Hello" + {} + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Foo'] + ) \ No newline at end of file diff --git a/maltoolbox/language/tests/test_mal_operations.py b/maltoolbox/language/tests/test_mal_operations.py new file mode 100644 index 00000000..32d84fd4 --- /dev/null +++ b/maltoolbox/language/tests/test_mal_operations.py @@ -0,0 +1,276 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + + + +def test_operation_1() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Computer + { + let allTargets = computers.operatingSystems[Windows] /\\ computers.operatingSystems[Linux] + | findWindowsAndLinux + -> allTargets[Windows].foundWindows + } + + abstract asset OperatingSystem + {} + asset Windows extends OperatingSystem + { + | foundWindows @debug + } + asset Linux extends OperatingSystem + { + | foundLinux @debug + } + } + + associations + { + Computer [computers] * <-- L --> 0..1 [operatingSystems] OperatingSystem + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Computer', 'Windows', 'Linux', 'OperatingSystem'] + ) + +def test_operation_2() -> None: + ''' + This test might be invalid. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Computer + { + let allTargets = computers.operatingSystems[Windows] /\\ computers.operatingSystems[Linux] + | findWindowsAndLinux + -> allTargets[Windows].foundLinux + } + + abstract asset OperatingSystem + {} + asset Windows extends OperatingSystem + { + | foundWindows @debug + } + asset Linux extends OperatingSystem + { + | foundLinux @debug + } + } + + associations + { + Computer [computers] * <-- L --> 0..1 [operatingSystems] OperatingSystem + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Computer', 'Windows', 'Linux', 'OperatingSystem'] + ) + +def test_operation_3() -> None: + ''' + This test might be invalid. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Computer + { + let allTargets = computers.operatingSystems[Windows] /\\ computers.operatingSystems[Linux] + | findWindowsAndLinux + -> allTargets[Windowss].foundWindows + } + + abstract asset OperatingSystem + {} + asset Windows extends OperatingSystem + { + | foundWindows @debug + } + asset Linux extends OperatingSystem + { + | foundLinux @debug + } + } + + associations + { + Computer [computers] * <-- L --> 0..1 [operatingSystems] OperatingSystem + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Computer', 'Windows', 'Linux', 'OperatingSystem'] + ) + +def test_operation_4() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category Test { + asset Policy { + & satisfy @trace + | calledByUser + -> satisfy + | calledByHost + -> satisfy + } + + asset User { + let effectivePolicies = policies + let policyReachableHosts = policies.hosts + + | compromise + -> effectivePolicies.calledByUser, + policyReachableHosts.connect + } + + asset Host { + let effectivePolicies = policies + | connect + -> effectivePolicies.calledByHost + + } + } + associations + { + User [user] * <-- L --> * [policies] Policy + Policy [policies] * <-- L --> * [hosts] Host + } + ''').test( + defines=['id', 'version'], + categories=['Test'], + assets=['Policy', 'User', 'Host'] + ) + +def test_operation_5() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Computer + { + let allTargets = computers.operatingSystems[Windows] \\/ computers.operatingSystems[Linux] + | findWindowsAndLinux + -> allTargets[Windows].foundWindows + } + + abstract asset OperatingSystem + {} + asset Windows extends OperatingSystem + { + | foundWindows @debug + } + asset Linux extends OperatingSystem + { + | foundLinux @debug + } + } + + associations + { + Computer [computers] * <-- L --> 0..1 [operatingSystems] OperatingSystem + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Computer', 'Windows', 'Linux', 'OperatingSystem'] + ) + +def test_operation_6() -> None: + ''' + This test might be invalid. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Computer + { + let allTargets = computers.operatingSystems[Windows] \\/ computers.operatingSystems[Linux] + | findWindowsAndLinux + -> allTargets[Windows].foundLinux + } + + abstract asset OperatingSystem + {} + asset Windows extends OperatingSystem + { + | foundWindows @debug + } + asset Linux extends OperatingSystem + { + | foundLinux @debug + } + } + + associations + { + Computer [computers] * <-- L --> 0..1 [operatingSystems] OperatingSystem + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Computer', 'Windows', 'Linux', 'OperatingSystem'] + ) + +def test_operation_7() -> None: + ''' + This test might be invalid. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Computer + { + let allTargets = computers.operatingSystems[Windows] \\/ computers.operatingSystems[Linux] + | findWindowsAndLinux + -> allTargets[Windowss].foundWindows + } + + abstract asset OperatingSystem + {} + asset Windows extends OperatingSystem + { + | foundWindows @debug + } + asset Linux extends OperatingSystem + { + | foundLinux @debug + } + } + + associations + { + Computer [computers] * <-- L --> 0..1 [operatingSystems] OperatingSystem + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Computer', 'Windows', 'Linux', 'OperatingSystem'] + ) diff --git a/maltoolbox/language/tests/test_mal_probability_distributions.py b/maltoolbox/language/tests/test_mal_probability_distributions.py new file mode 100644 index 00000000..641af13e --- /dev/null +++ b/maltoolbox/language/tests/test_mal_probability_distributions.py @@ -0,0 +1,245 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + +def test_probability_distributions_1() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem {} + asset Linux extends OperatingSystem { + # defense [Enabled] + | havePassword + -> login + | findPassword [Exponential(0.1)] + -> login + & login + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem', 'Linux'] + ) + +def test_probability_distributions_2() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem {} + asset Linux extends OperatingSystem { + & login [Enabled] + } + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem', 'Linux'] + ) + +def test_probability_distributions_3() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem {} + asset Linux extends OperatingSystem { + | login [Enabled] + } + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem', 'Linux'] + ) + +def test_probability_distributions_4() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem {} + asset Linux extends OperatingSystem { + | login [Disabled] + } + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem', 'Linux'] + ) + +def test_probability_distributions_5() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem {} + asset Linux extends OperatingSystem { + & login [Disabled] + } + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem', 'Linux'] + ) + +def test_probability_distributions_6() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem {} + asset Linux extends OperatingSystem { + | login [0.9] + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem', 'Linux'] + ) + +def test_probability_distributions_7() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem {} + asset Linux extends OperatingSystem { + | login [Bernoulli(0.5) * Exponential(0.1)] + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem', 'Linux'] + ) + +def test_probability_distributions_8() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem {} + asset Linux extends OperatingSystem { + | login [Exponential(0.5) * Exponential(0.1)] + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem', 'Linux'] + ) + +def test_probability_distributions_9() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem {} + asset Linux extends OperatingSystem { + | login [Exponential(0.5) / Exponential(0.1)] + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem', 'Linux'] + ) + +def test_probability_distributions_10() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem {} + asset Linux extends OperatingSystem { + | login [Exponential(0.5) ^ Exponential(0.1)] + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem', 'Linux'] + ) + +def test_probability_distributions_11() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem {} + asset Linux extends OperatingSystem { + | login [Exponential(0.5) + Exponential(0.1)] + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem', 'Linux'] + ) + +def test_probability_distributions_12() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem {} + asset Linux extends OperatingSystem { + | login [Exponential(0.5) - Exponential(0.1)] + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem', 'Linux'] + ) + +def test_probability_distributions_13() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem {} + asset Linux extends OperatingSystem { + | login [Bernoulli(0.5) ^ Exponential(0.1)] + } + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem', 'Linux'] + ) \ No newline at end of file diff --git a/maltoolbox/language/tests/test_mal_risk_type.py b/maltoolbox/language/tests/test_mal_risk_type.py new file mode 100644 index 00000000..9d00c338 --- /dev/null +++ b/maltoolbox/language/tests/test_mal_risk_type.py @@ -0,0 +1,38 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + +''' +A file to test different cases of the Risk type (C, I, A) instruction in MAL. +''' + +def test_user_info_1() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Defines risk types. + define I twice + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset CIA_TEST + { + | readOnly {C} + | readAndAppend {C, I, I} + | fullAccess {C, I, A} + } + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['CIA_TEST'] + ) + + + + + diff --git a/maltoolbox/language/tests/test_mal_step.py b/maltoolbox/language/tests/test_mal_step.py new file mode 100644 index 00000000..72d0d14b --- /dev/null +++ b/maltoolbox/language/tests/test_mal_step.py @@ -0,0 +1,293 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + +''' +A file to test different cases of the OR-step `|` instruction in MAL. +''' + +import os + +def test_step_1() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Defines steps with name. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Test + { + | step1 + | step2 + | step3 + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Test'] + ) + +def test_step_2() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Defines steps with same name and other type. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Test + { + | step1 + & step1 + } + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Test'] + ) + + +def test_step_3() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Defines steps with same name and same type. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Test + { + | step1 + | step1 + } + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Test'] + ) + +def test_step_4() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Test + { + | step1 + -> step3 + | step2 + -> step3 + | step3 + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Test'] + ) + +def test_step_5() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Test + { + | guessPassword + +> authenticate + | stealPassword + -> authenticate + } + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Test'] + ) + + +def test_step_6() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem + { + | spyware + -> logKeystrokes + | logKeystrokes + } + + asset Linux + { + | spyware + +> readBashHistory + & readBashHistory + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Linux', 'OperatingSystem'] + ) + +def test_step_7() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem + { + | spyware + } + + asset Linux + { + & spyware + } + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Linux', 'OperatingSystem'] + ) + +def test_step_8() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset OperatingSystem + { + | obtainPassword + -> authenticate + | obtainMFAToken + -> authenticate + & authenticate + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['OperatingSystem'] + ) + +def test_step_9() -> None: + path = "./generated_test_mal.mal" + with open(path, 'w') as file: + file.write(''' + category System { + asset OperatingSystem { + | spyware + -> logKeystrokes + | logKeystrokes + } + } + ''') + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + include "./generated_test_mal.mal" + + category System { + asset Linux { + | spyware + +> readBashHistory + & readBashHistory + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Linux', 'OperatingSystem'] + ) + + if os.path.exists(path): + os.remove(path) + +def test_step_10() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + + category System { + asset Linux { + E hasCamera + <- hardware[Camera] + -> hijackCamera + | hijackCamera + } + asset Camera { + | photo + } + } + + associations + { + Linux [linux] * <-- L --> * [hardware] Camera + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Linux', 'Camera'] + ) \ No newline at end of file diff --git a/maltoolbox/language/tests/test_mal_user_info.py b/maltoolbox/language/tests/test_mal_user_info.py new file mode 100644 index 00000000..5b90c465 --- /dev/null +++ b/maltoolbox/language/tests/test_mal_user_info.py @@ -0,0 +1,83 @@ +from .mal_analyzer_test_wrapper import AnalyzerTestWrapper + +''' +A file to test different cases of the `user info` instruction in MAL. +''' + +def test_user_info_1() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Defines user info. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Foo + user info: "Hello" + {} + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Foo'] + ) + +def test_user_info_2() -> None: + ''' + Defines correct version and ID. + Defines category with name. + Defines asset with name. + Defines user info twice. + ''' + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Foo + user info: "Hello" + user info: "Hello" + {} + } + + ''').test( + error=True, + defines=['id', 'version'], + categories=['System'], + assets=['Foo'] + ) + +def test_user_info_3() -> None: + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category Example + user info: "The 'Example' category contains example assets" + { + asset Asset1 + modeler info: "Connect the attacker to this asset" + { + | attack + modeler info: "This should be the entry point for the attacker" + developer info: "This attack step is used to reach compromise on all sub assets" + -> subAsset*.compromise + | compromise + user info: "The attacker has full access on this asset" + } + } + + associations { + Asset1 [superAsset] 0..1 <-- Hierarchy --> 0..1 [subAsset] Asset1 + user info: "Assets can be connected in a hierarchy" + } + ''').test( + defines=['id', 'version'], + categories=['Example'], + assets=['Asset1'] + ) \ No newline at end of file