Tasks contain property that tells the boefjerunner what network it is supposed to run on.

boefjes/boefjes/__main__.py

@@ -6,7 +6,6 @@
 
 from boefjes.app import get_runtime_manager
 from boefjes.config import settings
-from boefjes.runtime_interfaces import WorkerManager
 
 with settings.log_cfg.open() as f:
     logging.config.dictConfig(json.load(f))
@@ -35,21 +34,20 @@
 
 
 @click.command()
-@click.argument("worker_type", type=click.Choice([q.value for q in WorkerManager.Queue]))
+@click.argument("worker_type", type=click.Choice(["boefje", "normalizer"]))
 @click.option("--log-level", type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"]), help="Log level", default="INFO")
 def cli(worker_type: str, log_level: str) -> None:
     logger.setLevel(log_level)
     logger.info("Starting runtime for %s", worker_type)
 
-    queue = WorkerManager.Queue(worker_type)
-    runtime = get_runtime_manager(settings, queue, log_level)
+    runtime = get_runtime_manager(settings, worker_type, log_level)
 
     if worker_type == "boefje":
         import boefjes.api
 
         boefjes.api.run()
 
-    runtime.run(queue)
+    runtime.run(worker_type)
 
 
 if __name__ == "__main__":

boefjes/boefjes/app.py

@@ -49,8 +49,8 @@ def __init__(
 
         self.exited = False
 
-    def run(self, queue_type: WorkerManager.Queue) -> None:
-        logger.info("Created worker pool for queue '%s'", queue_type.value)
+    def run(self, queue_type: str) -> None:
+        logger.info("Created worker pool for queue '%s'", queue_type)
 
         self.workers = [
             ctx.Process(target=_start_working, args=self._worker_args()) for _ in range(self.settings.pool_size)
@@ -80,7 +80,7 @@ def run(self, queue_type: WorkerManager.Queue) -> None:
 
                 raise
 
-    def _fill_queue(self, task_queue: Queue, queue_type: WorkerManager.Queue) -> None:
+    def _fill_queue(self, task_queue: Queue, queue_type: str) -> None:
         if task_queue.qsize() > self.settings.pool_size:
             time.sleep(self.settings.worker_heartbeat)
             return
@@ -95,7 +95,7 @@ def _fill_queue(self, task_queue: Queue, queue_type: WorkerManager.Queue) -> Non
 
         # We do not target a specific queue since we start one runtime for all organisations
         # and queue ids contain the organisation_id
-        queues = [q for q in queues if q.id.startswith(queue_type.value) and q.size > 0]
+        queues = [q for q in queues if q.id.startswith(queue_type) and q.size > 0]
 
         logger.debug("Found queues: %s", [queue.id for queue in queues])
 
@@ -265,14 +265,15 @@ def _start_working(
                 logger.exception("Could not patch scheduler task to %s", status.value)
 
 
-def get_runtime_manager(settings: Settings, queue: WorkerManager.Queue, log_level: str) -> WorkerManager:
+def get_runtime_manager(settings: Settings, queue: str, log_level: str) -> WorkerManager:
     local_repository = get_local_repository()
 
     session = sessionmaker(bind=get_engine())()
     plugin_service = PluginService(create_plugin_storage(session), create_config_storage(session), local_repository)
 
     item_handler: Handler
-    if queue is WorkerManager.Queue.BOEFJES:
+
+    if queue == "boefje":
         item_handler = BoefjeHandler(LocalBoefjeJobRunner(local_repository), plugin_service, bytes_api_client)
     else:
         item_handler = NormalizerHandler(
@@ -281,7 +282,7 @@ def get_runtime_manager(settings: Settings, queue: WorkerManager.Queue, log_leve
 
     return SchedulerWorkerManager(
         item_handler,
-        SchedulerAPIClient(str(settings.scheduler_api)),  # Do not share a session between workers
+        SchedulerAPIClient(base_url=str(settings.scheduler_api)),  # Do not share a session between workers
         settings,
         log_level,
     )

boefjes/boefjes/clients/scheduler_client.py

@@ -1,6 +1,8 @@
 import datetime
+import json
 import uuid
 from enum import Enum
+from typing import Any
 
 from httpx import Client, HTTPTransport, Response
 from pydantic import BaseModel, TypeAdapter
@@ -39,6 +41,17 @@ class Task(BaseModel):
     modified_at: datetime.datetime
 
 
+class Filter(BaseModel):
+    column: str
+    field: str
+    operator: str
+    value: Any
+
+
+class QueuePopRequest(BaseModel):
+    filters: list[Filter]
+
+
 class SchedulerClientInterface:
     def get_queues(self) -> list[Queue]:
         raise NotImplementedError()
@@ -73,7 +86,31 @@ def get_queues(self) -> list[Queue]:
         return TypeAdapter(list[Queue]).validate_json(response.content)
 
     def pop_item(self, queue_id: str) -> Task | None:
-        response = self._session.post(f"/queues/{queue_id}/pop")
+        filters: list[Filter] = []
+
+        if settings.runner_type == "boefje":
+            # Client should only pop tasks that lie on a network that the
+            # runner is capable of reaching (e.g. the internet)
+            filters.append(
+                Filter(
+                    column="data", field="network", operator="<@", value=json.dumps(settings.boefje_reachable_networks)
+                )
+            )
+
+            # Client should only pop tasks that have requirements that this runner is capable of (e.g. being able
+            # to handle ipv6 requests)
+            filters.append(
+                Filter(
+                    column="data",
+                    field="requirements",
+                    operator="<@",
+                    value=json.dumps(settings.boefje_task_capabilities),
+                )
+            )
+
+        response = self._session.post(
+            f"/queues/{queue_id}/pop", data=QueuePopRequest(filters=filters).model_dump_json()
+        )
         self._verify_response(response)
 
         return TypeAdapter(Task | None).validate_json(response.content)

boefjes/boefjes/config.py

@@ -8,6 +8,7 @@
 from pydantic_settings.sources import EnvSettingsSource
 
 from boefjes.models import EncryptionMiddleware
+from boefjes.runtime_interfaces import WorkerManager
 
 BASE_DIR: Path = Path(__file__).parent.resolve()
 
@@ -120,6 +121,20 @@ class Settings(BaseSettings):
         None, description="OpenTelemetry endpoint", validation_alias="SPAN_EXPORT_GRPC_ENDPOINT"
     )
 
+    boefje_reachable_networks: list[str] | None = Field(
+        None,
+        description="List of networks the boefje-runner can reach",
+        examples=[["Network|internet", "Network|dentist"], []],
+    )
+
+    boefje_task_capabilities: list[str] | None = Field(
+        None,
+        description="List of technical requirements the boefje-runner is capable of running",
+        examples=[[], ["ipv4", "wifi-pineapple"]],
+    )
+
+    runner_type: WorkerManager.Queue | None = Field(None, examples=["boefje", "normalizer"])
+
     logging_format: Literal["text", "json"] = Field("text", description="Logging format")
 
     outgoing_request_timeout: int = Field(30, description="Timeout for outgoing HTTP requests")

boefjes/boefjes/runtime_interfaces.py

@@ -1,4 +1,4 @@
-from enum import Enum
+from typing import Literal
 
 from boefjes.job_models import BoefjeMeta, NormalizerMeta, NormalizerResults
 
@@ -19,11 +19,9 @@ def run(self, normalizer_meta: NormalizerMeta, raw: bytes) -> NormalizerResults:
 
 
 class WorkerManager:
-    class Queue(Enum):
-        BOEFJES = "boefje"
-        NORMALIZERS = "normalizer"
+    Queue = Literal["boefje", "normalizer"]
 
-    def run(self, queue: Queue) -> None:
+    def run(self, queue: str) -> None:
         raise NotImplementedError()
 
 

boefjes/entrypoint.sh

@@ -5,9 +5,9 @@ set -e
 shopt -s nocasematch
 
 if [ "$1" = "boefje" ]; then
-    exec python -m boefjes boefje
+    exec env BOEFJES_RUNNER_TYPE="$1" python -m boefjes boefje
 elif [ "$1" = "normalizer" ]; then
-    exec python -m boefjes normalizer
+    exec env BOEFJES_RUNNER_TYPE="$1" python -m boefjes normalizer
 fi
 
 # The migrations and seed are for the KATalogus. They are not inside the if because this way
@@ -17,7 +17,7 @@ if [ "$DATABASE_MIGRATION" = "1" ] || [[ $DATABASE_MIGRATION == "true" ]]; then
 fi
 
 if [ "$1" = "katalogus" ]; then
-    exec python -m uvicorn --host 0.0.0.0 boefjes.katalogus.root:app
+    exec env BOEFJES_RUNNER_TYPE="$1" python -m uvicorn --host 0.0.0.0 boefjes.katalogus.root:app
 fi
 
 exec "$@"

boefjes/tests/conftest.py

@@ -31,7 +31,7 @@
     get_normalizer_resource,
 )
 from boefjes.models import Organisation
-from boefjes.runtime_interfaces import Handler, WorkerManager
+from boefjes.runtime_interfaces import Handler
 from boefjes.sql.config_storage import SQLConfigStorage, create_encrypter
 from boefjes.sql.db import SQL_BASE, get_engine
 from boefjes.sql.organisation_storage import SQLOrganisationStorage, get_organisations_store
@@ -81,13 +81,13 @@ def pop_item(self, queue: str) -> Task | None:
         time.sleep(self.sleep_time)
 
         try:
-            if WorkerManager.Queue.BOEFJES.value in queue:
+            if "boefje" in queue:
                 p_item = TypeAdapter(Task).validate_json(self.boefje_responses.pop(0))
                 self._popped_items[str(p_item.id)] = p_item
                 self._tasks[str(p_item.id)] = self._task_from_id(p_item.id)
                 return p_item
 
-            if WorkerManager.Queue.NORMALIZERS.value in queue:
+            if "normalizer" in queue:
                 p_item = TypeAdapter(Task).validate_json(self.normalizer_responses.pop(0))
                 self._popped_items[str(p_item.id)] = p_item
                 self._tasks[str(p_item.id)] = self._task_from_id(p_item.id)

boefjes/tests/test_app.py

@@ -6,14 +6,13 @@
 
 from boefjes.app import SchedulerWorkerManager, get_runtime_manager
 from boefjes.config import Settings
-from boefjes.runtime_interfaces import WorkerManager
 from tests.conftest import MockHandler, MockSchedulerClient
 from tests.loading import get_dummy_data
 
 
 def test_one_process(manager: SchedulerWorkerManager, item_handler: MockHandler) -> None:
     with pytest.raises(KeyboardInterrupt):
-        manager.run(WorkerManager.Queue.BOEFJES)
+        manager.run("boefje")
 
     items = item_handler.get_all()
     assert len(items) == 2
@@ -38,7 +37,7 @@ def test_two_processes(manager: SchedulerWorkerManager, item_handler: MockHandle
     manager.task_queue = Manager().Queue()
 
     with pytest.raises(KeyboardInterrupt):
-        manager.run(WorkerManager.Queue.BOEFJES)
+        manager.run("boefje")
 
     items = item_handler.get_all()
     assert len(items) == 2
@@ -64,7 +63,7 @@ def test_two_processes_exception(manager: SchedulerWorkerManager, item_handler:
 
     manager.settings.pool_size = 2
     with pytest.raises(KeyboardInterrupt):
-        manager.run(WorkerManager.Queue.BOEFJES)
+        manager.run("boefje")
 
     assert item_handler.queue.empty()
     assert manager.scheduler_client.log_path.exists()
@@ -81,7 +80,7 @@ def test_two_processes_handler_exception(manager: SchedulerWorkerManager, item_h
     manager.settings.pool_size = 2
     manager.task_queue = Manager().Queue()
     with pytest.raises(KeyboardInterrupt):
-        manager.run(WorkerManager.Queue.BOEFJES)
+        manager.run("boefje")
 
     items = item_handler.get_all()
     assert len(items) == 1
@@ -137,7 +136,7 @@ def test_two_processes_cleanup_unfinished_tasks(
     item_handler.sleep_time = 200
 
     with pytest.raises(KeyboardInterrupt):
-        manager.run(WorkerManager.Queue.BOEFJES)
+        manager.run("boefje")
 
     items = item_handler.get_all()
     assert len(items) == 0
@@ -161,7 +160,7 @@ def test_two_processes_cleanup_unfinished_tasks(
 
 def test_normalizer_queue(manager: SchedulerWorkerManager, item_handler: MockHandler) -> None:
     with pytest.raises(KeyboardInterrupt):
-        manager.run(WorkerManager.Queue.NORMALIZERS)
+        manager.run("normalizer")
 
     items = item_handler.get_all()
     assert len(items) == 1
@@ -179,7 +178,7 @@ def test_null(manager: SchedulerWorkerManager, tmp_path: Path, item_handler: Moc
     )
 
     with pytest.raises(KeyboardInterrupt):
-        manager.run(WorkerManager.Queue.BOEFJES)
+        manager.run("boefje")
 
     items = item_handler.get_all()
     patched_tasks = manager.scheduler_client.get_all_patched_tasks()
@@ -197,5 +196,5 @@ def test_null(manager: SchedulerWorkerManager, tmp_path: Path, item_handler: Moc
 
 
 def test_create_manager():
-    get_runtime_manager(Settings(), WorkerManager.Queue.BOEFJES, "INFO")
-    get_runtime_manager(Settings(), WorkerManager.Queue.NORMALIZERS, "INFO")
+    get_runtime_manager(Settings(), "boefje", "INFO")
+    get_runtime_manager(Settings(), "normalizer", "INFO")

mula/scheduler/models/ooi.py

@@ -21,6 +21,7 @@ class OOI(BaseModel):
     primary_key: str
     object_type: str
     scan_profile: ScanProfile
+    network: str | None = None
 
 
 class ScanProfileMutation(BaseModel):

mula/scheduler/models/task.py

@@ -121,6 +121,9 @@ class BoefjeTask(BaseModel):
 
     dispatches: list[Normalizer] = Field(default_factory=list)
 
+    requirements: list[str] = []
+    network: str = "Network|internet"
+
     @property
     def hash(self) -> str:
         """Make BoefjeTask hashable, so that we can de-duplicate it when used