add: transform models to MISP format

Author: righel

api/app/models/attribute.py

@@ -1,10 +1,16 @@
 import uuid
 
+import logging
 from app.database import Base
 from app.models.event import DistributionLevel
 from sqlalchemy import BigInteger, Boolean, Column, Enum, ForeignKey, Integer, String
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Mapped, mapped_column, relationship
+from app.services.minio import get_minio_client
+from app.settings import Settings, get_settings
+
+
+logger = logging.getLogger(__name__)
 
 
 class Attribute(Base):
@@ -37,3 +43,45 @@ class Attribute(Base):
     last_seen = Column(BigInteger(), index=True)
 
     tags = relationship("Tag", secondary="attribute_tags", lazy="subquery")
+
+    def to_misp_format(
+        self,
+        settings: Settings = get_settings(),
+    ):
+        """Convert the Attribute to a MISP-compatible dictionary representation."""
+
+        attr_json = {
+            "id": self.id,
+            "event_id": self.event_id,
+            "object_id": self.object_id,
+            "object_relation": self.object_relation,
+            "category": self.category,
+            "type": self.type,
+            "value": self.value,
+            "to_ids": self.to_ids,
+            "uuid": str(self.uuid),
+            "timestamp": self.timestamp,
+            "distribution": self.distribution.name if self.distribution else None,
+            "sharing_group_id": self.sharing_group_id,
+            "comment": self.comment,
+            "deleted": self.deleted,
+            "disable_correlation": self.disable_correlation,
+            "first_seen": self.first_seen,
+            "last_seen": self.last_seen,
+            "Tags": [tag.to_misp_format() for tag in self.tags],
+        }
+
+        # if its a file attribute, we need to handle it differently
+        if self.type in ["malware-sample", "attachment"]:
+            try:
+                MinioClient = get_minio_client()
+                data = MinioClient.get_object(settings.Storage.minio.bucket, self.uuid)
+                file_content = data.read()
+                file_b64 = file_content.encode("base64")
+                attr_json["data"] = file_b64
+            except Exception as e:
+                file_b64 = None
+                logger.error(f"Error storing attachment: {str(e)}")
+                print(f"Error fetching file from storage: {str(e)}")
+
+        return attr_json

api/app/models/event.py

@@ -92,7 +92,9 @@ class Event(Base):
     tags = relationship("Tag", secondary="event_tags", lazy="joined")
     organisation = relationship("Organisation", lazy="joined", uselist=False, foreign_keys=[org_id])
 
-    def to_misp_event(self):
+    def to_misp_format(self):
+        """Convert the Event to a MISP-compatible dictionary representation."""
+
         return {
             "id": self.id,
             "org_id": self.org_id,
@@ -114,11 +116,12 @@ def to_misp_event(self):
             "extends_uuid": str(self.extends_uuid) if self.extends_uuid else None,
             "protected": self.protected,
             "deleted": self.deleted,
-            # "Attributes": [attribute.to_misp_attribute() for attribute in self.attributes],
-            # "Objects": [obj.to_misp_object() for obj in self.objects],
-            # "Tags": [tag.name for tag in self.tags],
+            "Attributes": [attribute.to_misp_format() for attribute in self.attributes],
+            "Objects": [obj.to_misp_format() for obj in self.objects],
+            "Tags": [tag.to_misp_format() for tag in self.tags],
             "Organisation": {
                 "id": self.organisation.id,
-                "name": self.organisation.name
+                "name": self.organisation.name,
+                "uuid": str(self.organisation.uuid),
             } if self.organisation else None
         }

api/app/models/object.py

@@ -32,3 +32,25 @@ class Object(Base):
 
     attributes = relationship("Attribute", lazy="subquery", cascade="all, delete-orphan")
     object_references = relationship("ObjectReference", lazy="subquery", cascade="all, delete-orphan")
+
+    def to_misp_format(self):
+        """Convert the Object to a MISP-compatible dictionary representation."""
+        return {
+            "id": self.id,
+            "name": self.name,
+            "meta_category": self.meta_category,
+            "description": self.description,
+            "template_uuid": self.template_uuid,
+            "template_version": self.template_version,
+            "event_id": self.event_id,
+            "uuid": str(self.uuid),
+            "timestamp": self.timestamp,
+            "distribution": self.distribution.name if self.distribution else None,
+            "sharing_group_id": self.sharing_group_id,
+            "comment": self.comment,
+            "deleted": self.deleted,
+            "first_seen": self.first_seen,
+            "last_seen": self.last_seen,
+            "Attributes": [attribute.to_misp_format() for attribute in self.attributes],
+            "ObjectReference": [ref.to_misp_format() for ref in self.object_references],
+        }

api/app/models/object_reference.py

@@ -33,3 +33,20 @@ class ObjectReference(Base):
     relationship_type = Column(String)
     comment = Column(String, nullable=False)
     deleted = Column(Boolean, nullable=False, default=False)
+
+    def to_misp_format(self):
+        """Convert the ObjectReference to a MISP-compatible dictionary representation."""
+        return {
+            "id": self.id,
+            "uuid": str(self.uuid),
+            "timestamp": self.timestamp,
+            "object_id": self.object_id,
+            "event_id": self.event_id,
+            "source_uuid": str(self.source_uuid) if self.source_uuid else None,
+            "referenced_uuid": str(self.referenced_uuid) if self.referenced_uuid else None,
+            "referenced_id": self.referenced_id,
+            "referenced_type": self.referenced_type.name if self.referenced_type else None,
+            "relationship_type": self.relationship_type,
+            "comment": self.comment,
+            "deleted": self.deleted,
+        }

api/app/models/tag.py

@@ -18,6 +18,22 @@ class Tag(Base):
     is_custom_galaxy = Column(Boolean, nullable=False, default=False)
     local_only = Column(Boolean, nullable=False, default=False)
 
+    def to_misp_format(self):
+        """Convert the Tag to a MISP-compatible dictionary representation."""
+        return {
+            "id": self.id,
+            "name": self.name,
+            "colour": self.colour,
+            "exportable": self.exportable,
+            "org_id": self.org_id,
+            "user_id": self.user_id,
+            "hide_tag": self.hide_tag,
+            "numerical_value": self.numerical_value,
+            "is_galaxy": self.is_galaxy,
+            "is_custom_galaxy": self.is_custom_galaxy,
+            "local_only": self.local_only,
+        }
+
 
 class EventTag(Base):
     __tablename__ = "event_tags"

api/app/repositories/servers.py

@@ -723,7 +723,7 @@ def push_event_by_uuid(
 
     try:
         response = remote_misp._prepare_request(
-            "POST", f"events/add/{event_uuid}", data=db_event.to_misp_event()
+            "POST", f"events/add/{event_uuid}", data=db_event.to_misp_format()
         )
         if response.status_code == 200:
             return True

api/app/routers/events.py

@@ -198,16 +198,20 @@ def untag_event(
     status_code=status.HTTP_200_OK,
 )
 async def upload_attachments(
-    event_id: int,
+    event_id: Union[int, UUID],
     attachments: list[UploadFile],
     attachments_meta: Annotated[str, Form()],
     db: Session = Depends(get_db),
     user: user_schemas.User = Security(
         get_current_active_user, scopes=["events:update"]
     ),
 ) -> list[object_schemas.Object]:
-    event = events_repository.get_event_by_id(db, event_id=event_id)
-    if event is None:
+    if isinstance(event_id, int):
+        db_event = events_repository.get_event_by_id(db, event_id=event_id)
+    else:
+        db_event = events_repository.get_event_by_uuid(db, event_uuid=event_id)
+
+    if db_event is None:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND, detail="Event not found"
         )
@@ -216,9 +220,9 @@ async def upload_attachments(
         attachments_meta = json.loads(attachments_meta)
 
     objects = attachments_repository.upload_attachments_to_event(
-        db=db, event=event, attachments=attachments, attachments_meta=attachments_meta
+        db=db, event=db_event, attachments=attachments, attachments_meta=attachments_meta
     )
-    tasks.index_event.delay(event.uuid)
+    tasks.index_event.delay(db_event.uuid)
 
     return objects
 

frontend/src/views/Login.vue

@@ -95,7 +95,6 @@ function onSubmit(values, { setErrors }) {
       <button class="w-100 btn btn-lg btn-outline-primary" type="submit">
         Sign in
       </button>
-      <p class="mt-3 mb-3 text-muted">&copy; 2024</p>
     </Form>
   </div>
 </template>