Device identifier module

artemis/binds.py

@@ -22,6 +22,9 @@ class TaskType(str, Enum):
     # {URL: just a HTTP URL. Must have content attached to make further operations faster}
     URL = "url"
 
+    # {device: having an ip, port, ssl, and a device type, e.g. Device.FORTIOS}
+    DEVICE = "device"
+
 
 class Service(str, Enum):
     # Each of the services can have the SSL flag enabled - therefore HTTP covers both HTTP and HTTPS.
@@ -58,6 +61,12 @@ def _missing_(cls, value: object) -> WebApplication:
         return WebApplication.UNKNOWN
 
 
+class Device(str, Enum):
+    UNKNOWN = "unknown"
+
+    FORTIOS = "fortios"
+
+
 class TaskStatus(str, Enum):
     OK = "OK"
     ERROR = "ERROR"

artemis/http_requests.py

@@ -26,6 +26,7 @@ class HTTPResponse:
     encoding: str
     is_redirect: bool
     url: str
+    headers: Dict[str, str]
 
     def json(self) -> Any:
         return json.loads(self.content)
@@ -79,6 +80,7 @@ def _internal_request() -> HTTPResponse:
             encoding=response.encoding if response.encoding else "utf-8",
             is_redirect=bool(response.history),
             url=response.url,
+            headers=response.headers,
         )
 
     return throttle_request(_internal_request)  # type: ignore

artemis/modules/device_identifier.py

@@ -0,0 +1,53 @@
+from karton.core import Task
+
+from artemis import http_requests
+from artemis.binds import Device, Service, TaskStatus, TaskType
+from artemis.module_base import ArtemisBase
+from artemis.task_utils import get_target_host, get_target_url
+
+
+class DeviceIdentifier(ArtemisBase):
+    """
+    Tries to identify the device (FortiOS, ...) and produces a SERVICE task with proper type (e.g. Device.FORTIOS)
+    """
+
+    identity = "device_identifier"
+
+    filters = [
+        {"type": TaskType.SERVICE.value, "service": Service.HTTP.value},
+    ]
+
+    @staticmethod
+    def _identify(url: str) -> Device:
+        response = http_requests.get(url, allow_redirects=True)
+        if "xxxxxxxx-xxxxx" in response.headers.get("Server", "") and "/remote/login" in response.url:
+            return Device.FORTIOS
+
+        return Device.UNKNOWN
+
+    def _process(self, current_task: Task, url: str, host: str, port: int) -> None:
+        device = self._identify(url)
+
+        new_task = Task(
+            {
+                "type": TaskType.DEVICE,
+                "device": device,
+            },
+            payload={"host": host, "port": int(port), "ssl": current_task.get_payload("ssl")},
+        )
+        self.add_task(current_task, new_task)
+
+        self.db.save_task_result(task=current_task, status=TaskStatus.OK, data=device)
+
+    def run(self, current_task: Task) -> None:
+        url = get_target_url(current_task)
+        host = get_target_host(current_task)
+        port = current_task.get_payload("port")
+
+        self.log.info(f"device identifier scanning {url}")
+
+        self._process(current_task, url, host, port)
+
+
+if __name__ == "__main__":
+    DeviceIdentifier().loop()

docker-compose.yaml

@@ -416,6 +416,17 @@ services:
     restart: always
     volumes: ["./docker/karton.ini:/etc/karton/karton.ini", "${DOCKER_COMPOSE_ADDITIONAL_SHARED_DIRECTORY:-./shared}:/shared/"]
 
+  karton-device_identifier:
+    build:
+      context: .
+      dockerfile: docker/Dockerfile
+    command: "python3 -m artemis.modules.device_identifier"
+    depends_on: [karton-system]
+    env_file: .env
+    restart: always
+    volumes: ["./docker/karton.ini:/etc/karton/karton.ini", "${DOCKER_COMPOSE_ADDITIONAL_SHARED_DIRECTORY:-./shared}:/shared/"]
+
+
 volumes:
   data-nuclei-templates:
   data-nuclei-config: