add package

core/schemas/observable.py

@@ -1,6 +1,7 @@
 # TODO Observable value normalization
 
 import datetime
+import unicodedata
 import re
 from enum import Enum
 from typing import ClassVar, Literal, Optional
@@ -90,7 +91,11 @@ def add_text(cls, text: str, tags: list[str] = []) -> "Observable":
         return observable
 
     def tag(
-        self, tags: list[str], strict: bool = False, expiration_days: int | None = None
+        self,
+        tags: list[str],
+        strict: bool = False,
+        normalize: bool = True,
+        expiration_days: int | None = None,
     ) -> "Observable":
         """Connects observable to tag graph."""
         expiration_days = expiration_days or DEFAULT_EXPIRATION_DAYS
@@ -101,17 +106,25 @@ def tag(
         extra_tags = set()
         for tag_name in tags:
             # Attempt to find replacement tag
-            replacements, _ = Tag.filter({"in__replaces": [tag_name]}, count=1)
+            if normalize:
+                nfkd_form = unicodedata.normalize("NFKD", tag_name)
+                tag_normalized = "".join(
+                    [c for c in nfkd_form if not unicodedata.combining(c)]
+                )
+            else:
+                tag_normalized = tag_name
+
+            replacements, _ = Tag.filter({"in__replaces": [tag_normalized]}, count=1)
             tag: Optional[Tag] = None
 
             if replacements:
                 tag = replacements[0]
             # Attempt to find actual tag
             else:
-                tag = Tag.find(name=tag_name)
+                tag = Tag.find(name=tag_normalized)
             # Create tag
             if not tag:
-                tag = Tag(name=tag_name).save()
+                tag = Tag(name=tag_normalized).save()
 
             tag_link = self.observable_tag(tag.name)
             self.tags[tag.name] = tag_link

extras/scripts/import_misp_galaxies.py

@@ -0,0 +1,109 @@
+import argparse
+import sys
+from os import path
+
+from pymispgalaxies import Clusters, Galaxies
+
+from core.schemas.entity import ThreatActor, Malware
+from core import database_arango
+
+
+class ImportMisp:
+    def __init__(self) -> None:
+        self.clusters = Clusters()
+        self.db = database_arango.db
+
+    def add_ThreatActor(self):
+        cluster_threat_actors = self.clusters.get("threat-actor")
+        for name_ta, obj_dict in cluster_threat_actors.cluster_values.items():
+            name = None
+            description = ""
+            tags = []
+            aliases = []
+            ta = obj_dict.to_dict()
+            try:
+                if "description" in ta:
+                    description = ta["description"]
+                name = ta["value"]
+                if "meta" in ta:
+                    meta = ta["meta"].to_dict()
+                    if (
+                        "cfr-suspected-state-sponsor" in meta
+                        and meta["cfr-suspected-state-sponsor"]
+                    ):
+                        country = "Country Supected: {}.\n".format(
+                            meta["cfr-suspected-state-sponsor"]
+                        )
+                        description = "{} {}".format(description, country)
+                    if "cfr-target-category" in meta and meta["cfr-target-category"]:
+                        target = "Target: {}.\n".format(
+                            ",".join(meta["cfr-target-category"])
+                        )
+                        description = "{} {}".format(description, target)
+                    if (
+                        "cfr-suspected-victims" in meta
+                        and meta["cfr-suspected-victims"]
+                    ):
+                        victims = "Victims: {}\n.".format(
+                            ",".join(meta["cfr-suspected-victims"])
+                        )
+                        description = "{} {}".format(description, victims)
+                    if "synonyms" in meta:
+                        for synonym in meta["synonyms"]:
+                            tags.append(synonym)
+                    if "synonyms" in meta:
+                        aliases = meta["synonyms"]
+                    tags.append(name)
+                actor = ThreatActor(name=name)
+                actor.description = description
+                actor.relevant_tags = tags
+
+                if aliases:
+                    actor.aliases = aliases
+                actor.save()
+            except Exception as e:
+                print(f"Error: {e}")
+
+    def add_Malware(self, name_cluster="malpedia"):
+        cluster_malware = self.clusters.get(name_cluster)
+        for name, obj_dict in cluster_malware.cluster_values.items():
+            malware = obj_dict.to_dict()
+            name = None
+            description = ""
+            tags = []
+            try:
+                if "description" in malware:
+                    description = malware["description"]
+                name = malware["value"]
+                if "meta" in malware:
+                    meta = malware["meta"].to_dict()
+                    if "synonyms" in meta:
+                        for synonym in meta["synonyms"]:
+                            tags.append(synonym)
+                    tags.append(name)
+                malware_new = Malware(name=name)
+                malware_new.description = description
+                malware_new.relevant_tags = tags
+                malware_new.save()
+            except Exception as e:
+                print(f"Error: {e}")
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description="MISP Galaxies")
+    parser.add_argument(
+        "-t",
+        "--type",
+        help="Type of galaxies to retrieve",
+        choices=["all", "TA", "Malware"],
+        default="all",
+    )
+    return parser.parse_args()
+
+
+if __name__ == "__main__":
+    import_misp = ImportMisp()
+    args = parse_args()
+    if args.type == "all":
+        import_misp.add_ThreatActor()
+        import_misp.add_Malware()

plugins/feeds/public/misp.py

@@ -190,7 +190,7 @@ def _add_tag(self, obs: observable.Observable, instance: dict, attribute: dict):
         nfkd_form = unicodedata.normalize("NFKD", instance_name)
         instance_name = "".join([c for c in nfkd_form if not unicodedata.combining(c)])
         tag = f"{instance_name}:{attribute['event_id']}"
-        obs.tag(tag)
+        obs.tag(tag,normalize=False)
 
 
 taskmanager.TaskManager.register_task(MispFeed)

pyproject.toml

@@ -29,8 +29,10 @@ httpx = "^0.23.3"
 pymisp = "^2.4.176"
 otxv2 = "^1.5.12"
 shodan = "^1.30.0"
+pymispgalaxies = {git = "https://github.com/MISP/PyMISPGalaxies.git"}
 timesketch-api-client = "^20230721"
 
+
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"

tests/schemas/fixture.py

@@ -5,7 +5,13 @@
 from core import database_arango
 from core.schemas.observable import Observable
 from core.schemas.observables import ipv4, hostname, url
+
+
+from core.schemas.entity import ThreatActor,Campaign,Malware
+
+
 from core.schemas.entity import ThreatActor, Malware, Investigation
+
 from core.schemas.tag import Tag
 from core.schemas.indicator import Regex, DiamondModel, Query, QueryType
 from core.schemas.template import Template
@@ -38,6 +44,9 @@ def test_something(self):
         hacker.tag(["hacker"])
         ta = ThreatActor(name="HackerActor", relevant_tags=["hacker_sus"]).save()
         ta.link_to(hacker, "c2", "C2 infrastructure")
+        campaign = Campaign(name="HackerCampaign").save()
+        campaign.link_to(ta, "actor", "Actor")
+        campaign.relevant_tags = ["hacker_sus"]
         www_hacker.tag(["web", "hacker"])
         c2_hacker.tag(["web", "hacker"])
         sus_hacker = hostname.Hostname(value="sus.hacker.com").save()