Merge pull request #1116 from SEKOIA-IO/lv/add_sophos_edr_actions

Add Sophos EDR endpoint actions

Sophos/CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+## 2024-10-02 - 1.17.0
+
+### Added
+
+- Added Sophos EDR endpoint actions
+
 ## 2024-08-06 - 1.16.5
 
 ### Fixed

Sophos/action_sophos_edr_deisolate.json

@@ -0,0 +1,19 @@
+{
+  "arguments": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "properties": {
+      "endpoint_id": {
+        "description": "Endpoint ID",
+        "type": "string"
+      }
+    },
+    "required": ["endpoint_id"],
+    "title": "Arguments",
+    "type": "object"
+  },
+  "description": "Turn off endpoint isolation",
+  "docker_parameters": "sophos_edr_deisolate_endpoint",
+  "name": "[BETA] Deisolate endpoint",
+  "results": {},
+  "uuid": "258030ab-8275-4056-877e-105cb74de49d"
+}

Sophos/action_sophos_edr_isolate.json

@@ -0,0 +1,19 @@
+{
+  "arguments": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "properties": {
+      "endpoint_id": {
+        "description": "Endpoint ID",
+        "type": "string"
+      }
+    },
+    "required": ["endpoint_id"],
+    "title": "Arguments",
+    "type": "object"
+  },
+  "description": "Turn on endpoint isolation",
+  "docker_parameters": "sophos_edr_isolate_endpoint",
+  "name": "[BETA] Isolate endpoint",
+  "results": {},
+  "uuid": "91d24356-6537-4a3e-ba6f-19c7963db8fe"
+}

Sophos/action_sophos_edr_run_scan.json

@@ -0,0 +1,19 @@
+{
+  "arguments": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "properties": {
+      "endpoint_id": {
+        "description": "Endpoint ID",
+        "type": "string"
+      }
+    },
+    "required": ["endpoint_id"],
+    "title": "Arguments",
+    "type": "object"
+  },
+  "description": "",
+  "docker_parameters": "sophos_edr_run_scan",
+  "name": "[BETA] Run scan",
+  "results": {},
+  "uuid": "bc98176f-10d5-46e8-9a43-8694e8ea4e68"
+}

Sophos/main.py

@@ -1,3 +1,6 @@
+from sophos_module.action_sophos_edr_deisolate import ActionSophosEDRDeIsolateEndpoint
+from sophos_module.action_sophos_edr_isolate import ActionSophosEDRIsolateEndpoint
+from sophos_module.action_sophos_edr_run_scan import ActionSophosEDRScan
 from sophos_module.base import SophosModule
 from sophos_module.trigger_sophos_edr_events import SophosEDREventsTrigger
 from sophos_module.trigger_sophos_xdr_query import SophosXDRIOCQuery
@@ -6,4 +9,8 @@
     module = SophosModule()
     module.register(SophosEDREventsTrigger, "sophos_events_trigger")
     module.register(SophosXDRIOCQuery, "sophos_query_ioc_trigger")
+    module.register(ActionSophosEDRIsolateEndpoint, "sophos_edr_isolate_endpoint")
+    module.register(ActionSophosEDRDeIsolateEndpoint, "sophos_edr_deisolate_endpoint")
+    module.register(ActionSophosEDRScan, "sophos_edr_run_scan")
+
     module.run()

Sophos/manifest.json

@@ -38,7 +38,7 @@
   "name": "Sophos",
   "uuid": "0de5216e-19b0-4ad3-9b91-a547cfaf52ca",
   "slug": "sophos",
-  "version": "1.16.5",
+  "version": "1.17.0",
   "categories": [
     "Endpoint"
   ]

Sophos/sophos_module/action_base.py

@@ -0,0 +1,73 @@
+from abc import ABC
+from functools import cached_property
+from typing import Any, Callable
+from urllib.parse import urljoin
+
+from sekoia_automation.action import Action
+
+from .base import SophosModule
+from .client import SophosApiClient
+from .client.auth import SophosApiAuthentication
+from .logging import get_logger
+
+logger = get_logger()
+
+
+class SophosEDRAction(Action, ABC):
+    module: SophosModule
+
+    @cached_property
+    def client(self) -> SophosApiClient:
+        auth = SophosApiAuthentication(
+            api_host=self.module.configuration.api_host,
+            authorization_url=self.module.configuration.oauth2_authorization_url,
+            client_id=self.module.configuration.client_id,
+            client_secret=self.module.configuration.client_secret,
+        )
+        return SophosApiClient(auth=auth)
+
+    @cached_property
+    def region_base_url(self) -> str:
+        url = urljoin(self.module.configuration.api_host, "whoami/v1")
+        response = self.client.get(url).json()
+
+        return str(response["apiHosts"]["dataRegion"])
+
+    def call_endpoint(
+        self, method: str, url: str, data: dict[str, Any] | None = None, use_region_url: bool = False
+    ) -> Any:
+        assert method.lower() in ("get", "post", "patch")
+
+        base_url = self.region_base_url if use_region_url else self.module.configuration.api_host
+        url = urljoin(base_url, url)
+
+        func: Callable[[Any], Any]
+        if method.lower() == "post":
+            func = self.client.post
+
+        elif method.lower() == "patch":
+            func = self.client.patch
+
+        else:
+            func = self.client.get
+
+        response = func(url=url, json=data)
+
+        if response.ok:
+            return response.json()
+
+        else:
+            raw = response.json()
+            logger.error(
+                f"Error {response.status_code}",
+                error=raw.get("error"),
+                message=raw.get("message"),
+                corellation_id=raw.get("correlationId"),
+                code=raw.get("code"),
+                created_at=raw.get("createdAt"),
+                request_id=raw.get("requestId"),
+                doc_url=raw.get("docUrl"),
+            )
+            response.raise_for_status()
+
+        return {}

Sophos/sophos_module/action_sophos_edr_deisolate.py

@@ -0,0 +1,11 @@
+from typing import Any
+
+from sophos_module.action_sophos_edr_isolate import ActionSophosEDRIsolateEndpoint
+
+
+class ActionSophosEDRDeIsolateEndpoint(ActionSophosEDRIsolateEndpoint):
+    def run(self, arguments: dict[str, Any]) -> Any:
+        endpoint_id = arguments["endpoint_id"]
+        comment = arguments.get("comment")
+
+        return self.set_isolation_status(endpoint_id=endpoint_id, enabled=False, comment=comment)

Sophos/sophos_module/action_sophos_edr_isolate.py

@@ -0,0 +1,61 @@
+from typing import Any
+
+import requests
+
+from sophos_module.action_base import SophosEDRAction
+
+
+class ActionSophosEDRIsolateEndpoint(SophosEDRAction):
+    def get_endpoint_isolation_status(self, endpoint_id: str) -> Any:
+        return self.call_endpoint(
+            method="get", url=f"endpoint/v1/endpoints/{endpoint_id}/isolation", use_region_url=True
+        )
+
+    def try_to_set_isolation_state(self, endpoint_id: str, enabled: bool, comment: str | None = None) -> Any:
+        data: dict[str, Any] = {"enabled": enabled}
+        if comment is not None:
+            data["comment"] = comment
+
+        return self.call_endpoint(
+            method="patch", url=f"endpoint/v1/endpoints/{endpoint_id}/isolation", data=data, use_region_url=True
+        )
+
+    def set_isolation_status(self, endpoint_id: str, enabled: bool, comment: str | None = None) -> Any:
+        """
+        Sophos Endpoint API will return `Bad Request` every time you will
+        try to enable/disable already enabled/disabled isolation on an endpoint,
+        or whenever there's too little time passed between status changes. So we
+        have to make double checks and enrich returned messages
+        """
+        # check whether endpoint is already in the desired state
+        current_status = self.get_endpoint_isolation_status(endpoint_id)
+        if current_status["enabled"] == enabled:
+            result_label = "enabled" if enabled else "disabled"
+            current_status["message"] = "Already %s" % result_label
+            return current_status
+
+        try:
+            response = self.try_to_set_isolation_state(endpoint_id=endpoint_id, enabled=enabled, comment=comment)
+            return response
+
+        except requests.exceptions.HTTPError as err:
+            if err.response.status_code == 400:
+                # returned as "bad request" in two cases:
+                # 1. endpoint is already in the desired state - we checked for that before
+                # 2. too little time passed since an isolation state was changed - need to wait
+                result = err.response.json()
+                result.update(
+                    {
+                        "error": "Isolation status was changed recently",
+                        "message": "We recommend that you wait for a period of time "
+                        "between turning endpoint isolation on and off",
+                    }
+                )
+                result.update(current_status)
+                return result
+
+    def run(self, arguments: dict[str, Any]) -> Any:
+        endpoint_id = arguments["endpoint_id"]
+        comment = arguments.get("comment")
+
+        return self.set_isolation_status(endpoint_id=endpoint_id, enabled=True, comment=comment)

Sophos/sophos_module/action_sophos_edr_run_scan.py

@@ -0,0 +1,12 @@
+from typing import Any
+
+from sophos_module.action_base import SophosEDRAction
+
+
+class ActionSophosEDRScan(SophosEDRAction):
+    def run(self, arguments: dict[str, Any]) -> Any:
+        endpoint_id = arguments["endpoint_id"]
+
+        return self.call_endpoint(
+            method="post", url=f"endpoint/v1/endpoints/{endpoint_id}/scans", data={}, use_region_url=True
+        )