new feed TweetLive

plugins/feeds/public/tweetlive.py

@@ -0,0 +1,68 @@
+from datetime import timedelta
+from typing import ClassVar
+
+import pandas as pd
+
+from core import taskmanager
+from core.schemas import observable, task
+
+MAPPING = {
+    "domain": observable.ObservableType.hostname,
+    "ip": observable.ObservableType.ipv4,
+    "sha256": observable.ObservableType.sha256,
+    "url": observable.ObservableType.url,
+    "md5": observable.ObservableType.md5,
+}
+
+
+class TweetLive(task.FeedTask):
+    _defaults = {
+        "frequency": timedelta(hours=1),
+        "name": "TweetLive",
+        "description": "This feed contains tweets",
+    }
+    _SOURCE: ClassVar["str"] = "https://api.tweetfeed.live/v1/today"
+
+    def run(self):
+        r = self._make_request(self._SOURCE, sort=False)
+
+        if r:
+            data = r.json()
+
+            if not data:
+                raise ValueError("No data returned")
+
+            df = pd.DataFrame(data)
+            df.fillna("")
+            df["date"] = pd.to_datetime(df["date"])
+
+            df = self._filter_observables_by_time(df, "date")
+
+        for _, line in df.iterrows():
+            self.analyze(line)
+
+    def analyze(self, item):
+        if item["type"] not in MAPPING:
+            return
+
+        obs_type = MAPPING[item["type"]]
+
+        if not obs_type:
+            raise (ValueError(f"Observable type {item['type']} not supported"))
+
+        obs = observable.TYPE_MAPPING["observable"](
+            value=item["value"], type=obs_type
+        ).save()
+
+        context = {}
+
+        if "tweet" in item:
+            context["tweet"] = item["tweet"]
+        if "user" in item:
+            context["user"] = item["user"]
+
+        if item["tags"]:
+            obs.tag(item["tags"])
+
+        if context:
+            obs.add_context(self.name, context)

tests/feeds.py

@@ -12,6 +12,7 @@
     openphish,
     timesketch,
     tor_exit_nodes,
+    tweetlive,
 )
 
 
@@ -70,3 +71,8 @@ def test_tor_exit_nodes(self):
         defaults = tor_exit_nodes.TorExitNodes._defaults.copy()
         feed = tor_exit_nodes.TorExitNodes(**defaults)
         feed.run()
+
+    def test_tweetlive(self):
+        defaults = tweetlive.TweetLive._defaults.copy()
+        feed = tweetlive.TweetLive(**defaults)
+        feed.run()