Merge pull request #347 from MISP/fix-misp-format-export

fix: exports misp format, remove uuids, set distribution level

Author: righel

api/alembic/versions/m6n7o8p9q0r1_add_export_distribution.py

@@ -0,0 +1,26 @@
+"""add export distribution
+
+Revision ID: m6n7o8p9q0r1
+Revises: l5m6n7o8p9q0
+Create Date: 2026-06-12 00:00:00.000000
+
+MISP-format exports build a single MISP event; the event distribution level is
+captured per export.
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "m6n7o8p9q0r1"
+down_revision = "l5m6n7o8p9q0"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column("exports", sa.Column("distribution", sa.Integer(), nullable=True))
+
+
+def downgrade():
+    op.drop_column("exports", "distribution")

api/app/models/export.py

@@ -31,6 +31,8 @@ class Export(Base):
     query = Column(Text, nullable=False)
     index_target = Column(String(50), nullable=False, default="attributes")
     format = Column(String(20), nullable=False, default="json")
+    # Event distribution level for MISP-format exports (0–4); null otherwise.
+    distribution = Column(Integer, nullable=True)
     status = Column(String(50), nullable=False, default="queued")
     storage_key = Column(String(512), nullable=True)
     file_size = Column(Integer, nullable=True)

api/app/repositories/exports.py

@@ -159,6 +159,20 @@ def update_export_schedule(
     return db_export
 
 
+def requeue_export(
+    db: Session, export_id: int, user_id: int
+) -> export_models.Export:
+    """Reset an existing export to ``queued`` so it can be re-run in place."""
+    db_export = get_export_by_id(db, export_id, user_id)
+    if db_export is None:
+        return None
+    db_export.status = "queued"
+    db_export.error = None
+    db.commit()
+    db.refresh(db_export)
+    return db_export
+
+
 def set_celery_task_id(db: Session, export_id: int, task_id: str) -> None:
     db_export = (
         db.query(export_models.Export)
@@ -211,7 +225,13 @@ def _fetch_hits(index: str, query: str) -> list[dict]:
     return hits
 
 
-def _to_misp_json(db: Session, hits: list[dict], index_target: str, name: str):
+def _to_misp_json(
+    db: Session,
+    hits: list[dict],
+    index_target: str,
+    name: str,
+    distribution: int = None,
+):
     """Serialize all matching attributes into a single MISP event.
 
     Every matching attribute is collected into one synthetic MISP event named
@@ -221,7 +241,6 @@ def _to_misp_json(db: Session, hits: list[dict], index_target: str, name: str):
     output matches MISP's event API shape (``{"Event": {...}}``).
     """
     import json
-    import uuid as uuid_module
 
     from app.schemas import attribute as attribute_schemas
     from app.schemas import event as event_schemas
@@ -248,19 +267,31 @@ def _to_misp_json(db: Session, hits: list[dict], index_target: str, name: str):
             except Exception as e:  # skip malformed rows
                 logger.warning("Skipping attribute in MISP export: %s", e)
 
-    # Stable UUID per export so re-runs (e.g. scheduled) update the same event
-    # on re-import rather than creating a new one each time.
-    event_uuid = uuid_module.uuid5(
-        uuid_module.NAMESPACE_URL, f"misp-workbench:export:{name}"
-    )
     misp_event = event_schemas.Event(
         info=name,
-        uuid=event_uuid,
         timestamp=int(datetime.now(timezone.utc).timestamp()),
+        distribution=distribution,
+        disable_correlation=False,
         attributes=attributes,
     )
 
     payload = misp_event.to_misp_format()
+
+    # Strip identifiers so importing the file always creates fresh records
+    # rather than colliding with existing event/attribute uuids.
+    def _strip_ids(obj: dict) -> None:
+        obj.pop("uuid", None)
+        obj.pop("id", None)
+
+    event = payload.get("Event", {})
+    _strip_ids(event)
+    for attribute in event.get("Attribute", []):
+        _strip_ids(attribute)
+    for misp_object in event.get("Object", []):
+        _strip_ids(misp_object)
+        for attribute in misp_object.get("Attribute", []):
+            _strip_ids(attribute)
+
     content = json.dumps(payload, default=str, indent=2).encode("utf-8")
     return content, "json", "application/json", len(attributes)
 
@@ -293,7 +324,11 @@ def run_export(db: Session, export_id: int) -> None:
             # MISP format merges all matches into one event via the server-push
             # serializer; record_count reflects the attributes in that event.
             content, extension, _content_type, record_count = _to_misp_json(
-                db, hits, db_export.index_target, db_export.name
+                db,
+                hits,
+                db_export.index_target,
+                db_export.name,
+                db_export.distribution,
             )
         else:
             content, extension, _content_type = converters.convert(

api/app/routers/exports.py

@@ -61,6 +61,27 @@ async def create_export(
     return db_export
 
 
+@router.post("/exports/{export_id}/run", response_model=export_schemas.Export)
+async def run_export_now(
+    export_id: int,
+    db: Session = Depends(get_db),
+    user: user_schemas.User = Security(
+        get_current_active_user, scopes=["exports:create"]
+    ),
+):
+    db_export = exports_repository.requeue_export(
+        db, export_id=export_id, user_id=user.id
+    )
+    if db_export is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Export not found"
+        )
+    result = tasks.run_export.delay(db_export.id)
+    exports_repository.set_celery_task_id(db, db_export.id, result.id)
+    db.refresh(db_export)
+    return db_export
+
+
 @router.get("/exports/{export_id}", response_model=export_schemas.Export)
 async def get_export_by_id(
     export_id: int,

api/app/schemas/export.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 from typing import Literal, Optional
 
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, model_validator
 
 from app.schemas.task import ScheduleTaskSchedule
 
@@ -19,12 +19,20 @@ class ExportBase(BaseModel):
     query: str
     index_target: ExportIndexTarget = "attributes"
     format: ExportFormat = "json"
+    # Event distribution level (0–4) — required for MISP-format exports.
+    distribution: Optional[int] = None
 
 
 class ExportCreate(ExportBase):
     schedule: Optional[ScheduleTaskSchedule] = None
     schedule_enabled: bool = False
 
+    @model_validator(mode="after")
+    def _require_distribution_for_misp(self):
+        if self.format == "misp" and self.distribution is None:
+            raise ValueError("distribution is required for MISP-format exports")
+        return self
+
 
 class ExportScheduleUpdate(BaseModel):
     # ``schedule=None`` clears the schedule (unschedule). ``schedule_enabled``

api/app/tests/api/test_exports.py

@@ -391,6 +391,11 @@ def test_run_export_misp_format(
             assert len(event["Attribute"]) == 2
             values = {a["value"] for a in event["Attribute"]}
             assert values == {"1.2.3.4", "5.6.7.8"}
+            # uuid/id are stripped from the event and every attribute so
+            # importing creates fresh records.
+            assert "uuid" not in event and "id" not in event
+            for attr in event["Attribute"]:
+                assert "uuid" not in attr and "id" not in attr
         finally:
             db.delete(export)
             db.commit()

docs/features/api/openapi.json

@@ -8867,6 +8867,55 @@
         }
       }
     },
+    "/exports/{export_id}/run": {
+      "post": {
+        "tags": [
+          "Exports"
+        ],
+        "summary": "Run Export Now",
+        "operationId": "run_export_now_exports__export_id__run_post",
+        "security": [
+          {
+            "OAuth2PasswordBearer": [
+              "exports:create"
+            ]
+          }
+        ],
+        "parameters": [
+          {
+            "name": "export_id",
+            "in": "path",
+            "required": true,
+            "schema": {
+              "type": "integer",
+              "title": "Export Id"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Successful Response",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/Export"
+                }
+              }
+            }
+          },
+          "422": {
+            "description": "Validation Error",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/HTTPValidationError"
+                }
+              }
+            }
+          }
+        }
+      }
+    },
     "/exports/{export_id}": {
       "get": {
         "tags": [
@@ -13327,6 +13376,17 @@
             "title": "Format",
             "default": "json"
           },
+          "distribution": {
+            "anyOf": [
+              {
+                "type": "integer"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Distribution"
+          },
           "id": {
             "type": "integer",
             "title": "Id"
@@ -13509,6 +13569,17 @@
             "title": "Format",
             "default": "json"
           },
+          "distribution": {
+            "anyOf": [
+              {
+                "type": "integer"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Distribution"
+          },
           "schedule": {
             "anyOf": [
               {

docs/features/exports.md

@@ -27,9 +27,17 @@ can always fetch the same URL for the latest results.
 |---|---|---|
 | `json` | Raw OpenSearch `_source` documents | Passthrough dump |
 | `csv` | Flattened rows of the most useful fields | Tags joined with `|` |
-| `misp` | MISP schema compatible JSON format | [MISP schema](https://github.com/MISP/misp-rfc) |
+| `misp` | A single [MISP-schema](https://github.com/MISP/misp-rfc) event | All matches are merged into one event named after the export |
 | `stix` | A STIX 2.1 bundle | Attributes are grouped into events and converted via the [misp-stix](https://github.com/MISP/misp-stix) library |
 
+!!! note "MISP format"
+    The `misp` format collects every matching attribute into a **single MISP
+    event** (named after the export), even when the attributes come from
+    different misp-workbench events. It requires a **distribution** level and
+    keeps correlation enabled. The `uuid` and `id` fields are stripped from the
+    event and its attributes, so importing the file always creates fresh
+    records rather than colliding with existing ones.
+
 !!! note "STIX limits"
     STIX 2.1 conversion is CPU-intensive, so STIX exports are capped at
     **10,000 records** — narrow the query (or use JSON/CSV) for larger result
@@ -50,6 +58,9 @@ can always fetch the same URL for the latest results.
    polls until it settles to `completed` or `failed`.
 4. Once `completed`, click ***Download*** to fetch the file.
 
+Use ***Run now*** on any row to re-run an export on demand — it re-executes the
+stored query and overwrites the existing file in place.
+
 Alternatively, from the ***explore*** view use the ***Save as export…*** option
 in the Download menu — it opens the export dialog pre-filled with the current
 query.
@@ -156,6 +167,7 @@ Required scopes: `exports:create`
 |---|---|---|---|
 | `GET` | `/exports/` | List all exports (paginated) | `exports:read` |
 | `POST` | `/exports/` | Create an export (runs immediately) | `exports:create` |
+| `POST` | `/exports/{id}/run` | Re-run an export in place (overwrites its file) | `exports:create` |
 | `GET` | `/exports/{id}` | Get an export | `exports:read` |
 | `PATCH` | `/exports/{id}/schedule` | Set, pause/resume, or clear the schedule | `exports:create` |
 | `GET` | `/exports/{id}/download` | Download the stored artifact | `exports:read` |