Merge feature/add-logging-to-reports into feature/add-logging-to-reports

.env-dist

@@ -65,9 +65,6 @@ BYTES_DB_URI=postgresql://${BYTES_DB_USER}:${BYTES_DB_PASSWORD}@postgres:5432/${
 # --- Octopoes --- #
 # See `octopoes/octopoes/config/settings.py`
 
-# Number of Celery workers (for the Octopoes API worker) that need to be started
-CELERY_WORKER_CONCURRENCY=${CELERY_WORKER_CONCURRENCY:-4}
-
 # --- Mula --- #
 # See `mula/scheduler/config/settings.py`
 

boefjes/boefjes/dependencies/plugins.py

@@ -17,11 +17,11 @@
 from boefjes.storage.interfaces import (
     ConfigStorage,
     DuplicatePlugin,
-    IntegrityError,
     NotFound,
     PluginNotFound,
     PluginStorage,
     SettingsNotConformingToSchema,
+    UniqueViolation,
 )
 
 logger = structlog.get_logger(__name__)
@@ -49,9 +49,9 @@ def get_all(self, organisation_id: str) -> list[PluginType]:
         return [self._set_plugin_enabled(plugin, organisation_id) for plugin in all_plugins.values()]
 
     def _get_all_without_enabled(self) -> dict[str, PluginType]:
-        all_plugins = {plugin.id: plugin for plugin in self.local_repo.get_all()}
+        all_plugins = {plugin.id: plugin for plugin in self.plugin_storage.get_all()}
 
-        for plugin in self.plugin_storage.get_all():
+        for plugin in self.local_repo.get_all():  # Local plugins take precedence
             all_plugins[plugin.id] = plugin
 
         return all_plugins
@@ -94,7 +94,7 @@ def clone_settings_to_organisation(self, from_organisation: str, to_organisation
             self.set_enabled_by_id(plugin_id, to_organisation, enabled=True)
 
     def upsert_settings(self, settings: dict, organisation_id: str, plugin_id: str):
-        self._assert_settings_match_schema(settings, plugin_id)
+        self._assert_settings_match_schema(settings, plugin_id, organisation_id)
         self._put_boefje(plugin_id)
 
         return self.config_storage.upsert(organisation_id, plugin_id, settings=settings)
@@ -113,29 +113,25 @@ def create_boefje(self, boefje: Boefje) -> None:
                     try:
                         with self.plugin_storage as storage:
                             storage.create_boefje(boefje)
-                    except IntegrityError as error:
-                        raise DuplicatePlugin(self._translate_duplicate_plugin(error.message))
+                    except UniqueViolation as error:
+                        raise DuplicatePlugin(error.field)
             except KeyError:
                 try:
                     with self.plugin_storage as storage:
                         storage.create_boefje(boefje)
-                except IntegrityError as error:
-                    raise DuplicatePlugin(self._translate_duplicate_plugin(error.message))
-
-    def _translate_duplicate_plugin(self, error_message):
-        translations = {"boefje_plugin_id": "id", "boefje_name": "name"}
-        return next((value for key, value in translations.items() if key in error_message), None)
+                except UniqueViolation as error:
+                    raise DuplicatePlugin(error.field)
 
     def create_normalizer(self, normalizer: Normalizer) -> None:
         try:
             self.local_repo.by_id(normalizer.id)
-            raise DuplicatePlugin("id")
+            raise DuplicatePlugin(field="id")
         except KeyError:
             try:
                 plugin = self.local_repo.by_name(normalizer.name)
 
                 if plugin.types == "normalizer":
-                    raise DuplicatePlugin("name")
+                    raise DuplicatePlugin(field="name")
                 else:
                     self.plugin_storage.create_normalizer(normalizer)
             except KeyError:
@@ -177,12 +173,12 @@ def delete_settings(self, organisation_id: str, plugin_id: str):
         # We don't check the schema anymore because we can provide entries through the global environment as well
 
     def schema(self, plugin_id: str) -> dict | None:
-        try:
-            boefje = self.plugin_storage.boefje_by_id(plugin_id)
+        plugin = self._get_all_without_enabled().get(plugin_id)
 
-            return boefje.boefje_schema
-        except PluginNotFound:
-            return self.local_repo.schema(plugin_id)
+        if plugin is None or not isinstance(plugin, Boefje):
+            return None
+
+        return plugin.boefje_schema
 
     def cover(self, plugin_id: str) -> Path:
         try:
@@ -212,8 +208,8 @@ def set_enabled_by_id(self, plugin_id: str, organisation_id: str, enabled: bool)
 
         self.config_storage.upsert(organisation_id, plugin_id, enabled=enabled)
 
-    def _assert_settings_match_schema(self, all_settings: dict, plugin_id: str):
-        schema = self.schema(plugin_id)
+    def _assert_settings_match_schema(self, all_settings: dict, plugin_id: str, organisation_id: str):
+        schema = self.by_plugin_id(plugin_id, organisation_id).boefje_schema
 
         if schema:  # No schema means that there is nothing to assert
             try:

boefjes/boefjes/plugins/kat_answer_parser/normalize.py

@@ -10,4 +10,4 @@ def run(input_ooi: dict, raw: bytes) -> Iterable[NormalizerOutput]:
 
     bit_id = data["schema"].removeprefix("/bit/")
 
-    yield Config(ooi=input_ooi["primary_key"], bit_id=bit_id, config=data["answer"])
+    yield Config(ooi=data["answer_ooi"], bit_id=bit_id, config=data["answer"])

boefjes/boefjes/plugins/kat_fierce/boefje.json

@@ -1,9 +1,9 @@
 {
   "id": "fierce",
   "name": "Fierce",
-  "description": "Perform DNS reconnaissance using Fierce. Helps to locate non-contiguous IP space and hostnames against specified hostnames. No exploitation is performed.",
+  "description": "Perform DNS reconnaissance using Fierce. Helps to locate non-contiguous IP space and hostnames against specified hostnames. No exploitation is performed. Beware if your DNS is managed by an external party. This boefjes performs a brute force attack against the name server.",
   "consumes": [
     "Hostname"
   ],
-  "scan_level": 1
+  "scan_level": 3
 }

boefjes/boefjes/plugins/kat_kat_finding_types/kat_finding_types.json

@@ -132,6 +132,13 @@
     "impact": "System administrator ports should only be reachable from safe and known locations to reduce attack surface.",
     "recommendation": "Determine if this port should be reachable from the identified location. Limit access to reduce the attack surface if necessary."
   },
+  "KAT-REMOTE-DESKTOP-PORT": {
+    "description": "An open Microsoft Remote Desktop Protocol (RDP) port was detected.",
+    "source": "https://www.cloudflare.com/en-gb/learning/access-management/rdp-security-risks/",
+    "risk": "medium",
+    "impact": "Remote desktop ports are often the root cause in ransomware attacks, due to weak password usage, outdated software or insecure configurations.",
+    "recommendation": "Disable the Microsoft RDP service on port 3389 if this is publicly reachable. Add additional security layers, such as VPN access if these ports do require to be enabled to limit the attack surface."
+  },
   "KAT-OPEN-DATABASE-PORT": {
     "description": "A database port is open.",
     "source": "https://en.wikipedia.org/wiki/List_of_TCP_and_UDP_port_numbers",

boefjes/boefjes/plugins/kat_shodan_internetdb/main.py

@@ -1,6 +1,6 @@
 from ipaddress import ip_address
 
-import requests
+import httpx
 
 from boefjes.job_models import BoefjeMeta
 
@@ -12,7 +12,8 @@ def run(boefje_meta: BoefjeMeta) -> list[tuple[set, bytes | str]]:
     ip = boefje_meta.arguments["input"]["address"]
     if ip_address(ip).is_private:
         return [({"info/boefje"}, "Skipping private IP address")]
-    response = requests.get(f"https://internetdb.shodan.io/{ip}", timeout=REQUEST_TIMEOUT)
-    response.raise_for_status()
+    response = httpx.get(f"https://internetdb.shodan.io/{ip}", timeout=REQUEST_TIMEOUT)
+    if response.status_code != httpx.codes.NOT_FOUND:
+        response.raise_for_status()
 
     return [(set(), response.content)]

boefjes/boefjes/plugins/pdio_subfinder/boefje.json

@@ -1,13 +1,13 @@
 {
   "id": "pdio-subfinder",
   "name": "Subfinder",
-  "description": "A subdomain discovery tool. (projectdiscovery.io)",
+  "description": "A subdomain discovery tool. (projectdiscovery.io). Returns valid subdomains for websites using passive online sources. Beware that many of the online sources require their own API key to get more accurate data.",
   "consumes": [
     "Hostname"
   ],
   "environment_keys": [
     "SUBFINDER_RATE_LIMIT",
     "SUBFINDER_VERSION"
   ],
-  "scan_level": 2
+  "scan_level": 1
 }

boefjes/boefjes/sql/session.py

@@ -4,7 +4,7 @@
 from sqlalchemy.orm import Session
 from typing_extensions import Self
 
-from boefjes.storage.interfaces import IntegrityError, StorageError
+from boefjes.storage.interfaces import IntegrityError, StorageError, UniqueViolation
 
 logger = structlog.get_logger(__name__)
 
@@ -40,7 +40,7 @@ def __exit__(self, exc_type: type[Exception], exc_value: str, exc_traceback: str
             self.session.commit()
         except exc.IntegrityError as e:
             if isinstance(e.orig, errors.UniqueViolation):
-                raise IntegrityError(str(e.orig))
+                raise UniqueViolation(str(e.orig))
             raise IntegrityError("An integrity error occurred") from e
         except exc.DatabaseError as e:
             raise StorageError("A storage error occurred") from e

boefjes/boefjes/storage/interfaces.py

@@ -1,3 +1,4 @@
+import re
 from abc import ABC
 
 from boefjes.models import Boefje, Normalizer, Organisation, PluginType
@@ -17,6 +18,20 @@ def __init__(self, message: str):
         self.message = message
 
 
+class UniqueViolation(IntegrityError):
+    def __init__(self, message: str):
+        self.field = self._get_field_name(message)
+        self.message = message
+
+    def _get_field_name(self, message: str) -> str | None:
+        matches = re.findall(r"Key \((.*)\)=", message)
+
+        if matches:
+            return matches[0]
+
+        return None
+
+
 class SettingsNotConformingToSchema(StorageError):
     def __init__(self, plugin_id: str, validation_error: str):
         super().__init__(f"Settings for plugin {plugin_id} are not conform the plugin schema: {validation_error}")
@@ -56,8 +71,8 @@ def __init__(self, plugin_id: str):
 
 
 class DuplicatePlugin(NotAllowed):
-    def __init__(self, key: str):
-        super().__init__(f"Duplicate plugin {key}")
+    def __init__(self, field: str | None):
+        super().__init__(f"Duplicate plugin: a plugin with this {field} already exists")
 
 
 class OrganisationStorage(ABC):

boefjes/tests/integration/test_api.py

@@ -45,12 +45,12 @@ def test_cannot_add_plugin_reserved_id(test_client, organisation):
     boefje = Boefje(id="dns-records", name="My test boefje", static=False)
     response = test_client.post(f"/v1/organisations/{organisation.id}/plugins", content=boefje.model_dump_json())
     assert response.status_code == 400
-    assert response.json() == {"detail": "Duplicate plugin id"}
+    assert response.json() == {"detail": "Duplicate plugin: a plugin with this id already exists"}
 
     normalizer = Normalizer(id="kat_nmap_normalize", name="My test normalizer")
     response = test_client.post(f"/v1/organisations/{organisation.id}/plugins", content=normalizer.model_dump_json())
     assert response.status_code == 400
-    assert response.json() == {"detail": "Duplicate plugin id"}
+    assert response.json() == {"detail": "Duplicate plugin: a plugin with this id already exists"}
 
 
 def test_add_boefje(test_client, organisation):
@@ -80,7 +80,7 @@ def test_cannot_add_static_plugin_with_duplicate_name(test_client, organisation)
     boefje = Boefje(id="test_plugin", name="DNS records", static=False)
     response = test_client.post(f"/v1/organisations/{organisation.id}/plugins", content=boefje.model_dump_json())
     assert response.status_code == 400
-    assert response.json() == {"detail": "Duplicate plugin name"}
+    assert response.json() == {"detail": "Duplicate plugin: a plugin with this name already exists"}
 
 
 def test_cannot_add_plugin_with_duplicate_name(test_client, organisation):
@@ -91,7 +91,7 @@ def test_cannot_add_plugin_with_duplicate_name(test_client, organisation):
     boefje = Boefje(id="test_plugin_2", name="My test boefje", static=False)
     response = test_client.post(f"/v1/organisations/{organisation.id}/plugins", content=boefje.model_dump_json())
     assert response.status_code == 400
-    assert response.json() == {"detail": "Duplicate plugin name"}
+    assert response.json() == {"detail": "Duplicate plugin: a plugin with this name already exists"}
 
     normalizer = Normalizer(id="test_normalizer", name="My test normalizer", static=False)
     response = test_client.post(f"/v1/organisations/{organisation.id}/plugins", content=normalizer.model_dump_json())
@@ -169,6 +169,16 @@ def test_cannot_create_boefje_with_invalid_schema(test_client, organisation):
     assert r.status_code == 422
 
 
+def test_schema_is_taken_from_disk(test_client, organisation, session):
+    # creates a database record of dns-records
+    test_client.patch(f"/v1/organisations/{organisation.id}/plugins/dns-records", json={"enabled": True})
+    session.execute("UPDATE boefje set schema = null where plugin_id = 'dns-records'")
+    session.commit()
+
+    response = test_client.get(f"/v1/organisations/{organisation.id}/plugins/dns-records").json()
+    assert response["boefje_schema"] is not None
+
+
 def test_cannot_set_invalid_cron(test_client, organisation):
     boefje = Boefje(id="test_plugin", name="My test boefje", description="123").model_dump(mode="json")
     boefje["cron"] = "bad format"

boefjes/tests/plugins/test_answer_parser.py

@@ -13,10 +13,10 @@ def test_config_yielded(normalizer_runner):
         normalizer_runner.run(meta, bytes(raw, "UTF-8"))
 
     with pytest.raises(ValidationError):
-        raw = '{"schema": "/bit/port-classification-ip", "answer": [{"key": "test"}]}'
+        raw = '{"schema": "/bit/port-classification-ip", "answer": [{"key": "test"}], "answer_ooi": "Network|internet"}'
         normalizer_runner.run(meta, bytes(raw, "UTF-8"))
 
-    raw = '{"schema": "/bit/port-classification-ip", "answer": {"key": "test"}}'
+    raw = '{"schema": "/bit/port-classification-ip", "answer": {"key": "test"}, "answer_ooi": "Network|internet"}'
     output = normalizer_runner.run(meta, bytes(raw, "UTF-8"))
 
     assert len(output.observations) == 1

docker-compose.yml

@@ -23,6 +23,9 @@ services:
       test: ["CMD", "gosu", "postgres", "pg_isready"]
       interval: 10s
       retries: 10
+    # Django runserver does not limit the number of threads. We need to increase
+    # the maximum number of connection to make sure that we don't hit the limit.
+    command: -c max_connections=500
     volumes:
       - postgres-data:/var/lib/postgresql/data
       - ./init-user-db.sh:/docker-entrypoint-initdb.d/init-user-db.sh