Investigations

core/schemas/entity.py

@@ -21,6 +21,7 @@ class EntityType(str, Enum):
     company = "company"
     phone = "phone"
     note = "note"
+    investigation = "investigation"
 
 
 class Entity(BaseModel, database_arango.ArangoYetiConnector):
@@ -115,6 +116,12 @@ class Identity(Entity):
     sectors: list[str] = []
     contact_information: str = ""
 
+class Investigation(Entity):
+    _type_filter: ClassVar[str] = EntityType.investigation
+    type: Literal["investigation"] = EntityType.investigation
+
+    reference: str = ""
+
 
 TYPE_MAPPING: dict[str, "EntityClasses"] = {
     "threat-actor": ThreatActor,
@@ -125,10 +132,12 @@ class Identity(Entity):
     "campaign": Campaign,
     "entities": Entity,
     "entity": Entity,
+    "identity": Identity,
     "company": Company,
     "phone": Phone,
     "note": Note,
     "identity": Identity,
+    "investigation": Investigation,
 }
 REGEXES_ENTITIES = {
     EntityType.exploit: re.compile(
@@ -152,6 +161,7 @@ class Identity(Entity):
     | Company
     | Phone
     | Note
+    | Investigation
 )
 EntityClasses = (
     Type[ThreatActor]
@@ -164,4 +174,5 @@ class Identity(Entity):
     | Type[Company]
     | Type[Phone]
     | Type[Note]
+    | Type[Investigation]
 )

core/schemas/observable.py

@@ -1,17 +1,18 @@
 # TODO Observable value normalization
 
 import datetime
+import re
 from enum import Enum
-from typing import Literal, Optional, ClassVar
+from typing import ClassVar, Literal, Optional
 
-from pydantic import BaseModel, Field
 import validators
+from pydantic import BaseModel, Field
+
 from core import database_arango
-from core.helpers import refang
+from core.helpers import now, refang
 from core.schemas.entity import Entity
-from core.schemas.tag import DEFAULT_EXPIRATION_DAYS, Tag
 from core.schemas.graph import TagRelationship
-from core.helpers import now
+from core.schemas.tag import DEFAULT_EXPIRATION_DAYS, Tag
 
 
 # Data Schema
@@ -177,7 +178,13 @@ def delete_context(
     ObservableType.email: validators.email,
 }
 
-REGEXES_OBSERVABLES = {}
+REGEXES_OBSERVABLES = {
+    # Unix
+    ObservableType.path : [
+        re.compile(r"^(\/[^\/\0]+)+$"),
+        re.compile(r"^(?:[a-zA-Z]\:|\\\\[\w\.]+\\[\w.$]+)\\(?:[\w]+\\)*\w([\w.])+")
+    ]
+}
 
 
 def validate_observable(obs: Observable) -> bool:
@@ -193,9 +200,10 @@ def find_type(value: str) -> ObservableType | None:
     for obs_type in TYPE_VALIDATOR_MAP:
         if TYPE_VALIDATOR_MAP[obs_type](value):
             return obs_type
-    for type_obs, regex in REGEXES_OBSERVABLES.items():
-        if regex.match(value):
-            return type_obs
+    for obs_type, regexes in REGEXES_OBSERVABLES.items():
+        for regex in regexes:
+            if regex.match(value):
+                return obs_type
     return None
 
 

core/taskmanager.py

@@ -69,6 +69,7 @@
         "plugins.feeds.public.sslblacklist_ip",
         "plugins.feeds.public.threatfox",
         "plugins.feeds.public.threatview_c2",
+        "plugins.feeds.public.timesketch",
         "plugins.feeds.public.tor_exit_nodes",
         "plugins.feeds.public.urlhaus",
         "plugins.feeds.public.viriback_tracker",

plugins/feeds/public/timesketch.py

@@ -0,0 +1,82 @@
+from datetime import timedelta, datetime
+from time import sleep
+import logging
+
+from core.schemas import observable
+from core.schemas.observables import ipv4, hostname, url, sha1, md5, sha256, path
+from core.schemas.entity import Investigation
+from core.schemas import task
+from core import taskmanager
+from core.config.config import yeti_config
+
+from timesketch_api_client import client
+
+
+TIMESKETCH_TYPE_MAPPING = {
+    'ipv4': ipv4.IPv4,
+    'hostname': hostname.Hostname,
+    'hash_sha1': sha1.SHA1,
+    'hash_md5': md5.MD5,
+    'hash_sha256': sha256.SHA256,
+    'url': url.Url,
+    'fs_path': path.Path,
+}
+
+class Timesketch(task.FeedTask):
+
+    _defaults = {
+        "name": "Timesketch",
+        "frequency": timedelta(hours=1),
+        "type": "feed",
+        "description": "This feed creates Investigations from a Timesketch server.",
+    }
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._endpoint = yeti_config.get("timesketch", "endpoint")
+        if not self._endpoint:
+            logging.error("Timesketch cannot proceed without an endpoint.")
+            return
+        username = yeti_config.get("timesketch", "username")
+        password = yeti_config.get("timesketch", "password")
+
+        self._ts_client = client.TimesketchApi(self._endpoint, username, password)
+
+
+    def run(self):
+        if not self._endpoint:
+            logging.error("Timesketch cannot proceed without an endpoint.")
+            return
+
+        sketches = self._ts_client.list_sketches()
+
+        for sketch in sketches:
+            description = "# Timelines\n\n"
+            for timeline in sketch.list_timelines():
+                description += f"- {timeline.name}\n"
+
+            created_at = datetime.strptime(sketch.resource_data['objects'][0]['created_at'], "%Y-%m-%dT%H:%M:%S.%f")
+
+            investigation = Investigation(
+                name=sketch.name,
+                created=created_at,
+                reference=f'{self._endpoint}/sketch/{sketch.id}',
+                description=description,
+            ).save()
+            for intel in sketch.get_intelligence_attribute():
+                observable_type = TIMESKETCH_TYPE_MAPPING.get(intel['type'])
+                if observable_type:
+                    obs = observable_type(value=intel['ioc']).save()
+                else:
+                    try:
+                        obs = observable.Observable.add_text(intel['ioc'])
+                    except ValueError as error:
+                        logging.error('Error adding observable %s from Timesketch: %s', intel['ioc'], error)
+                        continue
+
+                obs.tag(intel['tags'])
+                obs.add_context("timesketch", {'sketch_link': intel['externalURI'], 'sketch_tags': intel['tags']})
+                obs.link_to(investigation, "seen in", f"Observable seen in {investigation.name}")
+
+
+taskmanager.TaskManager.register_task(Timesketch)