Added deboard command

mantis/db/crud_assets.py

@@ -33,3 +33,19 @@ async def update_asset_query(asset: str, org: str, mongodb_query):
         return False
     else:
         logging.error(f'Asset {asset} does not exists in DB, Update failed')
+
+async def delete_assets_query(query: dict) -> bool:
+    try:
+        if not query:
+            return False 
+
+        logging.debug(f"Executing delete_assets_query: {query}")
+        result = await assets_collection.delete_many(query)
+        logging.info(f"Deleted {result.deleted_count} asset{'s' if result.deleted_count != 1 else ''} from the database")
+
+        return result.deleted_count > 0
+
+    except Exception as e:
+        logging.error(f"Error deleting assets: {e}")
+        return False
+

mantis/db/crud_extended_assets.py

@@ -52,3 +52,19 @@ async def update_extended_asset_query(asset: str, org: str, mongodb_query):
         return False
     else:
         logging.error(f'Asset {asset} does not exists in DB, Update failed')
+
+async def delete_extended_assets_query(query: dict) -> bool:
+    try:
+        if not query:
+            return False  
+
+        logging.debug(f"Executing delete_extended_assets_query: {query}")
+        result = await extended_assets_collection.delete_many(query)
+        logging.info(f"Deleted {result.deleted_count} extended asset{'s' if result.deleted_count != 1 else ''} from the database")
+
+        return result.deleted_count > 0
+
+    except Exception as e:
+        logging.error(f"Error deleting extended assets: {e}")
+        return False
+

mantis/db/crud_vulnerabilities.py

@@ -46,4 +46,20 @@ async def check_field_exists(field_name: str, value: str) -> bool:
 
     except Exception as e:
         logging.debug(f"Error checking field {field_name} in the database: {e}")
-        return False
+        return False
+
+async def delete_findings_query(query: dict) -> bool:
+    try:
+        if not query:
+            return False 
+
+        logging.debug(f"Executing delete_findings_query: {query}")
+        result = await findings_collection.delete_many(query)
+        logging.info(f"Deleted {result.deleted_count} finding{'s' if result.deleted_count != 1 else ''} from the database")
+
+        return result.deleted_count > 0
+
+    except Exception as e:
+        logging.error(f"Error deleting findings: {e}")
+        return False
+

mantis/models/args_model.py

@@ -28,4 +28,5 @@ class ArgsModel(BaseModel):
     after_datetime_filter: str = None
     before_datetime_filter: str = None
     in_scope: bool = False
-
+    deboard_: bool = False
+    collections: list[str] = False

mantis/utils/args_parse.py

@@ -59,6 +59,18 @@ def report_msg(name=None):
         \033[0;32mmantis report -o example_org\033[0m
             '''
 
+    @staticmethod
+    def deboard_msg(name=None):                                                            
+        return '''
+        \033[1;34mDEBOARD:\033[0m
+
+        \033[0;32mmantis deboard -o example_org\033[0m
+        \033[0;32mmantis deboard -o example_org -s example_domain\033[0m
+        \033[0;32mmantis deboard -o example_org -a\033[0m
+        \033[0;32mmantis deboard -o example_org -a -f\033[0m
+        \033[0;32mmantis deboard -o example_org -a -s example_domain\033[0m
+            '''
+
     @staticmethod
     def args_parse() -> ArgsModel:
         parsed_args = {}
@@ -266,6 +278,14 @@ def args_parse() -> ArgsModel:
                             required = True,
                             help = "name of the organisation")
 
+        deboard_parser = subparser.add_parser("deboard", help="Deboard a target", usage=ArgsParse.deboard_msg())
+
+        deboard_parser.add_argument('-o', '--org', dest='org', required=True, help="name of the organisation to deboard")
+        deboard_parser.add_argument('-s', '--sub', dest='subdomain', help="subdomain to deboard")
+        deboard_parser.add_argument('-a', '--assets', dest='assets', action='store_true', help="deboard all matching assets")
+        deboard_parser.add_argument('-f', '--findings', dest='findings', action='store_true', help="deboard all matching findings")
+        deboard_parser.add_argument('-e', '--extended-assets', dest='extended_assets', action='store_true', help="deboard all matching extended assets")
+
         # display help, if no arguments are passed
         args = parser.parse_args(args=None if argv[1:] else ['--help'])
         logging.info(f"Arguments Passed - {args}")
@@ -282,7 +302,7 @@ def args_parse() -> ArgsModel:
                 parsed_args['input_type'] = "file"
                 parsed_args['input'] = str(args.file_name)
 
-        if args.subcommand != "list" and args.subcommand != "report":
+        if args.subcommand != "list" and args.subcommand != "report" and args.subcommand != "deboard":
 
             if args.aws_profiles:
                 parsed_args["aws_profiles"] = args.aws_profiles.split(',')
@@ -359,6 +379,31 @@ def args_parse() -> ArgsModel:
 
                 if args.list_sub_command_ls_subs_before_filter:
                     parsed_args["before_datetime_filter"] = f"{args.list_sub_command_ls_subs_before_filter}T23:59:59Z"
+
+        if args.subcommand == "deboard":
+            parsed_args["deboard_"] = True
+            parsed_args['org'] = args.org
+            parsed_args["collections"] = []
+
+            if (args.subdomain):
+                parsed_args['subdomain'] = args.subdomain
+
+            collection_mapping = {
+                "assets": "assets",
+                "findings": "findings",
+                "extended_assets": "extended_assets",
+            }
+
+            for arg, collection in collection_mapping.items():
+                if getattr(args, arg, False):
+                    parsed_args["collections"].append(collection) 
+
+            if not parsed_args["collections"]:
+                # default to all collections
+                parsed_args["collections"] = ["assets", "findings", "extended_assets"]
+
+            parsed_args["collections"] = list(set(parsed_args["collections"]))
+
 
         args_pydantic_obj = ArgsModel.parse_obj(parsed_args)
         logging.info(f'parsed args - {args_pydantic_obj}')

mantis/utils/crud_utils.py

@@ -1,9 +1,9 @@
 import hashlib
 import logging
 from mantis.db.db_models import Assets, Findings, Extended
-from mantis.db.crud_assets import add_assets_query, update_asset_query
-from mantis.db.crud_extended_assets import add_extended_assets_query
-from mantis.db.crud_vulnerabilities import add_findings_query, findings_bulk_mixed_query
+from mantis.db.crud_assets import add_assets_query, delete_assets_query, update_asset_query
+from mantis.db.crud_extended_assets import add_extended_assets_query, delete_extended_assets_query
+from mantis.db.crud_vulnerabilities import add_findings_query, delete_findings_query, findings_bulk_mixed_query
 from mantis.utils.common_utils import CommonUtils
 from mantis.config_parsers.config_client import ConfigProvider
 from mantis.constants import ASSET_TYPE_SUBDOMAIN
@@ -241,3 +241,4 @@ def assign_app_context(domain):
                 if value in domain:
                     return key
         return default[0]
+

mantis/utils/deboard_utils.py

@@ -0,0 +1,45 @@
+import logging
+from mantis.db.crud_assets import delete_assets_query
+from mantis.db.crud_extended_assets import delete_extended_assets_query
+from mantis.db.crud_vulnerabilities import delete_findings_query
+
+async def deboard_organisation(args):
+    try:
+        org = args.org
+        subdomain = args.subdomain
+        collections = args.collections or ["assets", "findings", "extended_assets"]
+
+        logging.info(f"Target collections for deboard process: {', '.join(collections)}")
+
+        base_query = {"org": org}
+        queries = {
+            "assets": base_query.copy(),
+            "findings": base_query.copy(),
+            "extended_assets": base_query.copy(),
+        }
+
+        if subdomain:
+            queries["assets"]["asset"] = subdomain
+            queries["findings"]["url"] = subdomain
+            queries["extended_assets"]["asset"] = subdomain
+
+        delete_methods = {
+            "assets": delete_assets_query,
+            "findings": delete_findings_query,
+            "extended_assets": delete_extended_assets_query,
+        }
+
+        deletion_results = {}
+
+        for collection in collections:
+            if collection in delete_methods:
+                deletion_results[collection] = await delete_methods[collection](queries[collection])
+
+        if not any(deletion_results.values()):
+            return False
+
+        return True
+
+    except Exception as e:
+        logging.error(f"Error deboarding organisation {org}: {e}")
+        return False

mantis/workflows/deboard_workflow.py

@@ -0,0 +1,22 @@
+import logging
+from mantis.models.args_model import ArgsModel
+from mantis.utils.deboard_utils import deboard_organisation
+
+class DeboardWorkflow:
+    @staticmethod
+    async def executor(args: ArgsModel):
+        """
+        Executes the deboarding process for a given organisation based on provided arguments.
+        Deletes data from assets, findings, and extended assets collections accordingly.
+        """
+        if args.org:
+            logging.info(f"Starting deboard process for organisation {args.org}")
+
+            deboard = await deboard_organisation(args)
+
+            if (deboard):
+                logging.info(f"Successfully completed deboard process for organisation {args.org}")
+                print(f"\n\033[1;32mAll specified data for {args.org} has been successfully deleted from the database\033[0m\n")
+            else:
+                logging.info(f"Deboard process completed for organisation {args.org}, but no matching records were found")
+                print(f"\n\033[1;32mNo matching records found for {args.org}, Nothing was deleted\033[0m\n")

mantis/workflows/mantis_workflow.py

@@ -2,6 +2,7 @@
 from mantis.modules.workflow import Workflow
 from mantis.workflows.list_workflow import ListWorkflow
 from mantis.workflows.report_workflow import ReportWorkflow
+from mantis.workflows.deboard_workflow import DeboardWorkflow
 import asyncio
 
 class MantisWorkflow:
@@ -12,6 +13,8 @@ def select_workflow(args: ArgsModel) -> None:
             asyncio.run(ListWorkflow.executor(args))
         elif args.report_:
             asyncio.run(ReportWorkflow.executor())
+        elif args.deboard_:
+            asyncio.run(DeboardWorkflow.executor(args))
         else:
             asyncio.run(Workflow.workflow_executor(args))