Domain Scanner

artemis/module_base.py

@@ -20,7 +20,7 @@
 from artemis.db import DB
 from artemis.domains import is_domain
 from artemis.redis_cache import RedisCache
-from artemis.resolvers import lookup
+from artemis.resolvers import NoAnswer, ResolutionException, lookup
 from artemis.resource_lock import FailedToAcquireLockException, ResourceLock
 from artemis.retrying_resolver import setup_retrying_resolver
 from artemis.task_utils import (
@@ -123,6 +123,52 @@ def add_task(self, current_task: Task, new_task: Task) -> None:
         else:
             self.log.info("Task is not a new task, not adding: %s", new_task)
 
+    def add_task_if_domain_exists(self, current_task: Task, new_task: Task) -> None:
+        """
+        Add a new task if the domain in the task payload exists.
+
+        Args:
+            current_task (Task): The current task being processed.
+            new_task (Task): The new task to potentially add.
+        """
+        domain = new_task.payload.get("domain")
+        if not domain:
+            self.log.info("No domain found in new task payload - adding it, as it might be an IP task")
+            self.add_task(current_task, new_task)
+            return
+
+        if self.check_domain_exists(domain):
+            self.add_task(current_task, new_task)
+        else:
+            self.log.info("Skipping invalid domain: %s", domain)
+
+    def check_domain_exists(self, domain: str) -> bool:
+        """
+        Check if a domain exists by looking up its NS and A records.
+
+        Args:
+            domain (str): The domain to check.
+
+        Returns:
+            bool: True if the domain exists, False otherwise.
+        """
+        try:
+            # Check for NS records
+            try:
+                ns_records = lookup(domain, "NS")
+                if ns_records:
+                    return True
+            except NoAnswer:
+                # No NS records, continue to check A records
+                pass
+
+            # Check for A records
+            a_records = lookup(domain, "A")
+            return len(a_records) > 0  # returns true if found
+
+        except ResolutionException:
+            return False
+
     def loop(self) -> None:
         """
         Differs from the original karton implementation: consumes the tasks in random order, so that

artemis/modules/classifier.py

@@ -241,7 +241,7 @@ def run(self, current_task: Task) -> None:
                 },
             )
 
-            self.add_task(current_task, new_task)
+            self.add_task_if_domain_exists(current_task, new_task)
 
 
 if __name__ == "__main__":

artemis/modules/subdomain_enumeration.py

@@ -178,13 +178,14 @@ def run(self, current_task: Task) -> None:
 
             # We save the task as soon as we have results from a single tool so that other kartons can do something.
             for subdomain in valid_subdomains_from_tool:
-                task = Task(
-                    {"type": TaskType.DOMAIN},
-                    payload={
-                        "domain": subdomain,
-                    },
-                )
-                self.add_task(current_task, task)
+                if subdomain != domain:  # ensure we are not adding the parent domain again
+                    task = Task(
+                        {"type": TaskType.DOMAIN},
+                        payload={
+                            "domain": subdomain,
+                        },
+                    )
+                    self.add_task_if_domain_exists(current_task, task)
 
             valid_subdomains.update(valid_subdomains_from_tool)
 

docker-compose.yaml

@@ -99,7 +99,6 @@ services:
     restart: always
     volumes: ["./docker/karton.ini:/etc/karton/karton.ini", "${DOCKER_COMPOSE_ADDITIONAL_SHARED_DIRECTORY:-./shared}:/shared/"]
 
-
   karton-dashboard:
     depends_on: [karton-system, karton-logger]
     env_file: .env
@@ -163,7 +162,6 @@ services:
     restart: always
     volumes: ["./docker/karton.ini:/etc/karton/karton.ini", "${DOCKER_COMPOSE_ADDITIONAL_SHARED_DIRECTORY:-./shared}:/shared/"]
 
-
   karton-http_service_to_url:
     <<: *artemis-build-or-image
     command: "python3 -m artemis.modules.http_service_to_url"

test/base.py

@@ -2,7 +2,7 @@
 import socket
 import tempfile
 from pathlib import Path
-from typing import List
+from typing import Any, List, Set
 from unittest.mock import MagicMock, patch
 
 from jinja2 import BaseLoader, Environment, StrictUndefined, Template
@@ -40,10 +40,14 @@ def setUp(self) -> None:
         # Unfortunately, in the context of a test that is about to run and a respective module has already been
         # imported, to mock lookup we need to mock it in modules it has been imported to,
         # so we need to enumerate the locations it's used in in the list below.
+        def mock_lookup(host: str, *args: Any) -> Set[str]:
+            try:
+                return {socket.gethostbyname(host)}
+            except socket.gaierror:
+                return set()  # Return an empty set if resolution fails
+
         for item in ["artemis.module_base.lookup", "artemis.modules.port_scanner.lookup"]:
-            # We cannot use Artemis default DoH resolvers as they wouldn't be able to resolve
-            # internal test services' addresses.
-            self._lookup_mock = patch(item, MagicMock(side_effect=lambda host: {socket.gethostbyname(host)}))
+            self._lookup_mock = patch(item, MagicMock(side_effect=mock_lookup))
             self._lookup_mock.__enter__()
 
         self.mock_db = MagicMock()