From 08c32e867c668eefaac6ddb32fdc99c047fad5de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lukas=20Bergstr=C3=B6m?= Date: Mon, 12 Feb 2024 18:54:10 +0100 Subject: [PATCH 1/6] Analyzer/Checker draft --- .../language/lexer_parser/mal_analyzer.py | 235 ++++++++++++++++++ .../language/lexer_parser/mal_visitor.py | 13 +- 2 files changed, 244 insertions(+), 4 deletions(-) create mode 100644 maltoolbox/language/lexer_parser/mal_analyzer.py diff --git a/maltoolbox/language/lexer_parser/mal_analyzer.py b/maltoolbox/language/lexer_parser/mal_analyzer.py new file mode 100644 index 00000000..af5237a5 --- /dev/null +++ b/maltoolbox/language/lexer_parser/mal_analyzer.py @@ -0,0 +1,235 @@ +from antlr4 import ParseTreeVisitor +from .mal_parser import malParser + +from dataclasses import dataclass +import logging +import re + +class malAnalyzer(ParseTreeVisitor): + ''' + A class to preform syntax-checks for MAL. + ''' + + class Analyze: + ''' + A decoration to insert `super().(...)` in child. + If `malAnalyzer` haven't implemented that method an error message + will occur and a raise. + ''' + def __init__(self, func): + self.func = func + + def __get__(self, instance, owner): + def wrapper(*args, **kwargs): + super_method = getattr(super(owner, instance), self.func.__name__, None) + # Check if method is implemented. + if super_method: + super_method(*args, **kwargs) + else: + logging.error(f'There is no analyzer implemented for \'{self.func.__name__}\'') + raise + + return self.func(instance, *args, **kwargs) + return wrapper + + @dataclass + class ContextWrapper(): + ''' + A wrapper to store context + for post-analysis. + ''' + name: str + ctx: any + def __str__(self) -> str: + return self.name + + @dataclass + class AssetWrapper(ContextWrapper): + ''' + A wrapper to store assets. + ''' + attack_steps: [malParser.StepContext] + + def __init__(self, *args, **kwargs) -> None: + self._error: bool = False + self._defines: dict[str, self.ContextWrapper] = {} + self._assets: dict[str, self.AssetWrapper ] = {} + self._metas: dict[str, self.ContextWrapper] = {} + + super().__init__(*args, **kwargs) + + def visit(self, ctx: malParser.MalContext) -> any: + ''' + Override ParseTreeVisitor `visit` to be able + to preform `__post_analysis` as last step. + ''' + result: any = super().visit(ctx) + if ctx.depth() == 1: + self.__post_analysis() + return result + + def is_valid(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_steps() + + + def __analyse_defines(self) -> None: + ''' + Check for mandatory defines: ID & Version + ''' + + if 'id' in self._defines: + value: str = self._defines['id'].ctx.STRING().getText()[1:-1] + if(len(value.strip()) == 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'].ctx.STRING().getText()[1:-1] + 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.getChild(3).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 + print(parent_ctx.ID()[0].getText()) + 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_steps(self): + logging.warning('__analyse_steps not implemented.') + # for asset in self._assets: + # pass + + def __get_assets_extendee(self, ctx: malParser.AssetContext) -> malParser.AssetContext: + if (ctx.EXTENDS()): + return self._assets[ctx.ID()[1].getText()].ctx + return None + + def visitDefine(self, ctx: malParser.DefineContext) -> None: + define_id: str = ctx.ID().getText() + + # Check if define was previously defined. + if define_id in self._defines: + prevDef = self._defines[define_id].ctx.start.line + logging.error(f'Define \'{define_id}\' previously defined at {prevDef}') + self._error = True + return + + self._defines[define_id] = self.ContextWrapper(ctx=ctx, name=define_id) + + def visitCategory(self, ctx: malParser.CategoryContext) -> None: + category_name: str = ctx.ID().getText() + if (len(ctx.asset()) == 0 and len(ctx.meta()) == 0): + logging.error(f'Category \'{category_name}\' contains no assets or metadata') + self._error = True + return + + def visitAsset(self, ctx: malParser.AssetContext) -> None: + asset_id: str = ctx.ID()[0].getText() + + ''' + # Is this possible to check? + category: malParser.CategoryContext = ctx.parentCtx + if not category: + logging.error(f'Asset \'{asset_id}\' is outside an category') + ''' + + # Check if asset was previously defined. + if asset_id in self._assets: + prevDef = self._assets[asset_id].ctx.start.line + logging.error(f"Asset '{asset_id}' previously defined at {prevDef}") + self._error = True + return + else: + self._assets[asset_id] = self.AssetWrapper(ctx=ctx, name=asset_id, attack_steps=[]) + + def visitMeta(self, ctx: malParser.MetaContext) -> None: + meta_id: str = ctx.ID().getText() + + # Check if meta was previously defined. + if meta_id in self._metas: + prevDef = self._metas[meta_id].ctx.start.line + logging.error(f"Metadata '{meta_id}' previously defined at {prevDef}") + self._error = True + return + else: + self._metas[meta_id] = self.ContextWrapper(ctx=ctx, name=meta_id) + + def visitStep(self, ctx: malParser.StepContext) -> None: + logging.warn('visitStep not implemented.') + # step_id: str = ctx.ID().getText() + # parent_asset_name = ctx.parentCtx.ID()[0].getText() + # if parent_asset_name in self._assets: + # logging.error(f'Asset \'{parent_asset_name}\' was not found.') + # self._error = True + # return + + # parent_asset: self.AssetWrapper = self._assets[parent_asset_name] + # Can a step be define twice? + # parent_asset.attack_steps.append(ctx) diff --git a/maltoolbox/language/lexer_parser/mal_visitor.py b/maltoolbox/language/lexer_parser/mal_visitor.py index f9ea74b4..ce4bd8dd 100644 --- a/maltoolbox/language/lexer_parser/mal_visitor.py +++ b/maltoolbox/language/lexer_parser/mal_visitor.py @@ -1,6 +1,6 @@ -from antlr4 import ParseTreeVisitor from .mal_parser import malParser +from .mal_analyzer import malAnalyzer from collections.abc import MutableMapping, MutableSequence @@ -9,7 +9,7 @@ # - ctx.two() would be [] -class malVisitor(ParseTreeVisitor): +class malVisitor(malAnalyzer): def __init__(self, compiler, *args, **kwargs): self.compiler = compiler self.current_file = compiler.current_file # for debug purposes @@ -61,10 +61,12 @@ def visitMal(self, ctx): def visitInclude(self, ctx): return ("include", ctx.STRING().getText().strip('"')) - + + @malAnalyzer.Analyze def visitDefine(self, ctx): return ("defines", {ctx.ID().getText(): ctx.STRING().getText().strip('"')}) + @malAnalyzer.Analyze def visitCategory(self, ctx): category = {} category["name"] = ctx.ID().getText() @@ -74,9 +76,11 @@ def visitCategory(self, ctx): return ("categories", ([category], assets)) + @malAnalyzer.Analyze def visitMeta(self, ctx): return ((ctx.ID().getText(), ctx.STRING().getText().strip('"')),) + @malAnalyzer.Analyze def visitAsset(self, ctx): asset = {} asset["name"] = ctx.ID()[0].getText() @@ -92,7 +96,8 @@ def visitAsset(self, ctx): asset["attackSteps"] = [self.visit(step) for step in ctx.step()] return asset - + + @malAnalyzer.Analyze def visitStep(self, ctx): step = {} step["name"] = ctx.ID().getText() From a81c5e2963251d0ac8008d8e4ca5b9ffdbe4d622 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lukas=20Bergstr=C3=B6m?= Date: Sun, 18 Feb 2024 18:57:08 +0100 Subject: [PATCH 2/6] Redesign of analyzer (Draft) (WIP) --- maltoolbox/language/lexer_parser/__main__.py | 5 +- .../language/lexer_parser/mal_analyzer.py | 271 ++++++++++-------- .../language/lexer_parser/mal_visitor.py | 10 +- 3 files changed, 156 insertions(+), 130 deletions(-) diff --git a/maltoolbox/language/lexer_parser/__main__.py b/maltoolbox/language/lexer_parser/__main__.py index 9086db5c..9cbdf543 100644 --- a/maltoolbox/language/lexer_parser/__main__.py +++ b/maltoolbox/language/lexer_parser/__main__.py @@ -3,7 +3,8 @@ from antlr4 import FileStream, CommonTokenStream from .mal_lexer import malLexer from .mal_parser import malParser -from .mal_visitor import malVisitor +# from .mal_visitor import malVisitor +from .mal_analyzer import malAnalyzer import sys import os @@ -29,7 +30,7 @@ def compile(self, malfile: str = None): parser = malParser(stream) tree = parser.mal() - return malVisitor(compiler=self).visit(tree) + return malAnalyzer(compiler=self).visit(tree) if __name__ == "__main__": diff --git a/maltoolbox/language/lexer_parser/mal_analyzer.py b/maltoolbox/language/lexer_parser/mal_analyzer.py index af5237a5..90ef2354 100644 --- a/maltoolbox/language/lexer_parser/mal_analyzer.py +++ b/maltoolbox/language/lexer_parser/mal_analyzer.py @@ -1,73 +1,23 @@ -from antlr4 import ParseTreeVisitor from .mal_parser import malParser +from .mal_visitor import malVisitor -from dataclasses import dataclass import logging import re -class malAnalyzer(ParseTreeVisitor): +class malAnalyzer(malVisitor): ''' A class to preform syntax-checks for MAL. ''' - - class Analyze: - ''' - A decoration to insert `super().(...)` in child. - If `malAnalyzer` haven't implemented that method an error message - will occur and a raise. - ''' - def __init__(self, func): - self.func = func - - def __get__(self, instance, owner): - def wrapper(*args, **kwargs): - super_method = getattr(super(owner, instance), self.func.__name__, None) - # Check if method is implemented. - if super_method: - super_method(*args, **kwargs) - else: - logging.error(f'There is no analyzer implemented for \'{self.func.__name__}\'') - raise - - return self.func(instance, *args, **kwargs) - return wrapper - - @dataclass - class ContextWrapper(): - ''' - A wrapper to store context - for post-analysis. - ''' - name: str - ctx: any - def __str__(self) -> str: - return self.name - - @dataclass - class AssetWrapper(ContextWrapper): - ''' - A wrapper to store assets. - ''' - attack_steps: [malParser.StepContext] def __init__(self, *args, **kwargs) -> None: self._error: bool = False - self._defines: dict[str, self.ContextWrapper] = {} - self._assets: dict[str, self.AssetWrapper ] = {} - self._metas: dict[str, self.ContextWrapper] = {} - + self._defines: dict = {} + self._assets: dict = {} + self._category: dict = {} + self._metas: dict = {} + self._steps: dict = {} super().__init__(*args, **kwargs) - def visit(self, ctx: malParser.MalContext) -> any: - ''' - Override ParseTreeVisitor `visit` to be able - to preform `__post_analysis` as last step. - ''' - result: any = super().visit(ctx) - if ctx.depth() == 1: - self.__post_analysis() - return result - def is_valid(self) -> bool: return self._error @@ -80,17 +30,15 @@ def __post_analysis(self) -> None: self.__analyse_extends() self.__analyse_abstract() self.__analyse_parents() - self.__analyse_steps() - def __analyse_defines(self) -> None: ''' Check for mandatory defines: ID & Version ''' - if 'id' in self._defines: - value: str = self._defines['id'].ctx.STRING().getText()[1:-1] - if(len(value.strip()) == 0): + if 'id' in self._defines.keys(): + define_value: str = self._defines['id']['obj']['id'] + if(len(define_value) == 0): logging.error('Define \'id\' cannot be empty') self._error = True else: @@ -98,7 +46,7 @@ def __analyse_defines(self) -> None: self._error = True if 'version' in self._defines: - version: str = self._defines['version'].ctx.STRING().getText()[1:-1] + version: str = self._defines['version']['obj']['version'] 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 @@ -110,9 +58,9 @@ 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 + asset_context: malParser.AssetContext = self._assets[asset]['ctx'] if(asset_context.EXTENDS()): - extend_asset_name = asset_context.getChild(3).getText() + 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 @@ -127,15 +75,14 @@ def __analyse_extends(self) -> None: def __analyse_abstract(self) -> None: for parent in self._assets: - parent_ctx: malParser.AssetContext = self._assets[parent].ctx - print(parent_ctx.ID()[0].getText()) + 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 + 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 @@ -146,7 +93,7 @@ def __analyse_parents(self) -> None: error: bool = False for asset in self._assets: parents: list[str] = [] - parent_ctx: malParser.AssetContext = self._assets[asset].ctx + 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): @@ -161,75 +108,157 @@ def __analyse_parents(self) -> None: self._error = True raise - def __analyse_steps(self): - logging.warning('__analyse_steps not implemented.') - # for asset in self._assets: - # pass - def __get_assets_extendee(self, ctx: malParser.AssetContext) -> malParser.AssetContext: if (ctx.EXTENDS()): - return self._assets[ctx.ID()[1].getText()].ctx + return self._assets[ctx.ID()[1].getText()]['ctx'] return None + + def visitMal(self, ctx: malParser.MalContext) -> None: + result = super().visitMal(ctx) + self.__post_analysis() + return result def visitDefine(self, ctx: malParser.DefineContext) -> None: - define_id: str = ctx.ID().getText() + result = super().visitDefine(ctx) + [_, obj] = result - # Check if define was previously defined. - if define_id in self._defines: - prevDef = self._defines[define_id].ctx.start.line - logging.error(f'Define \'{define_id}\' previously defined at {prevDef}') + if(len(obj.keys()) != 1): + raise + + define_id = next(iter(obj)) + if(define_id in self._defines.keys()): + prev_define_line = self._defines[define_id]['ctx'].start.line + logging.error(f'Define \'{define_id}\' previously defined at line {prev_define_line}') self._error = True - return + return result - self._defines[define_id] = self.ContextWrapper(ctx=ctx, name=define_id) - + self._defines[define_id] = {'ctx': ctx, 'obj': obj} + return result + def visitCategory(self, ctx: malParser.CategoryContext) -> None: - category_name: str = ctx.ID().getText() - if (len(ctx.asset()) == 0 and len(ctx.meta()) == 0): - logging.error(f'Category \'{category_name}\' contains no assets or metadata') + result = super().visitCategory(ctx) + + category = result[1][0][0] + if(category['name'] == '>'): + category_line = ctx.start.line + logging.error(f'Category has no name at line {category_line}') self._error = True - return + return result + + if len(category['meta']) == 0 and len(result[1][1]) == 0: + logging.warning(f'Category \'{category["name"]}\' contains no assets or metadata') + self._error = True + return result + + self._category[category['name']] = {'ctx': ctx, 'obj': result} + + return result def visitAsset(self, ctx: malParser.AssetContext) -> None: - asset_id: str = ctx.ID()[0].getText() + result = super().visitAsset(ctx) + asset_name = result['name'] + category_name = ctx.parentCtx.ID() - ''' - # Is this possible to check? - category: malParser.CategoryContext = ctx.parentCtx - if not category: - logging.error(f'Asset \'{asset_id}\' is outside an category') - ''' - - # Check if asset was previously defined. - if asset_id in self._assets: - prevDef = self._assets[asset_id].ctx.start.line - logging.error(f"Asset '{asset_id}' previously defined at {prevDef}") + # Check if asset was previously defined in same category. + if asset_name in self._assets.keys() and self._assets[asset_name]['parent']['name'] == 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 + return result else: - self._assets[asset_id] = self.AssetWrapper(ctx=ctx, name=asset_id, attack_steps=[]) + self._assets[asset_name] = {'ctx': ctx, 'obj': result, 'parent': {'name': ctx.parentCtx.ID() ,'ctx': ctx.parentCtx}} + return result def visitMeta(self, ctx: malParser.MetaContext) -> None: - meta_id: str = ctx.ID().getText() - - # Check if meta was previously defined. - if meta_id in self._metas: - prevDef = self._metas[meta_id].ctx.start.line - logging.error(f"Metadata '{meta_id}' previously defined at {prevDef}") + result = super().visitMeta(ctx) + meta_name = result[0][0] + parent_name = '' + location_name = '' + + # Finding metadata type + if isinstance(ctx.parentCtx, malParser.CategoryContext): + parent_name = ctx.parentCtx.ID() + location_name = 'category' + elif isinstance(ctx.parentCtx, malParser.AssetContext): + parent_name = ctx.parentCtx.ID()[0] + location_name = 'asset' + elif isinstance(ctx.parentCtx, malParser.StepContext): + parent_name = ctx.parentCtx.ID() + location_name = 'step' + + # 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 + + # TODO: check for Associations + + return result + + + def visitStep(self, ctx: malParser.StepContext): + step = super().visitStep(ctx) + step_name = step['name'] + + if isinstance(ctx.parentCtx, malParser.AssetContext): + asset_name = ctx.parentCtx.ID()[0] + + # TODO: Validate step + # if (not asset_name in self._steps.keys()): + # self._steps[asset_name] = {step_name: ctx} + self._validate_CIA(ctx, step) + self._validate_TTC(ctx, step) + + return step + + def _validate_TTC(self, ctx: malParser.StepContext, step): + if not step['ttc']: return - else: - self._metas[meta_id] = self.ContextWrapper(ctx=ctx, name=meta_id) - - def visitStep(self, ctx: malParser.StepContext) -> None: - logging.warn('visitStep not implemented.') - # step_id: str = ctx.ID().getText() - # parent_asset_name = ctx.parentCtx.ID()[0].getText() - # if parent_asset_name in self._assets: - # logging.error(f'Asset \'{parent_asset_name}\' was not found.') - # self._error = True - # return - # parent_asset: self.AssetWrapper = self._assets[parent_asset_name] - # Can a step be define twice? - # parent_asset.attack_steps.append(ctx) + if step['type'] == 'defense': + # TODO: TTCFuncExpr + # if !(ttc instanceof AST.TTCFuncExpr) + # error + # elif fname = Enabled, Disabled, Bernoulli + # Distributions.validate(fname, fparams); + # else + # ERROR Defense %s.%s may only have 'Enabled', 'Disabled', or 'Bernoulli(p)' as TTC" + pass + + def visitTtcexpr(self, ctx: malParser.TtcexprContext): + pass + + def _validate_CIA(self, ctx: malParser.StepContext, step): + if not ctx.cias(): + return + + step_name = step['name'] + asset_name = ctx.parentCtx.ID()[0] + + 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 \ 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 ce4bd8dd..06c19c07 100644 --- a/maltoolbox/language/lexer_parser/mal_visitor.py +++ b/maltoolbox/language/lexer_parser/mal_visitor.py @@ -1,15 +1,16 @@ from .mal_parser import malParser -from .mal_analyzer import malAnalyzer +from antlr4 import ParseTreeVisitor from collections.abc import MutableMapping, MutableSequence + # In a rule like `rule: one? two* three`: # - ctx.one() would be None if the token was not found on a matching line # - ctx.two() would be [] -class malVisitor(malAnalyzer): +class malVisitor(ParseTreeVisitor): def __init__(self, compiler, *args, **kwargs): self.compiler = compiler self.current_file = compiler.current_file # for debug purposes @@ -62,11 +63,9 @@ def visitMal(self, ctx): def visitInclude(self, ctx): return ("include", ctx.STRING().getText().strip('"')) - @malAnalyzer.Analyze def visitDefine(self, ctx): return ("defines", {ctx.ID().getText(): ctx.STRING().getText().strip('"')}) - @malAnalyzer.Analyze def visitCategory(self, ctx): category = {} category["name"] = ctx.ID().getText() @@ -76,11 +75,9 @@ def visitCategory(self, ctx): return ("categories", ([category], assets)) - @malAnalyzer.Analyze def visitMeta(self, ctx): return ((ctx.ID().getText(), ctx.STRING().getText().strip('"')),) - @malAnalyzer.Analyze def visitAsset(self, ctx): asset = {} asset["name"] = ctx.ID()[0].getText() @@ -97,7 +94,6 @@ def visitAsset(self, ctx): return asset - @malAnalyzer.Analyze def visitStep(self, ctx): step = {} step["name"] = ctx.ID().getText() From 3330a0611104ca97110873e0c0835bfada887798 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lukas=20Bergstr=C3=B6m?= Date: Fri, 15 Mar 2024 13:22:11 +0100 Subject: [PATCH 3/6] Draft update, added tests --- maltoolbox/language/lexer_parser/__main__.py | 5 +- .../language/lexer_parser/mal_analyzer.py | 140 +++--- .../language/lexer_parser/mal_visitor.py | 20 +- maltoolbox/language/tests/__init__.py | 0 .../language/tests/test_mal_analyzer.py | 422 ++++++++++++++++++ 5 files changed, 517 insertions(+), 70 deletions(-) create mode 100644 maltoolbox/language/tests/__init__.py create mode 100644 maltoolbox/language/tests/test_mal_analyzer.py diff --git a/maltoolbox/language/lexer_parser/__main__.py b/maltoolbox/language/lexer_parser/__main__.py index 9cbdf543..eabcc37d 100644 --- a/maltoolbox/language/lexer_parser/__main__.py +++ b/maltoolbox/language/lexer_parser/__main__.py @@ -3,7 +3,7 @@ from antlr4 import FileStream, CommonTokenStream from .mal_lexer import malLexer from .mal_parser import malParser -# from .mal_visitor import malVisitor +from .mal_visitor import malVisitor from .mal_analyzer import malAnalyzer import sys @@ -30,7 +30,8 @@ def compile(self, malfile: str = None): parser = malParser(stream) tree = parser.mal() - return malAnalyzer(compiler=self).visit(tree) + analyzer = malAnalyzer() + return malVisitor(compiler=self, analyzer=analyzer).visit(tree) if __name__ == "__main__": diff --git a/maltoolbox/language/lexer_parser/mal_analyzer.py b/maltoolbox/language/lexer_parser/mal_analyzer.py index 90ef2354..b0d20f35 100644 --- a/maltoolbox/language/lexer_parser/mal_analyzer.py +++ b/maltoolbox/language/lexer_parser/mal_analyzer.py @@ -1,10 +1,25 @@ from .mal_parser import malParser -from .mal_visitor import malVisitor import logging import re - -class malAnalyzer(malVisitor): + +class malAnalyzerInterface: + def checkMal(self, ctx: malParser.MalContext) -> None: + pass + def checkDefine(self, ctx: malParser.DefineContext, obj: any) -> None: + pass + def checkCategory(self, ctx: malParser.CategoryContext, category, assets) -> None: + pass + def checkAsset(self, ctx: malParser.AssetContext, asset) -> None: + pass + def checkMeta(self, ctx: malParser.MetaContext, meta_name: str) -> None: + pass + def checkStep(self, ctx: malParser.StepContext, step) -> None: + pass + def checkVariable(self, ctx: malParser.VariableContext, var) -> None: + pass + +class malAnalyzer(malAnalyzerInterface): ''' A class to preform syntax-checks for MAL. ''' @@ -16,22 +31,24 @@ def __init__(self, *args, **kwargs) -> None: self._category: dict = {} self._metas: dict = {} self._steps: dict = {} + self._vars: dict = {} + self._error = False super().__init__(*args, **kwargs) - def is_valid(self) -> bool: + def has_error(self) -> bool: return self._error - def __post_analysis(self) -> None: + 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_defines() + self._analyse_extends() + self._analyse_abstract() + self._analyse_parents() - def __analyse_defines(self) -> None: + def _analyse_defines(self) -> None: ''' Check for mandatory defines: ID & Version ''' @@ -54,7 +71,7 @@ def __analyse_defines(self) -> None: logging.error('Missing required define \'#version: ""\'') self._error = True - def __analyse_extends(self) -> None: + def _analyse_extends(self) -> None: raise_error: bool = False extend_asset_name: str = '' for asset in self._assets: @@ -73,7 +90,7 @@ def __analyse_extends(self) -> None: self._error = True # Maybe unnecessary if we raise raise SyntaxError(f'Asset \'{extend_asset_name}\' not defined') - def __analyse_abstract(self) -> None: + def _analyse_abstract(self) -> None: for parent in self._assets: parent_ctx: malParser.AssetContext = self._assets[parent]['ctx'] if(parent_ctx.ABSTRACT()): @@ -89,7 +106,7 @@ def __analyse_abstract(self) -> None: if not found: logging.warn(f'Asset \'{parent_ctx.ID()[0].getText()}\' is abstract but never extended to') - def __analyse_parents(self) -> None: + def _analyse_parents(self) -> None: error: bool = False for asset in self._assets: parents: list[str] = [] @@ -103,25 +120,20 @@ def __analyse_parents(self) -> None: error = True break parents.append(parent_name) - parent_ctx = self.__get_assets_extendee(parent_ctx) + parent_ctx = self._get_assets_extendee(parent_ctx) if error: self._error = True raise - def __get_assets_extendee(self, ctx: malParser.AssetContext) -> malParser.AssetContext: + def _get_assets_extendee(self, ctx: malParser.AssetContext) -> malParser.AssetContext: if (ctx.EXTENDS()): return self._assets[ctx.ID()[1].getText()]['ctx'] return None - def visitMal(self, ctx: malParser.MalContext) -> None: - result = super().visitMal(ctx) - self.__post_analysis() - return result - - def visitDefine(self, ctx: malParser.DefineContext) -> None: - result = super().visitDefine(ctx) - [_, obj] = result + def checkMal(self, ctx: malParser.MalContext) -> None: + self._post_analysis() + def checkDefine(self, ctx: malParser.DefineContext, obj: any) -> None: if(len(obj.keys()) != 1): raise @@ -130,60 +142,50 @@ def visitDefine(self, ctx: malParser.DefineContext) -> None: prev_define_line = self._defines[define_id]['ctx'].start.line logging.error(f'Define \'{define_id}\' previously defined at line {prev_define_line}') self._error = True - return result + return self._defines[define_id] = {'ctx': ctx, 'obj': obj} - return result - def visitCategory(self, ctx: malParser.CategoryContext) -> None: - result = super().visitCategory(ctx) - - category = result[1][0][0] - if(category['name'] == '>'): + def checkCategory(self, ctx: malParser.CategoryContext, category, assets) -> None: + if(str(category['name']) == '>'): category_line = ctx.start.line logging.error(f'Category has no name at line {category_line}') self._error = True - return result + return - if len(category['meta']) == 0 and len(result[1][1]) == 0: + if len(category['meta']) == 0 and len(assets) == 0: logging.warning(f'Category \'{category["name"]}\' contains no assets or metadata') - self._error = True - return result - - self._category[category['name']] = {'ctx': ctx, 'obj': result} - - return result + # Warning might not be checked as error. + # self._error = True - def visitAsset(self, ctx: malParser.AssetContext) -> None: - result = super().visitAsset(ctx) - asset_name = result['name'] + self._category[category['name']] = {'ctx': ctx, 'obj': {'category': category, 'assets': assets}} + + def checkAsset(self, ctx: malParser.AssetContext, asset) -> None: + asset_name = asset['name'] category_name = ctx.parentCtx.ID() # Check if asset was previously defined in same category. - if asset_name in self._assets.keys() and self._assets[asset_name]['parent']['name'] == category_name: + 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 result + return else: - self._assets[asset_name] = {'ctx': ctx, 'obj': result, 'parent': {'name': ctx.parentCtx.ID() ,'ctx': ctx.parentCtx}} - return result + self._assets[asset_name] = {'ctx': ctx, 'obj': asset, 'parent': {'name': ctx.parentCtx.ID() ,'ctx': ctx.parentCtx}} - def visitMeta(self, ctx: malParser.MetaContext) -> None: - result = super().visitMeta(ctx) - meta_name = result[0][0] + def checkMeta(self, ctx: malParser.MetaContext, meta_name: str) -> None: parent_name = '' location_name = '' # Finding metadata type if isinstance(ctx.parentCtx, malParser.CategoryContext): - parent_name = ctx.parentCtx.ID() + parent_name = str(ctx.parentCtx.ID()) location_name = 'category' elif isinstance(ctx.parentCtx, malParser.AssetContext): - parent_name = ctx.parentCtx.ID()[0] + parent_name = str(ctx.parentCtx.ID()[0]) location_name = 'asset' elif isinstance(ctx.parentCtx, malParser.StepContext): - parent_name = ctx.parentCtx.ID() + parent_name = str(ctx.parentCtx.ID()) location_name = 'step' # Validate that the metadata is unique @@ -199,12 +201,8 @@ def visitMeta(self, ctx: malParser.MetaContext) -> None: self._error = True # TODO: check for Associations - - return result - - def visitStep(self, ctx: malParser.StepContext): - step = super().visitStep(ctx) + def checkStep(self, ctx: malParser.StepContext, step) -> None: step_name = step['name'] if isinstance(ctx.parentCtx, malParser.AssetContext): @@ -218,7 +216,7 @@ def visitStep(self, ctx: malParser.StepContext): return step - def _validate_TTC(self, ctx: malParser.StepContext, step): + def _validate_TTC(self, ctx: malParser.StepContext, step) -> None: if not step['ttc']: return @@ -232,10 +230,7 @@ def _validate_TTC(self, ctx: malParser.StepContext, step): # ERROR Defense %s.%s may only have 'Enabled', 'Disabled', or 'Bernoulli(p)' as TTC" pass - def visitTtcexpr(self, ctx: malParser.TtcexprContext): - pass - - def _validate_CIA(self, ctx: malParser.StepContext, step): + def _validate_CIA(self, ctx: malParser.StepContext, step) -> None: if not ctx.cias(): return @@ -261,4 +256,25 @@ def _validate_CIA(self, ctx: malParser.StepContext, step): self._error = True return cias.append(letter) - index += 1 \ No newline at end of file + index += 1 + + def checkVariable(self, ctx: malParser.VariableContext, var) -> 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} + elif (var_name not in self._vars[asset_name]): + self._vars[asset_name][var_name] = ctx + else: + prev_define_line = self._vars[asset_name][var_name].start.line + logging.error(f'Variable \'{var_name}\' previously defined at line {prev_define_line}') + self._error = True \ 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 06c19c07..cc3b0ec0 100644 --- a/maltoolbox/language/lexer_parser/mal_visitor.py +++ b/maltoolbox/language/lexer_parser/mal_visitor.py @@ -1,5 +1,6 @@ from .mal_parser import malParser +from .mal_analyzer import malAnalyzerInterface from antlr4 import ParseTreeVisitor from collections.abc import MutableMapping, MutableSequence @@ -11,10 +12,10 @@ 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 visitMal(self, ctx): @@ -58,13 +59,16 @@ def visitMal(self, ctx): unique.append(item) langspec[key] = unique + self.analyzer.checkMal(ctx) return langspec def visitInclude(self, ctx): return ("include", ctx.STRING().getText().strip('"')) def visitDefine(self, ctx): - return ("defines", {ctx.ID().getText(): ctx.STRING().getText().strip('"')}) + define_object = {ctx.ID().getText(): ctx.STRING().getText().strip('"')} + self.analyzer.checkDefine(ctx, define_object) + return ("defines", define_object) def visitCategory(self, ctx): category = {} @@ -72,10 +76,12 @@ def visitCategory(self, ctx): category["meta"] = {k: v for meta in ctx.meta() for k, v in self.visit(meta)} assets = [self.visit(asset) for asset in ctx.asset()] - + + self.analyzer.checkCategory(ctx, category, assets) return ("categories", ([category], assets)) def visitMeta(self, ctx): + self.analyzer.checkMeta(ctx, ctx.ID().getText()) return ((ctx.ID().getText(), ctx.STRING().getText().strip('"')),) def visitAsset(self, ctx): @@ -92,6 +98,7 @@ def visitAsset(self, ctx): asset["variables"] = [self.visit(variable) for variable in ctx.variable()] asset["attackSteps"] = [self.visit(step) for step in ctx.step()] + self.analyzer.checkAsset(ctx, asset) return asset def visitStep(self, ctx): @@ -106,7 +113,8 @@ def visitStep(self, ctx): self.visit(ctx.precondition()) if ctx.precondition() else None ) step["reaches"] = self.visit(ctx.reaches()) if ctx.reaches() else None - + + self.analyzer.checkStep(ctx, step) return step def visitSteptype(self, ctx): @@ -242,7 +250,7 @@ def visitVariable(self, ctx): ret = {} ret["name"] = ctx.ID().getText() ret["stepExpression"] = self.visit(ctx.expr()) - + self.analyzer.checkVariable(ctx, ret) return ret def visitExpr(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/test_mal_analyzer.py b/maltoolbox/language/tests/test_mal_analyzer.py new file mode 100644 index 00000000..1a98fbe0 --- /dev/null +++ b/maltoolbox/language/tests/test_mal_analyzer.py @@ -0,0 +1,422 @@ +import pytest + +from antlr4 import 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 + +class MockCompiler(): + def __init__(self): + self.path = None + self.current_file = None + +def compile_text(analyzer: malAnalyzer, input_string: str): + ''' + A function to test the compiler flow with data from a string. + ''' + input_stream = InputStream(input_string) + lexer = malLexer(input_stream) + stream = CommonTokenStream(lexer) + parser = malParser(stream) + tree = parser.mal() + compiler = MockCompiler() + return malVisitor(compiler=compiler, analyzer=analyzer).visit(tree) + +def test_construct(): + analyzer = malAnalyzer() + assert analyzer.has_error() == False + +def test_full_flow_empty(): + analyzer = malAnalyzer() + result = compile_text(analyzer, "") + assert analyzer.has_error() == True + +def test_full_flow_missing_define_id(): + analyzer = malAnalyzer() + input = """ + #version:"0.0.0" + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + defines = list(analyzer._defines.keys()) + assert defines == ['version'] + +def test_full_flow_missing_define_version(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + defines = list(analyzer._defines.keys()) + assert defines == ['id'] + +def test_full_flow_wrong_format_version(): + ''' + The analyzer will add 'version' to the define-dict, + but will fail during the post-analyzer stage. + ''' + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"version1" + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + defines = list(analyzer._defines.keys()) + assert defines == ['id', 'version'] + +def test_full_flow_correct_require_defines(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == False + defines = list(analyzer._defines.keys()) + assert defines == ['id', 'version'] + +def test_full_flow_prev_def_defines(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + #version:"1.0.0" + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + defines = list(analyzer._defines.keys()) + assert defines == ['id', 'version'] + +def test_full_flow_category_missing_name(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category { + + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + +def test_full_flow_correct_category(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == False + +def test_full_flow_prev_def_asset_1(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Test {} + asset Test {} + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + +def test_full_flow_prev_def_asset_2(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System {asset Test {}} + category System {asset Test {}} + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + defines = list(analyzer._defines.keys()) + assert defines == ['id', 'version'] + +def test_full_flow_correct_asset(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Test { + + } + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == False + +def test_full_flow_prev_def_category_meta_1(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System + User: "Owner" + User: "Borrow" + {} + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + +def test_full_flow_prev_def_category_meta_2(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System + User: "Owner" + {} + category System + User: "Borrow" + {} + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + +def test_full_flow_prev_def_asset_meta_1(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System + { + asset Test + User: "Owner" + User: "Borrow" + {} + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + +def test_full_flow_prev_def_asset_meta_1(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System + { + asset Test + User: "Borrow" + {} + } + + category System + { + asset Test + User: "Owner" + {} + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + + +def test_full_flow_prev_def_asset_meta_1(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System + { + asset Test + { + | Attack + User: "Attacking" + User: "Looking" + } + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + +def test_full_flow_ok_meta(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System + User: "Owner" + { + asset Network + User: "Owner" + { + | Attack + User: "Attacking" + } + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == False + +def test_full_flow_extends(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category Systems { + asset Foo1 extends Foo2 {} + } + """ + with pytest.raises(SyntaxError): + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + +def test_full_flow_circular_dependency(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + // Test Circular Dependency + category Systems { + asset Foo1 extends Foo2 {} + asset Foo2 extends Foo3 {} + asset Foo3 extends Foo4 {} + asset Foo4 extends Foo5 {} + asset Foo5 extends Foo1 {} + } + """ + with pytest.raises(Exception): + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + +def test_full_flow_CIA_fail(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category Systems { + asset CIA_TEST + { + | readOnly {C} + | readAndAppend {C, I, I} + | fullAccess {C, I, A} + } + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + +def test_full_flow_CIA_defense_fail(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category Systems { + asset CIA_TEST + { + # readOnly {C} + } + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + + +def test_full_flow_CIA_OK(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category Systems { + asset CIA_TEST + { + | readOnly {C} + | readAndAppend {C, I} + | fullAccess {C, I, A} + } + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == False + +def test_full_flow_var_1(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Computer { + let components = hardware + } + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == False + +def test_full_flow_var_2(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Computer { + let components = hardware + let components = software + } + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + +def test_full_flow_var_2(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Computer { + let components = hardware + } + } + category System { + asset Computer { + let components = software + } + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == True + +def test_full_flow_var_3(): + analyzer = malAnalyzer() + input = """ + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + category System { + asset Computer { + let components = software \\/ hardware + } + } + """ + result = compile_text(analyzer, input) + assert analyzer.has_error() == False \ No newline at end of file From dfa716f68f99675d2c8fe8b0c1358a9c1bb6303b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lukas=20Bergstr=C3=B6m?= Date: Mon, 26 Aug 2024 12:58:44 +0200 Subject: [PATCH 4/6] Generate calls to analyzers-method via visit-method. --- maltoolbox/language/lexer_parser/__main__.py | 5 +- .../language/lexer_parser/mal_analyzer.py | 93 +++- .../language/lexer_parser/mal_visitor.py | 30 +- .../tests/mal_analyzer_test_wrapper.py | 65 +++ .../language/tests/test_mal_abstract_asset.py | 47 ++ .../language/tests/test_mal_analyzer.py | 422 ------------------ maltoolbox/language/tests/test_mal_assets.py | 91 ++++ .../language/tests/test_mal_association.py | 83 ++++ .../language/tests/test_mal_category.py | 54 +++ maltoolbox/language/tests/test_mal_define.py | 91 ++++ .../language/tests/test_mal_developer_info.py | 53 +++ maltoolbox/language/tests/test_mal_extends.py | 78 ++++ maltoolbox/language/tests/test_mal_include.py | 153 +++++++ maltoolbox/language/tests/test_mal_let.py | 85 ++++ .../language/tests/test_mal_modeler_info.py | 53 +++ .../language/tests/test_mal_user_info.py | 53 +++ 16 files changed, 999 insertions(+), 457 deletions(-) create mode 100644 maltoolbox/language/tests/mal_analyzer_test_wrapper.py create mode 100644 maltoolbox/language/tests/test_mal_abstract_asset.py delete mode 100644 maltoolbox/language/tests/test_mal_analyzer.py create mode 100644 maltoolbox/language/tests/test_mal_assets.py create mode 100644 maltoolbox/language/tests/test_mal_association.py create mode 100644 maltoolbox/language/tests/test_mal_category.py create mode 100644 maltoolbox/language/tests/test_mal_define.py create mode 100644 maltoolbox/language/tests/test_mal_developer_info.py create mode 100644 maltoolbox/language/tests/test_mal_extends.py create mode 100644 maltoolbox/language/tests/test_mal_include.py create mode 100644 maltoolbox/language/tests/test_mal_let.py create mode 100644 maltoolbox/language/tests/test_mal_modeler_info.py create mode 100644 maltoolbox/language/tests/test_mal_user_info.py diff --git a/maltoolbox/language/lexer_parser/__main__.py b/maltoolbox/language/lexer_parser/__main__.py index eabcc37d..5f76ebdd 100644 --- a/maltoolbox/language/lexer_parser/__main__.py +++ b/maltoolbox/language/lexer_parser/__main__.py @@ -15,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: @@ -29,9 +30,7 @@ def compile(self, malfile: str = None): stream = CommonTokenStream(lexer) parser = malParser(stream) tree = parser.mal() - - analyzer = malAnalyzer() - return malVisitor(compiler=self, analyzer=analyzer).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 index b0d20f35..bb266da9 100644 --- a/maltoolbox/language/lexer_parser/mal_analyzer.py +++ b/maltoolbox/language/lexer_parser/mal_analyzer.py @@ -3,20 +3,26 @@ 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, obj: any) -> None: + def checkDefine(self, ctx: malParser.DefineContext, data: Tuple[str, Any]) -> 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 checkCategory(self, ctx: malParser.CategoryContext, category, assets) -> None: + def checkAsset(self, ctx: malParser.AssetContext, asset: dict) -> None: pass - def checkAsset(self, ctx: malParser.AssetContext, asset) -> None: + def checkMeta(self, ctx: malParser.MetaContext, data: Tuple[Tuple[str, str],]) -> None: pass - def checkMeta(self, ctx: malParser.MetaContext, meta_name: str) -> None: + def checkStep(self, ctx: malParser.StepContext, step: dict) -> None: pass - def checkStep(self, ctx: malParser.StepContext, step) -> None: + def checkVariable(self, ctx: malParser.VariableContext, var: dict) -> None: pass - def checkVariable(self, ctx: malParser.VariableContext, var) -> None: + def checkAssociation(self, ctx: malParser.AssociationContext, association: dict) -> None: pass class malAnalyzer(malAnalyzerInterface): @@ -25,16 +31,20 @@ class malAnalyzer(malAnalyzerInterface): ''' def __init__(self, *args, **kwargs) -> None: - self._error: bool = False - self._defines: dict = {} - self._assets: dict = {} - self._category: dict = {} - self._metas: dict = {} - self._steps: dict = {} - self._vars: dict = {} - self._error = False - super().__init__(*args, **kwargs) + 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 @@ -47,6 +57,7 @@ def _post_analysis(self) -> None: self._analyse_extends() self._analyse_abstract() self._analyse_parents() + self._analyse_association() def _analyse_defines(self) -> None: ''' @@ -125,15 +136,39 @@ def _analyse_parents(self) -> None: self._error = True raise + 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: - self._post_analysis() + ''' + 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, Any]) -> None: + _, obj = data - def checkDefine(self, ctx: malParser.DefineContext, obj: any) -> None: if(len(obj.keys()) != 1): raise @@ -146,7 +181,9 @@ def checkDefine(self, ctx: malParser.DefineContext, obj: any) -> None: self._defines[define_id] = {'ctx': ctx, 'obj': obj} - def checkCategory(self, ctx: malParser.CategoryContext, category, assets) -> None: + 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}') @@ -160,10 +197,14 @@ def checkCategory(self, ctx: malParser.CategoryContext, category, assets) -> Non self._category[category['name']] = {'ctx': ctx, 'obj': {'category': category, 'assets': assets}} - def checkAsset(self, ctx: malParser.AssetContext, asset) -> None: + def checkAsset(self, ctx: malParser.AssetContext, asset: dict) -> None: asset_name = asset['name'] category_name = ctx.parentCtx.ID() - + + 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 @@ -173,7 +214,8 @@ def checkAsset(self, ctx: malParser.AssetContext, asset) -> None: else: self._assets[asset_name] = {'ctx': ctx, 'obj': asset, 'parent': {'name': ctx.parentCtx.ID() ,'ctx': ctx.parentCtx}} - def checkMeta(self, ctx: malParser.MetaContext, meta_name: str) -> None: + def checkMeta(self, ctx: malParser.MetaContext, data: Tuple[Tuple[str, str],]) -> None: + ((meta_name, _),) = data parent_name = '' location_name = '' @@ -202,7 +244,7 @@ def checkMeta(self, ctx: malParser.MetaContext, meta_name: str) -> None: # TODO: check for Associations - def checkStep(self, ctx: malParser.StepContext, step) -> None: + def checkStep(self, ctx: malParser.StepContext, step: dict) -> None: step_name = step['name'] if isinstance(ctx.parentCtx, malParser.AssetContext): @@ -258,7 +300,7 @@ def _validate_CIA(self, ctx: malParser.StepContext, step) -> None: cias.append(letter) index += 1 - def checkVariable(self, ctx: malParser.VariableContext, var) -> None: + def checkVariable(self, ctx: malParser.VariableContext, var: dict) -> None: ''' self._vars = { : { @@ -277,4 +319,7 @@ def checkVariable(self, ctx: malParser.VariableContext, var) -> None: else: prev_define_line = self._vars[asset_name][var_name].start.line logging.error(f'Variable \'{var_name}\' previously defined at line {prev_define_line}') - self._error = True \ No newline at end of file + self._error = True + + 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 cc3b0ec0..fbe97d81 100644 --- a/maltoolbox/language/lexer_parser/mal_visitor.py +++ b/maltoolbox/language/lexer_parser/mal_visitor.py @@ -18,6 +18,28 @@ def __init__(self, compiler, analyzer: malAnalyzerInterface, *args, **kwargs): 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", @@ -59,7 +81,6 @@ def visitMal(self, ctx): unique.append(item) langspec[key] = unique - self.analyzer.checkMal(ctx) return langspec def visitInclude(self, ctx): @@ -67,7 +88,6 @@ def visitInclude(self, ctx): def visitDefine(self, ctx): define_object = {ctx.ID().getText(): ctx.STRING().getText().strip('"')} - self.analyzer.checkDefine(ctx, define_object) return ("defines", define_object) def visitCategory(self, ctx): @@ -76,12 +96,9 @@ def visitCategory(self, ctx): category["meta"] = {k: v for meta in ctx.meta() for k, v in self.visit(meta)} assets = [self.visit(asset) for asset in ctx.asset()] - - self.analyzer.checkCategory(ctx, category, assets) return ("categories", ([category], assets)) def visitMeta(self, ctx): - self.analyzer.checkMeta(ctx, ctx.ID().getText()) return ((ctx.ID().getText(), ctx.STRING().getText().strip('"')),) def visitAsset(self, ctx): @@ -98,7 +115,6 @@ def visitAsset(self, ctx): asset["variables"] = [self.visit(variable) for variable in ctx.variable()] asset["attackSteps"] = [self.visit(step) for step in ctx.step()] - self.analyzer.checkAsset(ctx, asset) return asset def visitStep(self, ctx): @@ -114,7 +130,6 @@ def visitStep(self, ctx): ) step["reaches"] = self.visit(ctx.reaches()) if ctx.reaches() else None - self.analyzer.checkStep(ctx, step) return step def visitSteptype(self, ctx): @@ -250,7 +265,6 @@ def visitVariable(self, ctx): ret = {} ret["name"] = ctx.ID().getText() ret["stepExpression"] = self.visit(ctx.expr()) - self.analyzer.checkVariable(ctx, ret) return ret def visitExpr(self, ctx): 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..3a69d604 --- /dev/null +++ b/maltoolbox/language/tests/mal_analyzer_test_wrapper.py @@ -0,0 +1,65 @@ + +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=[], steps: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) + if (steps): + assert(set(steps) == set(self._steps.keys())) + + + 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_analyzer.py b/maltoolbox/language/tests/test_mal_analyzer.py deleted file mode 100644 index 1a98fbe0..00000000 --- a/maltoolbox/language/tests/test_mal_analyzer.py +++ /dev/null @@ -1,422 +0,0 @@ -import pytest - -from antlr4 import 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 - -class MockCompiler(): - def __init__(self): - self.path = None - self.current_file = None - -def compile_text(analyzer: malAnalyzer, input_string: str): - ''' - A function to test the compiler flow with data from a string. - ''' - input_stream = InputStream(input_string) - lexer = malLexer(input_stream) - stream = CommonTokenStream(lexer) - parser = malParser(stream) - tree = parser.mal() - compiler = MockCompiler() - return malVisitor(compiler=compiler, analyzer=analyzer).visit(tree) - -def test_construct(): - analyzer = malAnalyzer() - assert analyzer.has_error() == False - -def test_full_flow_empty(): - analyzer = malAnalyzer() - result = compile_text(analyzer, "") - assert analyzer.has_error() == True - -def test_full_flow_missing_define_id(): - analyzer = malAnalyzer() - input = """ - #version:"0.0.0" - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - defines = list(analyzer._defines.keys()) - assert defines == ['version'] - -def test_full_flow_missing_define_version(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - defines = list(analyzer._defines.keys()) - assert defines == ['id'] - -def test_full_flow_wrong_format_version(): - ''' - The analyzer will add 'version' to the define-dict, - but will fail during the post-analyzer stage. - ''' - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"version1" - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - defines = list(analyzer._defines.keys()) - assert defines == ['id', 'version'] - -def test_full_flow_correct_require_defines(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == False - defines = list(analyzer._defines.keys()) - assert defines == ['id', 'version'] - -def test_full_flow_prev_def_defines(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - #version:"1.0.0" - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - defines = list(analyzer._defines.keys()) - assert defines == ['id', 'version'] - -def test_full_flow_category_missing_name(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category { - - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - -def test_full_flow_correct_category(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category System { - - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == False - -def test_full_flow_prev_def_asset_1(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category System { - asset Test {} - asset Test {} - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - -def test_full_flow_prev_def_asset_2(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category System {asset Test {}} - category System {asset Test {}} - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - defines = list(analyzer._defines.keys()) - assert defines == ['id', 'version'] - -def test_full_flow_correct_asset(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category System { - asset Test { - - } - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == False - -def test_full_flow_prev_def_category_meta_1(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category System - User: "Owner" - User: "Borrow" - {} - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - -def test_full_flow_prev_def_category_meta_2(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category System - User: "Owner" - {} - category System - User: "Borrow" - {} - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - -def test_full_flow_prev_def_asset_meta_1(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category System - { - asset Test - User: "Owner" - User: "Borrow" - {} - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - -def test_full_flow_prev_def_asset_meta_1(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category System - { - asset Test - User: "Borrow" - {} - } - - category System - { - asset Test - User: "Owner" - {} - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - - -def test_full_flow_prev_def_asset_meta_1(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category System - { - asset Test - { - | Attack - User: "Attacking" - User: "Looking" - } - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - -def test_full_flow_ok_meta(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category System - User: "Owner" - { - asset Network - User: "Owner" - { - | Attack - User: "Attacking" - } - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == False - -def test_full_flow_extends(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category Systems { - asset Foo1 extends Foo2 {} - } - """ - with pytest.raises(SyntaxError): - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - -def test_full_flow_circular_dependency(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - // Test Circular Dependency - category Systems { - asset Foo1 extends Foo2 {} - asset Foo2 extends Foo3 {} - asset Foo3 extends Foo4 {} - asset Foo4 extends Foo5 {} - asset Foo5 extends Foo1 {} - } - """ - with pytest.raises(Exception): - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - -def test_full_flow_CIA_fail(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category Systems { - asset CIA_TEST - { - | readOnly {C} - | readAndAppend {C, I, I} - | fullAccess {C, I, A} - } - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - -def test_full_flow_CIA_defense_fail(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category Systems { - asset CIA_TEST - { - # readOnly {C} - } - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - - -def test_full_flow_CIA_OK(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category Systems { - asset CIA_TEST - { - | readOnly {C} - | readAndAppend {C, I} - | fullAccess {C, I, A} - } - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == False - -def test_full_flow_var_1(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category System { - asset Computer { - let components = hardware - } - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == False - -def test_full_flow_var_2(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category System { - asset Computer { - let components = hardware - let components = software - } - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - -def test_full_flow_var_2(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category System { - asset Computer { - let components = hardware - } - } - category System { - asset Computer { - let components = software - } - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == True - -def test_full_flow_var_3(): - analyzer = malAnalyzer() - input = """ - #id: "org.mal-lang.testAnalyzer" - #version:"0.0.0" - - category System { - asset Computer { - let components = software \\/ hardware - } - } - """ - result = compile_text(analyzer, input) - assert analyzer.has_error() == False \ No newline at end of file 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..c86b6eb8 --- /dev/null +++ b/maltoolbox/language/tests/test_mal_association.py @@ -0,0 +1,83 @@ +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'] + ) \ 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..04f7f3cc --- /dev/null +++ b/maltoolbox/language/tests/test_mal_let.py @@ -0,0 +1,85 @@ +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')] + ) \ 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_user_info.py b/maltoolbox/language/tests/test_mal_user_info.py new file mode 100644 index 00000000..a58ae2c7 --- /dev/null +++ b/maltoolbox/language/tests/test_mal_user_info.py @@ -0,0 +1,53 @@ +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'] + ) \ No newline at end of file From c591539be5c9f1cd260f52bc93ae3ac1c17f652f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lukas=20Bergstr=C3=B6m?= Date: Wed, 28 Aug 2024 16:55:24 +0200 Subject: [PATCH 5/6] [WIP] --- .../language/lexer_parser/mal_analyzer.py | 146 +++++++-- .../language/lexer_parser/mal_visitor.py | 17 +- .../tests/mal_analyzer_test_wrapper.py | 4 +- .../language/tests/test_mal_risk_type.py | 38 +++ maltoolbox/language/tests/test_mal_step.py | 279 ++++++++++++++++++ .../language/tests/test_mal_user_info.py | 30 ++ 6 files changed, 474 insertions(+), 40 deletions(-) create mode 100644 maltoolbox/language/tests/test_mal_risk_type.py create mode 100644 maltoolbox/language/tests/test_mal_step.py diff --git a/maltoolbox/language/lexer_parser/mal_analyzer.py b/maltoolbox/language/lexer_parser/mal_analyzer.py index bb266da9..d1a471cf 100644 --- a/maltoolbox/language/lexer_parser/mal_analyzer.py +++ b/maltoolbox/language/lexer_parser/mal_analyzer.py @@ -8,7 +8,7 @@ class malAnalyzerInterface: def checkMal(self, ctx: malParser.MalContext) -> None: pass - def checkDefine(self, ctx: malParser.DefineContext, data: Tuple[str, Any]) -> None: + def checkDefine(self, ctx: malParser.DefineContext, data: Tuple[str, dict]) -> None: pass def checkInclude(self, ctx: malParser.IncludeContext, data: Tuple[str, str]) -> None: pass @@ -24,6 +24,8 @@ 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): ''' @@ -57,6 +59,7 @@ def _post_analysis(self) -> None: self._analyse_extends() self._analyse_abstract() self._analyse_parents() + self._analyse_reaches() self._analyse_association() def _analyse_defines(self) -> None: @@ -65,8 +68,8 @@ def _analyse_defines(self) -> None: ''' if 'id' in self._defines.keys(): - define_value: str = self._defines['id']['obj']['id'] - if(len(define_value) == 0): + define_value: str = self._defines['id']['value'] + if (len(define_value) == 0): logging.error('Define \'id\' cannot be empty') self._error = True else: @@ -74,7 +77,7 @@ def _analyse_defines(self) -> None: self._error = True if 'version' in self._defines: - version: str = self._defines['version']['obj']['version'] + 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 @@ -135,7 +138,65 @@ def _analyse_parents(self) -> None: 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: + pass + + def _check_to_asset(self, asset, expr) -> None: + match (expr['type']): + case 'StepExpr': + raise + case 'IDExpr': + 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 'CallExpr': + raise + case _: + logging.error(f'Unexpected expression \'{expr["type"]}\'') + self._error = True + # exit(1) + return None + + def _check_set_expr(self, asset, expr) -> None: + pass + def _check_sub_type_expr(self, asset, expr) -> None: + pass + def _check_transitive_expr(self, asset, expr) -> None: + pass + def _analyse_association(self) -> None: for association in self._associations: leftAsset = association['leftAsset'] @@ -166,20 +227,17 @@ def checkMal(self, ctx: malParser.MalContext) -> None: 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, Any]) -> None: + def checkDefine(self, ctx: malParser.DefineContext, data: Tuple[str, dict]) -> None: _, obj = data - - if(len(obj.keys()) != 1): - raise + key, value = list(obj.items())[0] - define_id = next(iter(obj)) - if(define_id in self._defines.keys()): - prev_define_line = self._defines[define_id]['ctx'].start.line - logging.error(f'Define \'{define_id}\' previously defined at line {prev_define_line}') + 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[define_id] = {'ctx': ctx, 'obj': obj} + self._defines[key] = {'ctx': ctx, 'value': value} def checkCategory(self, ctx: malParser.CategoryContext, data: Tuple[str, Tuple[List, Any]]) -> None: _, [[category], assets] = data @@ -192,8 +250,6 @@ def checkCategory(self, ctx: malParser.CategoryContext, data: Tuple[str, Tuple[L if len(category['meta']) == 0 and len(assets) == 0: logging.warning(f'Category \'{category["name"]}\' contains no assets or metadata') - # Warning might not be checked as error. - # self._error = True self._category[category['name']] = {'ctx': ctx, 'obj': {'category': category, 'assets': assets}} @@ -216,8 +272,8 @@ def checkAsset(self, ctx: malParser.AssetContext, asset: dict) -> None: def checkMeta(self, ctx: malParser.MetaContext, data: Tuple[Tuple[str, str],]) -> None: ((meta_name, _),) = data - parent_name = '' - location_name = '' + parent_name: str = '' + location_name: str = '' # Finding metadata type if isinstance(ctx.parentCtx, malParser.CategoryContext): @@ -229,7 +285,10 @@ def checkMeta(self, ctx: malParser.MetaContext, data: Tuple[Tuple[str, str],]) - elif isinstance(ctx.parentCtx, malParser.StepContext): parent_name = str(ctx.parentCtx.ID()) location_name = 'step' - + elif isinstance(ctx.parentCtx, malParser.AssociationContext): + parent_name = str(ctx.parentCtx.ID()) + 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}} @@ -242,23 +301,52 @@ def checkMeta(self, ctx: malParser.MetaContext, data: Tuple[Tuple[str, str],]) - logging.error(f'Metadata {meta_name} previously defined at {prev_ctx.start.line}') self._error = True - # TODO: check for Associations - def checkStep(self, ctx: malParser.StepContext, step: dict) -> None: step_name = step['name'] - if isinstance(ctx.parentCtx, malParser.AssetContext): + if isinstance(ctx.parentCtx, malParser.AssetContext): asset_name = ctx.parentCtx.ID()[0] + # 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 - # TODO: Validate step - # if (not asset_name in self._steps.keys()): - # self._steps[asset_name] = {step_name: ctx} self._validate_CIA(ctx, step) self._validate_TTC(ctx, step) - return step - - def _validate_TTC(self, ctx: malParser.StepContext, step) -> None: + def checkReaches(self, ctx: malParser.ReachesContext, data: dict) -> None: + pass + + def _validate_TTC(self, ctx: malParser.StepContext, step: dict) -> None: if not step['ttc']: return @@ -272,7 +360,7 @@ def _validate_TTC(self, ctx: malParser.StepContext, step) -> None: # ERROR Defense %s.%s may only have 'Enabled', 'Disabled', or 'Bernoulli(p)' as TTC" pass - def _validate_CIA(self, ctx: malParser.StepContext, step) -> None: + def _validate_CIA(self, ctx: malParser.StepContext, step: dict) -> None: if not ctx.cias(): return diff --git a/maltoolbox/language/lexer_parser/mal_visitor.py b/maltoolbox/language/lexer_parser/mal_visitor.py index fbe97d81..27fecbb1 100644 --- a/maltoolbox/language/lexer_parser/mal_visitor.py +++ b/maltoolbox/language/lexer_parser/mal_visitor.py @@ -1,11 +1,10 @@ +from antlr4 import ParseTreeVisitor from .mal_parser import malParser from .mal_analyzer import malAnalyzerInterface -from antlr4 import ParseTreeVisitor from collections.abc import MutableMapping, MutableSequence - # In a rule like `rule: one? two* three`: # - ctx.one() would be None if the token was not found on a matching line # - ctx.two() would be [] @@ -16,6 +15,7 @@ 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): @@ -85,10 +85,9 @@ def visitMal(self, ctx): def visitInclude(self, ctx): return ("include", ctx.STRING().getText().strip('"')) - + def visitDefine(self, ctx): - define_object = {ctx.ID().getText(): ctx.STRING().getText().strip('"')} - return ("defines", define_object) + return ("defines", {ctx.ID().getText(): ctx.STRING().getText().strip('"')}) def visitCategory(self, ctx): category = {} @@ -96,6 +95,7 @@ def visitCategory(self, ctx): category["meta"] = {k: v for meta in ctx.meta() for k, v in self.visit(meta)} assets = [self.visit(asset) for asset in ctx.asset()] + return ("categories", ([category], assets)) def visitMeta(self, ctx): @@ -116,7 +116,7 @@ def visitAsset(self, ctx): asset["attackSteps"] = [self.visit(step) for step in ctx.step()] return asset - + def visitStep(self, ctx): step = {} step["name"] = ctx.ID().getText() @@ -129,7 +129,7 @@ def visitStep(self, ctx): self.visit(ctx.precondition()) if ctx.precondition() else None ) step["reaches"] = self.visit(ctx.reaches()) if ctx.reaches() else None - + return step def visitSteptype(self, ctx): @@ -265,6 +265,7 @@ def visitVariable(self, ctx): ret = {} ret["name"] = ctx.ID().getText() ret["stepExpression"] = self.visit(ctx.expr()) + return ret def visitExpr(self, ctx): @@ -278,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/mal_analyzer_test_wrapper.py b/maltoolbox/language/tests/mal_analyzer_test_wrapper.py index 3a69d604..0af3957b 100644 --- a/maltoolbox/language/tests/mal_analyzer_test_wrapper.py +++ b/maltoolbox/language/tests/mal_analyzer_test_wrapper.py @@ -45,7 +45,7 @@ def __init__(self, input_string: str) -> None: except RuntimeError: self._error = True - def test(self, error:bool=False, defines:list=[], categories:list=[], assets:list=[], lets:list=[], steps:list=[]): + 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())) @@ -58,8 +58,6 @@ def test(self, error:bool=False, defines:list=[], categories:list=[], assets:lis where, name = let if (not (self._vars[where] and self._vars[where][name])): assert(False) - if (steps): - assert(set(steps) == set(self._steps.keys())) 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..c9f59bac --- /dev/null +++ b/maltoolbox/language/tests/test_mal_step.py @@ -0,0 +1,279 @@ +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 + } + + asset Linux + { + | spyware + +> 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 + } + } + ''') + AnalyzerTestWrapper(''' + #id: "org.mal-lang.testAnalyzer" + #version:"0.0.0" + + include "./generated_test_mal.mal" + + category System { + asset Linux { + | spyware + +> 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 + } + } + + ''').test( + defines=['id', 'version'], + categories=['System'], + assets=['Linux'] + ) \ 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 index a58ae2c7..5b90c465 100644 --- a/maltoolbox/language/tests/test_mal_user_info.py +++ b/maltoolbox/language/tests/test_mal_user_info.py @@ -50,4 +50,34 @@ def test_user_info_2() -> None: 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 From b883e66e39a4f8cad261507dc3aa91617762e988 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lukas=20Bergstr=C3=B6m?= Date: Mon, 2 Sep 2024 14:25:05 +0200 Subject: [PATCH 6/6] [WIP] --- .../language/lexer_parser/mal_analyzer.py | 213 +++++++++++--- .../language/tests/test_mal_association.py | 56 ++++ maltoolbox/language/tests/test_mal_let.py | 33 +++ .../language/tests/test_mal_operations.py | 276 ++++++++++++++++++ .../test_mal_probability_distributions.py | 245 ++++++++++++++++ maltoolbox/language/tests/test_mal_step.py | 24 +- 6 files changed, 809 insertions(+), 38 deletions(-) create mode 100644 maltoolbox/language/tests/test_mal_operations.py create mode 100644 maltoolbox/language/tests/test_mal_probability_distributions.py diff --git a/maltoolbox/language/lexer_parser/mal_analyzer.py b/maltoolbox/language/lexer_parser/mal_analyzer.py index d1a471cf..55ab6bc9 100644 --- a/maltoolbox/language/lexer_parser/mal_analyzer.py +++ b/maltoolbox/language/lexer_parser/mal_analyzer.py @@ -168,13 +168,37 @@ def _analyse_reaches(self) -> None: raise '' def _check_to_step(self, asset, expr) -> None: - pass - + 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 'StepExpr': - raise - case 'IDExpr': + 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) @@ -182,20 +206,105 @@ def _check_to_asset(self, asset, expr) -> None: return self._check_transitive_expr(asset, expr) case 'subType': return self._check_sub_type_expr(asset, expr) - case 'CallExpr': - raise 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: - pass + 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: - pass + 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: - pass + 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: @@ -255,7 +364,7 @@ def checkCategory(self, ctx: malParser.CategoryContext, data: Tuple[str, Tuple[L def checkAsset(self, ctx: malParser.AssetContext, asset: dict) -> None: asset_name = asset['name'] - category_name = ctx.parentCtx.ID() + 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}") @@ -268,7 +377,7 @@ def checkAsset(self, ctx: malParser.AssetContext, asset: dict) -> None: self._error = True return else: - self._assets[asset_name] = {'ctx': ctx, 'obj': asset, 'parent': {'name': ctx.parentCtx.ID() ,'ctx': ctx.parentCtx}} + 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 @@ -277,16 +386,16 @@ def checkMeta(self, ctx: malParser.MetaContext, data: Tuple[Tuple[str, str],]) - # Finding metadata type if isinstance(ctx.parentCtx, malParser.CategoryContext): - parent_name = str(ctx.parentCtx.ID()) + parent_name = str(ctx.parentCtx.ID().getText()) location_name = 'category' elif isinstance(ctx.parentCtx, malParser.AssetContext): - parent_name = str(ctx.parentCtx.ID()[0]) + parent_name = str(ctx.parentCtx.ID()[0].getText()) location_name = 'asset' elif isinstance(ctx.parentCtx, malParser.StepContext): - parent_name = str(ctx.parentCtx.ID()) + parent_name = str(ctx.parentCtx.ID().getText()) location_name = 'step' elif isinstance(ctx.parentCtx, malParser.AssociationContext): - parent_name = str(ctx.parentCtx.ID()) + parent_name = str(ctx.parentCtx.ID()[0].getText()) location_name = 'association' # Validate that the metadata is unique @@ -305,7 +414,7 @@ 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] + 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): @@ -341,31 +450,66 @@ def checkStep(self, ctx: malParser.StepContext, step: dict) -> None: self._error = True self._validate_CIA(ctx, step) - self._validate_TTC(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, step: dict) -> None: + 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']) - if step['type'] == 'defense': - # TODO: TTCFuncExpr - # if !(ttc instanceof AST.TTCFuncExpr) - # error - # elif fname = Enabled, Disabled, Bernoulli - # Distributions.validate(fname, fparams); - # else - # ERROR Defense %s.%s may only have 'Enabled', 'Disabled', or 'Bernoulli(p)' as TTC" - pass + 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] + 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') @@ -401,13 +545,16 @@ def checkVariable(self, ctx: malParser.VariableContext, var: dict) -> None: 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} + 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 + self._vars[asset_name][var_name] = {'ctx': ctx, 'var': var} else: - prev_define_line = self._vars[asset_name][var_name].start.line + 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/tests/test_mal_association.py b/maltoolbox/language/tests/test_mal_association.py index c86b6eb8..186a72a4 100644 --- a/maltoolbox/language/tests/test_mal_association.py +++ b/maltoolbox/language/tests/test_mal_association.py @@ -80,4 +80,60 @@ def test_association_3() -> None: 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_let.py b/maltoolbox/language/tests/test_mal_let.py index 04f7f3cc..34781ab9 100644 --- a/maltoolbox/language/tests/test_mal_let.py +++ b/maltoolbox/language/tests/test_mal_let.py @@ -82,4 +82,37 @@ def test_let_3() -> None: 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_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_step.py b/maltoolbox/language/tests/test_mal_step.py index c9f59bac..72d0d14b 100644 --- a/maltoolbox/language/tests/test_mal_step.py +++ b/maltoolbox/language/tests/test_mal_step.py @@ -155,12 +155,14 @@ def test_step_6() -> None: { | spyware -> logKeystrokes + | logKeystrokes } asset Linux { | spyware +> readBashHistory + & readBashHistory } } @@ -234,6 +236,7 @@ def test_step_9() -> None: asset OperatingSystem { | spyware -> logKeystrokes + | logKeystrokes } } ''') @@ -247,6 +250,7 @@ def test_step_9() -> None: asset Linux { | spyware +> readBashHistory + & readBashHistory } } @@ -263,17 +267,27 @@ def test_step_10() -> None: AnalyzerTestWrapper(''' #id: "org.mal-lang.testAnalyzer" #version:"0.0.0" + - category System { + category System { asset Linux { - E hasCamera - <- hardware[Camera] - -> hijackCamera + E hasCamera + <- hardware[Camera] + -> hijackCamera + | hijackCamera + } + asset Camera { + | photo } } + + associations + { + Linux [linux] * <-- L --> * [hardware] Camera + } ''').test( defines=['id', 'version'], categories=['System'], - assets=['Linux'] + assets=['Linux', 'Camera'] ) \ No newline at end of file