Added constraint for the import of persistence from the API

Author: c8y3

pyproject.toml

@@ -7,6 +7,13 @@ ignore = ["E402", "E711", "E712", "E721", "E722"]
 root_package = "app"
 include_external_packages = true
 
+[[tool.importlinter.contracts]]
+name = "Do not import the persistence layer from the API layer"
+type = "forbidden"
+source_modules = "app.blueprints.rest.v2.alerts"
+forbidden_modules = "app.datamgmt.alerts"
+allow_indirect_imports = true
+
 [[tool.importlinter.contracts]]
 name = "Do not import sqlalchemy from the API layer"
 type = "forbidden"

source/app/blueprints/rest/v2/alerts.py

@@ -31,7 +31,7 @@
 from app.blueprints.rest.parsing import parse_comma_separated_identifiers
 from app.blueprints.rest.v2.alerts_routes.comments import alerts_comments_blueprint
 from app.iris_engine.access_control.iris_user import iris_current_user
-from app.datamgmt.alerts.alerts_db import get_filtered_alerts
+from app.business.alerts import alerts_search
 from app.models.authorization import Permissions
 from app.schema.marshables import AlertSchema
 from app.schema.marshables import IocSchema
@@ -91,31 +91,31 @@ def search(self):
         else:
             fields = None
 
-        filtered_alerts = get_filtered_alerts(
-            start_date=request.args.get('creation_start_date'),
-            end_date=request.args.get('creation_end_date'),
-            source_start_date=request.args.get('source_start_date'),
-            source_end_date=request.args.get('source_end_date'),
-            source_reference=request.args.get('source_reference'),
-            title=request.args.get('alert_title'),
-            description=request.args.get('alert_description'),
-            status=request.args.get('alert_status_id', type=int),
-            severity=request.args.get('alert_severity_id', type=int),
-            owner=request.args.get('alert_owner_id', type=int),
-            source=request.args.get('alert_source'),
-            tags=request.args.get('alert_tags'),
-            classification=request.args.get('alert_classification_id', type=int),
-            client=request.args.get('alert_customer_id'),
-            case_id=request.args.get('case_id', type=int),
-            alert_ids=alert_ids,
-            page=page,
-            per_page=per_page,
-            sort=request.args.get('sort'),
-            custom_conditions=request.args.get('custom_conditions'),
-            assets=alert_assets,
-            iocs=alert_iocs,
-            resolution_status=request.args.get('alert_resolution_id', type=int),
-            current_user_id=iris_current_user.id
+        filtered_alerts = alerts_search(
+            iris_current_user.id,
+            request.args.get('creation_start_date'),
+            request.args.get('creation_end_date'),
+            request.args.get('source_start_date'),
+            request.args.get('source_end_date'),
+            request.args.get('alert_title'),
+            request.args.get('alert_description'),
+            request.args.get('alert_status_id', type=int),
+            request.args.get('alert_severity_id', type=int),
+            request.args.get('alert_owner_id', type=int),
+            request.args.get('alert_source'),
+            request.args.get('alert_tags'),
+            request.args.get('case_id', type=int),
+            request.args.get('alert_customer_id'),
+            request.args.get('alert_classification_id', type=int),
+            alert_ids,
+            alert_assets,
+            alert_iocs,
+            request.args.get('alert_resolution_id', type=int),
+            request.args.get('source_reference'),
+            request.args.get('custom_conditions'),
+            page,
+            per_page,
+            request.args.get('sort')
         )
 
         if filtered_alerts is None:

source/app/business/alerts.py

@@ -30,13 +30,45 @@
 from app.datamgmt.alerts.alerts_db import delete_related_alerts_cache
 from app.datamgmt.alerts.alerts_db import get_alert_by_id
 from app.datamgmt.alerts.alerts_db import delete_alert
+from app.datamgmt.alerts.alerts_db import get_filtered_alerts
 from app.iris_engine.module_handler.module_handler import call_modules_hook
 from app.iris_engine.utils.tracker import track_activity
 from app.util import add_obj_history_entry
 from app.business.errors import ObjectNotFoundError
 from app.datamgmt.manage.manage_access_control_db import user_has_client_access
 
 
+def alerts_search(user_identifier, start_date, end_date, source_start_date, source_end_date, title, description,
+                  status, severity, owner, source, tags, case_identifier, customer, classification, alert_identifiers,
+                  assets, iocs, resolution_status, source_reference, custom_conditions, page, per_page, sort):
+    return get_filtered_alerts(
+        start_date=start_date,
+        end_date=end_date,
+        source_start_date=source_start_date,
+        source_end_date=source_end_date,
+        title=title,
+        description=description,
+        status=status,
+        severity=severity,
+        owner=owner,
+        source=source,
+        tags=tags,
+        case_id=case_identifier,
+        client=customer,
+        classification=classification,
+        alert_ids=alert_identifiers,
+        assets=assets,
+        iocs=iocs,
+        resolution_status=resolution_status,
+        page=page,
+        per_page=per_page,
+        sort=sort,
+        current_user_id=user_identifier,
+        source_reference=source_reference,
+        custom_conditions=custom_conditions
+    )
+
+
 def alerts_create(alert: Alert, iocs: list[Ioc], assets: list[CaseAssets]) -> Alert:
 
     alert.alert_creation_time = datetime.utcnow()