feat: kafka message dispatcher

backend/.dockerignore

@@ -4,6 +4,7 @@
 .git*
 .pytest*
 .idea*
+.venv
 .dockerignore
 Dockerfile
 db

backend/core/views.py

@@ -1315,6 +1315,7 @@ class Meta:
             "owner": ["exact"],
             "findings": ["exact"],
             "eta": ["exact", "lte", "gte", "lt", "gt", "month", "year"],
+            "ref_id": ["exact"],
         }
 
 
@@ -4544,10 +4545,14 @@ class RequirementAssessmentViewSet(BaseModelViewSet):
     model = RequirementAssessment
     filterset_fields = [
         "folder",
+        "folder__name",
         "evidences",
         "compliance_assessment",
         "applied_controls",
         "security_exceptions",
+        "requirement__ref_id",
+        "compliance_assessment__ref_id",
+        "compliance_assessment__assets__ref_id",
     ]
     search_fields = ["requirement__name", "requirement__description"]
 

config/make_config.py

@@ -85,6 +85,88 @@ def get_config():
             ).ask(),
         }
 
+    # Kafka Dispatcher Configuration
+    enable_dispatcher = questionary.confirm(
+        "Would you like to configure the Kafka dispatcher? This requires a kafka broker to be running.",
+        default=False,
+    ).ask()
+
+    if enable_dispatcher:
+        config["kafka_dispatcher"] = {
+            "enabled": True,
+            "broker_url": questionary.text(
+                "Enter the Kafka broker URL (e.g., localhost:9092):",
+                default="localhost:9092",
+            ).ask(),
+            "observation_topic": questionary.text(
+                "Enter the Kafka topic for dispatcher events (e.g., observation):",
+                default="observation",
+            ).ask(),
+            "errors_topic": questionary.text(
+                "Enter the topic name for error messages (e.g., errors):",
+                default="errors",
+            ).ask(),
+            "authentication": questionary.select(
+                "How would you like to authenticate requests to the CISO Assistant API using the dispatcher (credentials/token)?",
+                choices=["credentials", "token"],
+                default="credentials",
+            ).ask(),
+        }
+        if config["kafka_dispatcher"]["authentication"] == "credentials":
+            config["kafka_dispatcher"]["credentials"] = {
+                "user_email": questionary.text(
+                    "Enter the email of the CISO Assistant user account"
+                ).ask(),
+                "user_password": questionary.password(
+                    "Enter the password of the CISO Assistant user account"
+                ).ask(),
+            }
+            config["kafka_dispatcher"]["auto_renew_session"] = questionary.confirm(
+                "Enable silent reauthentication to the CISO Assistant API on session expiry?",
+                default=True,
+            ).ask()
+        elif config["kafka_dispatcher"]["authentication"] == "token":
+            config["kafka_dispatcher"]["token"] = questionary.password(
+                "Enter access token"
+            ).ask()
+        kafka_use_auth = questionary.confirm(
+            "Does your Kafka broker require authentication?",
+            default=False,
+        ).ask()
+        config["kafka_dispatcher"]["use_auth"] = kafka_use_auth
+        if kafka_use_auth:
+            config["kafka_dispatcher"]["sasl_mechanism"] = "PLAIN"
+            config["kafka_dispatcher"]["sasl_username"] = questionary.text(
+                "Enter the username of your Kafka service account"
+            ).ask()
+            config["kafka_dispatcher"]["sasl_password"] = questionary.password(
+                "Enter the password of your Kafka service account"
+            ).ask()
+
+        use_s3 = questionary.confirm(
+            "Would you like to connect a S3 bucket to the dispatcher? This can be used e.g. to upload files to the CISO Assistant backend.",
+            default=True,
+        ).ask()
+
+        config["kafka_dispatcher"]["s3_url"] = ""
+        config["kafka_dispatcher"]["s3_access_key"] = ""
+        config["kafka_dispatcher"]["s3_secret_key"] = ""
+        if use_s3:
+            config["kafka_dispatcher"]["s3_url"] = questionary.text(
+                "Enter the S3 storage URL (e.g., http://localhost:9000)",
+                default="http://localhost:9000",
+            ).ask()
+            config["kafka_dispatcher"]["s3_access_key"] = questionary.text(
+                "Enter the S3 access key (leave blank if using public S3 storage)",
+            ).ask()
+            config["kafka_dispatcher"]["s3_secret_key"] = ""
+            if config["kafka_dispatcher"]["s3_access_key"]:
+                config["kafka_dispatcher"]["s3_secret_key"] = questionary.password(
+                    "Enter the S3 secret key",
+                ).ask()
+    else:
+        config["kafka_dispatcher"] = {"enabled": False}
+
     # Debug mode for local development
     config["enable_debug"] = questionary.confirm(
         "Enable debug mode?", default=True if config["db"] == "sqlite" else False
@@ -105,22 +187,25 @@ def generate_compose_file(config):
     except Exception as e:
         print(f"[red]Error: Template {template_name} not found![/red]")
         print(
-            f"[yellow]Please ensure you have the following template file in your templates directory:[/yellow]"
+            "[yellow]Please ensure you have the following template file in your templates directory:[/yellow]"
         )
         print(f"[yellow]templates/{template_name}[/yellow]")
         print(
-            f"[yellow]Expected template name format: docker-compose-local-<db>-<proxy>.yml.j2[/yellow]"
+            "[yellow]Expected template name format: docker-compose-local-<db>-<proxy>.yml.j2[/yellow]"
         )
-        print(f"[yellow]Where <db> is one of: sqlite, postgresql[/yellow]")
-        print(f"[yellow]And <proxy> is one of: caddy, traefik[/yellow]")
+        print("[yellow]Where <db> is one of: sqlite, postgresql[/yellow]")
+        print("[yellow]And <proxy> is one of: caddy, traefik[/yellow]")
         raise e
 
     # Render template with configuration
     compose_content = template.render(config)
 
+    lines = compose_content.splitlines()
+    filtered = [line for line in lines if line.strip() != ""]
+
     # Write to docker-compose-custom.yml
     with open("docker-compose-custom.yml", "w") as f:
-        f.write(compose_content)
+        f.write("\n".join(filtered))
 
 
 def validate_cert_paths(config):
@@ -170,7 +255,7 @@ def main():
     if config["need_mailer"]:
         print("4. Verify email configuration settings")
     print(
-        f"5. Run './docker-compose.sh' and follow the instructions to create the first admin user"
+        "5. Run './docker-compose.sh' and follow the instructions to create the first admin user"
     )
     print(f"6. Access the application at https://{config['fqdn']}:{config['port']}")
 

config/templates/docker-compose-postgresql-caddy.yml.j2

@@ -126,3 +126,38 @@ services:
       tls /etc/caddy/cert.pem /etc/caddy/key.pem{% else %}
       tls internal{% endif %}{% endif %}
       }" > Caddyfile && caddy run'
+
+  {% if kafka_dispatcher.enabled %}
+  dispatcher:
+    container_name: dispatcher
+    image: ghcr.io/intuitem/ciso-assistant-community/dispatcher:latest
+    # build:
+    #   context: ../dispatcher
+    restart: always
+    environment:
+      - API_URL=http://backend:8000/api
+      - BOOTSTRAP_SERVERS={{ kafka_dispatcher.broker_url }}
+      - KAFKA_USE_AUTH={{ kafka_dispatcher.kafka_use_auth }}
+      {% if kafka_dispatcher.kafka_use_auth %}
+      - KAFKA_SASL_MECHANISM={{ kafka_dispatcher.kafka_sasl_mechanism }}
+      - KAFKA_USERNAME={{ kafka_dispatcher.kafka_username }}
+      - KAFKA_PASSWORD={{ kafka_dispatcher.kafka_password }}
+      {% endif %}
+      - OBSERVATION_TOPIC={{ kafka_dispatcher.observation_topic }}
+      - ERRORS_TOPIC={{ kafka_dispatcher.errors_topic }}
+      {% if kafka_dispatcher.authentication == 'credentials' %}
+      - USER_EMAIL={{ kafka_dispatcher.credentials.user_email }}
+      - USER_PASSWORD={{ kafka_dispatcher.credentials.user_password }}
+      - AUTO_RENEW_SESSION={{ kafka_dispatcher.auto_renew_session }}
+      {% elif kafka_dispatcher.authentication == 'token' %}
+      - TOKEN={{ kafka_dispatcher.token }}
+      {% endif %}
+      {% if kafka_dispatcher.s3_url %}
+      - S3_URL={{ kafka_dispatcher.s3_url }}
+      - S3_ACCESS_KEY={{ kafka_dispatcher.s3_access_key }}
+      - S3_SECRET_KEY={{ kafka_dispatcher.s3_secret_key }}
+      {% endif %}
+    depends_on:
+      backend:
+        condition: service_healthy
+  {% endif %}

config/templates/docker-compose-postgresql-traefik.yml.j2

@@ -150,3 +150,38 @@ services:
       - "--providers.file.directory=/etc/traefik"
       - "--providers.file.watch=true"{% else %}
       - "--experimental.localPlugins.selfsigned.moduleName=traefik.tls"{% endif %}
+
+  {% if kafka_dispatcher.enabled %}
+  dispatcher:
+    container_name: dispatcher
+    image: ghcr.io/intuitem/ciso-assistant-community/dispatcher:latest
+    # build:
+    #   context: ../dispatcher
+    restart: always
+    environment:
+      - API_URL=http://backend:8000/api
+      - BOOTSTRAP_SERVERS={{ kafka_dispatcher.broker_url }}
+      - KAFKA_USE_AUTH={{ kafka_dispatcher.kafka_use_auth }}
+      {% if kafka_dispatcher.kafka_use_auth %}
+      - KAFKA_SASL_MECHANISM={{ kafka_dispatcher.kafka_sasl_mechanism }}
+      - KAFKA_USERNAME={{ kafka_dispatcher.kafka_username }}
+      - KAFKA_PASSWORD={{ kafka_dispatcher.kafka_password }}
+      {% endif %}
+      - OBSERVATION_TOPIC={{ kafka_dispatcher.observation_topic }}
+      - ERRORS_TOPIC={{ kafka_dispatcher.errors_topic }}
+      {% if kafka_dispatcher.authentication == 'credentials' %}
+      - USER_EMAIL={{ kafka_dispatcher.credentials.user_email }}
+      - USER_PASSWORD={{ kafka_dispatcher.credentials.user_password }}
+      - AUTO_RENEW_SESSION={{ kafka_dispatcher.auto_renew_session }}
+      {% elif kafka_dispatcher.authentication == 'token' %}
+      - TOKEN={{ kafka_dispatcher.token }}
+      {% endif %}
+      {% if kafka_dispatcher.s3_url %}
+      - S3_URL={{ kafka_dispatcher.s3_url }}
+      - S3_ACCESS_KEY={{ kafka_dispatcher.s3_access_key }}
+      - S3_SECRET_KEY={{ kafka_dispatcher.s3_secret_key }}
+      {% endif %}
+    depends_on:
+      backend:
+        condition: service_healthy
+  {% endif %}

config/templates/docker-compose-sqlite-caddy.yml.j2

@@ -91,3 +91,38 @@ services:
       tls /etc/caddy/cert.pem /etc/caddy/key.pem{% else %}
       tls internal{% endif %}{% endif %}
       }" > Caddyfile && caddy run'
+
+  {% if kafka_dispatcher.enabled %}
+  dispatcher:
+    container_name: dispatcher
+    image: ghcr.io/intuitem/ciso-assistant-community/dispatcher:latest
+    # build:
+    #   context: ../dispatcher
+    restart: always
+    environment:
+      - API_URL=http://backend:8000/api
+      - BOOTSTRAP_SERVERS={{ kafka_dispatcher.broker_url }}
+      - KAFKA_USE_AUTH={{ kafka_dispatcher.kafka_use_auth }}
+      {% if kafka_dispatcher.kafka_use_auth %}
+      - KAFKA_SASL_MECHANISM={{ kafka_dispatcher.kafka_sasl_mechanism }}
+      - KAFKA_USERNAME={{ kafka_dispatcher.kafka_username }}
+      - KAFKA_PASSWORD={{ kafka_dispatcher.kafka_password }}
+      {% endif %}
+      - OBSERVATION_TOPIC={{ kafka_dispatcher.observation_topic }}
+      - ERRORS_TOPIC={{ kafka_dispatcher.errors_topic }}
+      {% if kafka_dispatcher.authentication == 'credentials' %}
+      - USER_EMAIL={{ kafka_dispatcher.credentials.user_email }}
+      - USER_PASSWORD={{ kafka_dispatcher.credentials.user_password }}
+      - AUTO_RENEW_SESSION={{ kafka_dispatcher.auto_renew_session }}
+      {% elif kafka_dispatcher.authentication == 'token' %}
+      - TOKEN={{ kafka_dispatcher.token }}
+      {% endif %}
+      {% if kafka_dispatcher.s3_url %}
+      - S3_URL={{ kafka_dispatcher.s3_url }}
+      - S3_ACCESS_KEY={{ kafka_dispatcher.s3_access_key }}
+      - S3_SECRET_KEY={{ kafka_dispatcher.s3_secret_key }}
+      {% endif %}
+    depends_on:
+      backend:
+        condition: service_healthy
+  {% endif %}

config/templates/docker-compose-sqlite-traefik.yml.j2

@@ -112,3 +112,38 @@ services:
       - "--providers.file.directory=/etc/traefik"
       - "--providers.file.watch=true"{% else %}
       - "--experimental.localPlugins.selfsigned.moduleName=traefik.tls"{% endif %}
+
+  {% if kafka_dispatcher.enabled %}
+  dispatcher:
+    container_name: dispatcher
+    image: ghcr.io/intuitem/ciso-assistant-community/dispatcher:latest
+    # build:
+    #   context: ../dispatcher
+    restart: always
+    environment:
+      - API_URL=http://backend:8000/api
+      - BOOTSTRAP_SERVERS={{ kafka_dispatcher.broker_url }}
+      - KAFKA_USE_AUTH={{ kafka_dispatcher.kafka_use_auth }}
+      {% if kafka_dispatcher.kafka_use_auth %}
+      - KAFKA_SASL_MECHANISM={{ kafka_dispatcher.kafka_sasl_mechanism }}
+      - KAFKA_USERNAME={{ kafka_dispatcher.kafka_username }}
+      - KAFKA_PASSWORD={{ kafka_dispatcher.kafka_password }}
+      {% endif %}
+      - OBSERVATION_TOPIC={{ kafka_dispatcher.observation_topic }}
+      - ERRORS_TOPIC={{ kafka_dispatcher.errors_topic }}
+      {% if kafka_dispatcher.authentication == 'credentials' %}
+      - USER_EMAIL={{ kafka_dispatcher.credentials.user_email }}
+      - USER_PASSWORD={{ kafka_dispatcher.credentials.user_password }}
+      - AUTO_RENEW_SESSION={{ kafka_dispatcher.auto_renew_session }}
+      {% elif kafka_dispatcher.authentication == 'token' %}
+      - TOKEN={{ kafka_dispatcher.token }}
+      {% endif %}
+      {% if kafka_dispatcher.s3_url %}
+      - S3_URL={{ kafka_dispatcher.s3_url }}
+      - S3_ACCESS_KEY={{ kafka_dispatcher.s3_access_key }}
+      - S3_SECRET_KEY={{ kafka_dispatcher.s3_secret_key }}
+      {% endif %}
+    depends_on:
+      backend:
+        condition: service_healthy
+  {% endif %}

dispatcher/.dockerignore

@@ -0,0 +1,20 @@
+*.DS_Store
+*~$*
+.env
+.venv
+venv
+**/node_modules/
+.vscode
+*.sqlite3
+django_secret_key
+temp/
+db/
+.dccache
+/backend/profiles
+./backend/ciso_assistant/.meta
+caddy_data/
+**/dist/
+**/.meta
+charts/custom-values.yaml
+**/charts/*/charts
+*.bak