Support setting a custom JSON schema for copied boefjes (#3344)

Signed-off-by: Donny Peeters <[email protected]>
Co-authored-by: Jan Klopper <[email protected]>
Co-authored-by: stephanie0x00 <[email protected]>

boefjes/boefjes/dependencies/plugins.py

@@ -158,7 +158,12 @@ def delete_settings(self, organisation_id: str, plugin_id: str):
             self.set_enabled_by_id(plugin_id, organisation_id, False)
 
     def schema(self, plugin_id: str) -> dict | None:
-        return self.local_repo.schema(plugin_id)
+        try:
+            boefje = self.plugin_storage.boefje_by_id(plugin_id)
+
+            return boefje.schema
+        except PluginNotFound:
+            return self.local_repo.schema(plugin_id)
 
     def cover(self, plugin_id: str) -> Path:
         try:

boefjes/boefjes/katalogus/plugins.py

@@ -3,7 +3,8 @@
 
 from fastapi import APIRouter, Body, Depends, HTTPException, status
 from fastapi.responses import FileResponse, JSONResponse, Response
-from pydantic import BaseModel, Field
+from jsonschema.validators import Draft202012Validator
+from pydantic import BaseModel, Field, field_validator
 
 from boefjes.dependencies.plugins import (
     PluginService,
@@ -90,6 +91,8 @@ def get_plugin(
 @router.post("/plugins", status_code=status.HTTP_201_CREATED)
 def add_plugin(plugin: PluginType, plugin_service: PluginService = Depends(get_plugin_service)):
     with plugin_service as service:
+        plugin.static = False  # Creation through the API implies that these cannot be static
+
         if plugin.type == "boefje":
             return service.create_boefje(plugin)
 
@@ -124,9 +127,19 @@ class BoefjeIn(BaseModel):
     scan_level: int = 1
     consumes: set[str] = Field(default_factory=set)
     produces: set[str] = Field(default_factory=set)
+    schema: dict | None = None
     oci_image: str | None = None
     oci_arguments: list[str] = Field(default_factory=list)
 
+    @field_validator("schema")
+    @classmethod
+    def json_schema_valid(cls, schema: dict | None) -> dict | None:
+        if schema is not None:
+            Draft202012Validator.check_schema(schema)
+            return schema
+
+        return None
+
 
 @router.patch("/boefjes/{boefje_id}", status_code=status.HTTP_204_NO_CONTENT)
 def update_boefje(

boefjes/boefjes/katalogus/root.py

@@ -5,6 +5,7 @@
 import structlog
 from fastapi import APIRouter, FastAPI, Request, status
 from fastapi.responses import JSONResponse, RedirectResponse
+from jsonschema.exceptions import SchemaError
 from opentelemetry import trace
 from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
@@ -96,6 +97,14 @@ def storage_error_handler(request: Request, exc: StorageError):
     )
 
 
+@app.exception_handler(SchemaError)
+def schema_error_handler(request: Request, exc: StorageError):
+    return JSONResponse(
+        status_code=status.HTTP_400_BAD_REQUEST,
+        content={"message": "Invalid jsonschema provided"},
+    )
+
+
 class ServiceHealth(BaseModel):
     service: str
     healthy: bool = False

boefjes/boefjes/local_repository.py

@@ -11,6 +11,7 @@
     BOEFJES_DIR,
     ENTRYPOINT_NORMALIZERS,
     NORMALIZER_DEFINITION_FILE,
+    SCHEMA_FILE,
     BoefjeResource,
     ModuleException,
     NormalizerResource,
@@ -52,10 +53,10 @@ def schema(self, id_: str) -> dict | None:
         if id_ not in boefjes:
             return None
 
-        path = boefjes[id_].path / "schema.json"
+        path = boefjes[id_].path / SCHEMA_FILE
 
         if not path.exists():
-            logger.debug("Did not find schema for boefje %s", boefjes[id_])
+            logger.debug("Did not find schema for boefje %s", id_)
             return None
 
         return json.loads(path.read_text())

boefjes/boefjes/migrations/versions/5be152459a7b_introduce_schema_field_to_boefje_model.py

@@ -0,0 +1,62 @@
+"""Introduce schema field to Boefje model
+
+Revision ID: 5be152459a7b
+Revises: f9de6eb7824b
+Create Date: 2024-08-08 14:47:12.582017
+
+"""
+
+import logging
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.orm import sessionmaker
+
+from boefjes.local_repository import get_local_repository
+from boefjes.sql.plugin_storage import create_plugin_storage
+from boefjes.storage.interfaces import PluginNotFound
+
+# revision identifiers, used by Alembic.
+revision = "5be152459a7b"
+down_revision = "f9de6eb7824b"
+branch_labels = None
+depends_on = None
+
+logger = logging.getLogger(__name__)
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column("boefje", sa.Column("schema", sa.JSON(), nullable=True))
+
+    local_repo = get_local_repository()
+    session = sessionmaker(bind=op.get_bind())()
+
+    with create_plugin_storage(session) as storage:
+        plugins = local_repo.get_all()
+        logger.info("Found %s plugins", len(plugins))
+
+        for plugin in local_repo.get_all():
+            schema = local_repo.schema(plugin.id)
+
+            if schema:
+                try:
+                    # This way we avoid the safeguard that updating static boefjes is not allowed
+                    instance = storage._db_boefje_instance_by_id(plugin.id)
+                    instance.schema = schema
+                    storage.session.add(instance)
+                    logger.info("Updated database entry for plugin %s", plugin.id)
+                except PluginNotFound:
+                    logger.info("No database entry for plugin %s", plugin.id)
+                    continue
+            else:
+                logger.info("No schema present for plugin %s", plugin.id)
+
+    session.close()
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("boefje", "schema")
+    # ### end Alembic commands ###

boefjes/boefjes/migrations/versions/f9de6eb7824b_introduce_boefjeconfig_model.py

@@ -10,11 +10,12 @@
 
 import sqlalchemy as sa
 from alembic import op
-from sqlalchemy.orm import sessionmaker
+from psycopg2._json import Json
+from psycopg2.extensions import register_adapter
+from psycopg2.extras import execute_values
 
 from boefjes.local_repository import get_local_repository
-from boefjes.sql.plugin_storage import create_plugin_storage
-from boefjes.storage.interfaces import PluginNotFound
+from boefjes.models import Boefje, Normalizer
 
 # revision identifiers, used by Alembic.
 revision = "f9de6eb7824b"
@@ -57,72 +58,136 @@ def upgrade() -> None:
     op.add_column("boefje", sa.Column("static", sa.Boolean(), server_default="false", nullable=False))
     op.add_column("normalizer", sa.Column("static", sa.Boolean(), server_default="false", nullable=False))
 
+    register_adapter(dict, Json)
+
     local_plugins = {plugin.id: plugin for plugin in get_local_repository().get_all()}
     connection = op.get_bind()
-    session = sessionmaker(bind=connection)()
-
-    with create_plugin_storage(session) as storage:
-        # Get unique plugin_ids from the settings table for boefjes that do not exist yet in the database
-        query = """
-        SELECT DISTINCT s.plugin_id FROM settings s left join boefje b on b.plugin_id = s.plugin_id
-            where b.plugin_id IS NULL
-        """
-        for plugin_id_output in op.get_bind().execute(query).fetchall():
-            plugin_id = plugin_id_output[0]
-            if plugin_id not in local_plugins:
-                raise ValueError(f"Invalid plugin id found: {plugin_id}")
-
-            # Since settings are boefje-only at this moment
-            if local_plugins[plugin_id].type != "boefje":
-                raise ValueError(f"Settings for normalizer or bit found: {plugin_id}. Remove these entries first.")
-
-            try:
-                storage.boefje_by_id(plugin_id)
-                continue  # The Boefje already exists
-            except PluginNotFound:
-                pass  # The raw query bypasses the session "cache", so this just checks for duplicates
-
-            storage.create_boefje(local_plugins[plugin_id])  # type: ignore
-
-        query = """
-        SELECT DISTINCT p.plugin_id FROM plugin_state p left join boefje b on b.plugin_id = p.plugin_id
-            where b.plugin_id IS NULL
-        """
-
-        for plugin_id_output in op.get_bind().execute(query).fetchall():
-            plugin_id = plugin_id_output[0]
-            if plugin_id not in local_plugins:
-                logger.warning("Unknown plugin id found: %s. You might have to re-enable the plugin!", plugin_id)
-                continue
-
-            try:
-                storage.boefje_by_id(plugin_id)
-                continue  # The Boefje already exists
-            except PluginNotFound:
-                pass  # The raw query bypasses the session "cache", so this just checks for duplicates
-
-            if local_plugins[plugin_id].type == "boefje":
-                storage.create_boefje(local_plugins[plugin_id])  # type: ignore
-
-        query = """
-        SELECT DISTINCT p.plugin_id FROM plugin_state p left join normalizer n on n.plugin_id = p.plugin_id
-            where n.plugin_id IS NULL
-        """
-
-        for plugin_id_output in op.get_bind().execute(query).fetchall():
-            plugin_id = plugin_id_output[0]
-            if plugin_id not in local_plugins:
-                logger.warning("Unknown plugin id found: %s. You might have to re-enable the plugin!", plugin_id)
-                continue
-
-            try:
-                storage.normalizer_by_id(plugin_id)
-                continue  # The Normalizer already exists
-            except PluginNotFound:
-                pass  # The raw query bypasses the session "cache", so this just checks for duplicates
-
-            if local_plugins[plugin_id].type == "normalizer":
-                storage.create_normalizer(local_plugins[plugin_id])  # type: ignore
+
+    # Get unique plugin_ids from the settings table for boefjes that do not exist yet in the database
+    query = """
+    SELECT DISTINCT s.plugin_id FROM settings s left join boefje b on b.plugin_id = s.plugin_id
+        where b.plugin_id IS NULL
+    """  # noqa: S608
+
+    to_insert: list[Boefje] = []
+
+    for plugin_id_output in connection.execute(query).fetchall():
+        plugin_id = plugin_id_output[0]
+        if plugin_id not in local_plugins:
+            raise ValueError(f"Invalid plugin id found: {plugin_id}")
+
+        # Since settings are boefje-only at this moment
+        if local_plugins[plugin_id].type != "boefje":
+            raise ValueError(f"Settings for normalizer or bit found: {plugin_id}. Remove these entries first.")
+
+        res = connection.execute(f"SELECT id FROM boefje where plugin_id = '{plugin_id}'")  # noqa: S608
+        if res.fetchone() is not None:
+            continue  # The Boefje already exists
+
+        if local_plugins[plugin_id].type == "boefje":
+            to_insert.append(local_plugins[plugin_id])
+
+    entries = [
+        (
+            boefje.id,
+            boefje.name,
+            boefje.description,
+            str(boefje.scan_level),
+            list(boefje.consumes),
+            list(boefje.produces),
+            boefje.environment_keys,
+            boefje.oci_image,
+            boefje.oci_arguments,
+            boefje.version,
+        )
+        for boefje in to_insert
+    ]
+    query = """INSERT INTO boefje (plugin_id, name, description, scan_level, consumes, produces, environment_keys,
+        oci_image, oci_arguments, version) values %s"""
+
+    with connection.begin():
+        cursor = connection.connection.cursor()
+        execute_values(cursor, query, entries)
+
+    to_insert = []
+
+    query = """
+    SELECT DISTINCT p.plugin_id FROM plugin_state p left join boefje b on b.plugin_id = p.plugin_id
+        where b.plugin_id IS NULL
+    """
+
+    for plugin_id_output in connection.execute(query).fetchall():
+        plugin_id = plugin_id_output[0]
+        if plugin_id not in local_plugins:
+            logger.warning("Unknown plugin id found: %s. You might have to re-enable the plugin!", plugin_id)
+            continue
+
+        res = connection.execute(f"SELECT id FROM boefje where plugin_id = '{plugin_id}'")  # noqa: S608
+        if res.fetchone() is not None:
+            continue  # The Boefje already exists
+
+        if local_plugins[plugin_id].type == "boefje":
+            to_insert.append(local_plugins[plugin_id])
+
+    entries = [
+        (
+            boefje.id,
+            boefje.name,
+            boefje.description,
+            str(boefje.scan_level),
+            list(boefje.consumes),
+            list(boefje.produces),
+            boefje.environment_keys,
+            boefje.oci_image,
+            boefje.oci_arguments,
+            boefje.version,
+        )
+        for boefje in to_insert
+    ]
+    query = """INSERT INTO boefje (plugin_id, name, description, scan_level, consumes, produces, environment_keys,
+        oci_image, oci_arguments, version) values %s"""  # noqa: S608
+
+    with connection.begin():
+        cursor = connection.connection.cursor()
+        execute_values(cursor, query, entries)
+
+    normalizers_to_insert: list[Normalizer] = []
+    query = """
+    SELECT DISTINCT p.plugin_id FROM plugin_state p left join normalizer n on n.plugin_id = p.plugin_id
+        where n.plugin_id IS NULL
+    """  # noqa: S608
+
+    for plugin_id_output in connection.execute(query).fetchall():
+        plugin_id = plugin_id_output[0]
+        if plugin_id not in local_plugins:
+            logger.warning("Unknown plugin id found: %s. You might have to re-enable the plugin!", plugin_id)
+            continue
+
+        res = connection.execute(f"SELECT id FROM normalizer where plugin_id = '{plugin_id}'")  # noqa: S608
+        if res.fetchone() is not None:
+            continue  # The Normalizer already exists
+
+        if local_plugins[plugin_id].type == "normalizer":
+            normalizers_to_insert.append(local_plugins[plugin_id])
+
+    normalizer_entries = [
+        (
+            normalizer.id,
+            normalizer.name,
+            normalizer.description,
+            normalizer.consumes,
+            normalizer.produces,
+            normalizer.environment_keys,
+            normalizer.version,
+        )
+        for normalizer in normalizers_to_insert
+    ]
+    query = """INSERT INTO normalizer (plugin_id, name, description, consumes, produces, environment_keys, version)
+        values %s"""  # noqa: S608
+
+    with connection.begin():
+        cursor = connection.connection.cursor()
+        execute_values(cursor, query, normalizer_entries)
 
     with connection.begin():
         connection.execute("""