feedly · Mathieu4141 · Aug 23, 2022 · Aug 16, 2022 · Aug 23, 2022 · Aug 23, 2022
diff --git a/...indicators_of_compromise_from_a_stream.py → ...rs_of_compromise_from_a_stream_as_stix.py b/...indicators_of_compromise_from_a_stream.py → ...rs_of_compromise_from_a_stream_as_stix.py
@@ -3,33 +3,36 @@
 from pathlib import Path
 from pprint import pprint
 
-from feedly.api_client.enterprise.indicators_of_compromise import IoCDownloader
+from feedly.api_client.enterprise.indicators_of_compromise import IoCDownloaderBuilder, IoCFormat
 from feedly.api_client.session import FeedlySession
 from feedly.api_client.utils import run_example
 
 RESULTS_DIR = Path(__file__).parent / "results"
 RESULTS_DIR.mkdir(exist_ok=True)
 
 
-def example_export_indicators_of_compromise_from_all_enterprise_feeds():
+def example_export_indicators_of_compromise_from_all_enterprise_feeds_as_stix():
     """
     This example will save a STIX 2.1 bundle containing the contextualized IoCs that Leo extracted during the past 12
      hours in all your enterprise feeds. 
     """
     # Authenticate using the default auth directory
     session = FeedlySession()
 
-    # Create the IoC fetcher object, and limit it to 12 hours
+    # Create the STIX IoC downloader builder object, and limit it to 12 hours
     # Usually newer_than will be the datetime of the last fetch
-    downloader = IoCDownloader(session=session, newer_than=datetime.now() - timedelta(hours=12))
+    downloader_builder = IoCDownloaderBuilder(
+        session=session, newer_than=datetime.now() - timedelta(hours=12), format=IoCFormat.STIX
+    )
 
     # Fetch the IoC from all the enterprise categories, and create a bundle containing them
     # You can use a different method to get the iocs from you personal categories, personal or enterprise boards,
     #  or from specific categories/boards using their names or ids
-    iocs_bundle = downloader.from_all_enterprise_categories()
+    downloader = downloader_builder.from_all_enterprise_categories()
+    iocs_bundle = downloader.download_all()
 
     # Save the bundle in a file
-    with (RESULTS_DIR / "ioc_example.json").open("w") as f:
+    with (RESULTS_DIR / "ioc_example_stix.json").open("w") as f:
         json.dump(iocs_bundle, f, indent=2)
 
     # Console display
@@ -39,4 +42,4 @@ def example_export_indicators_of_compromise_from_all_enterprise_feeds():
 if __name__ == "__main__":
     # Will prompt for the token if missing, and launch the example above
     # If a token expired error is raised, will prompt for a new token and restart the example
-    run_example(example_export_indicators_of_compromise_from_all_enterprise_feeds)
+    run_example(example_export_indicators_of_compromise_from_all_enterprise_feeds_as_stix)
diff --git a/examples/enterprise/export_indicators_of_compromise_to_misp.py b/examples/enterprise/export_indicators_of_compromise_to_misp.py
@@ -0,0 +1,46 @@
+import logging
+from datetime import datetime, timedelta
+from warnings import filterwarnings
+
+from feedly.api_client.enterprise.indicators_of_compromise import IoCDownloaderBuilder, IoCFormat
+from feedly.api_client.enterprise.misp_exporter import MispExporter
+from feedly.api_client.session import FeedlySession
+from feedly.api_client.utils import run_example
+
+# Paste your MISP key and URL below
+MISP_KEY = ""
+MISP_URL = ""
+
+assert MISP_KEY, "Please paste your MISP key"
+assert MISP_URL, "Please paste MISP url"
+
+
+def export_indicators_of_compromise_to_misp():
+    """
+    This example will export to your MISP instance the contextualized IoCs that Leo extracted during the past 6 hours
+     in all your enterprise feeds.
+    """
+    # Authenticate using the default auth directory
+    session = FeedlySession()
+
+    # Create the MISP IoC downloader builder object, and limit it to 6 hours
+    # Usually newer_than will be the datetime of the last fetch
+    downloader_builder = IoCDownloaderBuilder(
+        session=session, newer_than=datetime.now() - timedelta(hours=6), format=IoCFormat.MISP
+    )
+
+    # Fetch the IoC from all the enterprise categories, and feed them to the exporter
+    # You can use a different method to get the iocs from you personal categories, personal or enterprise boards,
+    #  or from specific categories/boards using their names or ids
+    downloader = downloader_builder.from_all_enterprise_categories()
+    exporter = MispExporter(MISP_URL, MISP_KEY, ignore_errors=True, verify_certificate=False)
+    exporter.send_bundles(downloader.stream_bundles())
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level="INFO")
+    filterwarnings("ignore")
+
+    # Will prompt for the token if missing, and launch the example above
+    # If a token expired error is raised, will prompt for a new token and restart the example
+    run_example(export_indicators_of_compromise_to_misp)
diff --git a/feedly/api_client/enterprise/indicators_of_compromise.py b/feedly/api_client/enterprise/indicators_of_compromise.py
@@ -1,16 +1,29 @@
 import uuid
+from abc import ABC, abstractmethod
 from datetime import datetime
-from typing import Dict, List, Optional
+from enum import Enum
+from itertools import chain
+from typing import ClassVar, Dict, Generic, Iterable, List, Optional, TypeVar
 from urllib.parse import parse_qs
 
+from requests import Response
+
 from feedly.api_client.data import Streamable
 from feedly.api_client.session import FeedlySession
 
+T = TypeVar("T")
+
+
+class IoCFormat(Enum):
+    MISP = "misp"
+    STIX = "stix2.1"
+
 
-class IoCDownloader:
+class IoCDownloaderABC(ABC, Generic[T]):
     RELATIVE_URL = "/v3/enterprise/ioc"
+    FORMAT: ClassVar[str]
 
-    def __init__(self, session: FeedlySession, newer_than: Optional[datetime] = None):
+    def __init__(self, session: FeedlySession, newer_than: Optional[datetime], stream_id: str):
         """
         Use this class to export the contextualized IoCs from a stream.
         Enterprise/personals feeds/boards are supported (see dedicated methods below).
@@ -24,50 +37,93 @@ def __init__(self, session: FeedlySession, newer_than: Optional[datetime] = None
         """
         self.newer_than = newer_than
         self.session = session
+        self.stream_id = stream_id
+
+    def download_all(self) -> List[T]:
+        return self._merge(self.stream_bundles())
+
+    def stream_bundles(self) -> Iterable[T]:
+        continuation = None
+        while True:
+            resp = self.session.make_api_request(
+                f"{self.RELATIVE_URL}",
+                params={
+                    "newerThan": int(self.newer_than.timestamp()) if self.newer_than else None,
+                    "continuation": continuation,
+                    "streamId": self.stream_id,
+                    "format": self.FORMAT,
+                },
+            )
+            yield self._parse_response(resp)
+            if not self.newer_than or "link" not in resp.headers:
+                return
+            next_url = resp.headers["link"][1:].split(">")[0]
+            continuation = parse_qs(next_url)["continuation"][0]
+
+    def _parse_response(self, resp: Response) -> T:
+        return resp.json()
+
+    @abstractmethod
+    def _merge(self, resp_jsons: Iterable[T]) -> T:
+        ...
+
+
+class IoCDownloaderBuilder:
+    def __init__(self, session: FeedlySession, format: IoCFormat, newer_than: Optional[datetime] = None):
+        """
+        Use this class to export the contextualized IoCs from a stream.
+        Enterprise/personals feeds/boards are supported (see dedicated methods below).
+        The IoCs will be returned along with their context and relationships in a dictionary representing a valid
+         STIX v2.1 Bundle object. https://docs.oasis-open.org/cti/stix/v2.1/os/stix-v2.1-os.html#_gms872kuzdmg
+        Use the newer_than parameter to filter articles that are newer than your last call.
+
+        :param session: The authenticated session to use to make the api calls
+        :param newer_than: Only articles newer than this parameter will be used. If None only one call will be make,
+         and the continuation will be ignored
+        """
+        self.session = session
+        self.format = format
+        self.newer_than = newer_than
+
         self.session.api_host = "https://cloud.feedly.com"
         self.user = self.session.user
 
-    def from_all_enterprise_categories(self) -> Dict:
+    def from_all_enterprise_categories(self) -> IoCDownloaderABC:
         return self.from_stream(self.user.get_all_enterprise_categories_stream())
 
-    def from_all_user_categories(self) -> Dict:
+    def from_all_user_categories(self) -> IoCDownloaderABC:
         return self.from_stream(self.user.get_all_user_categories_stream())
 
-    def from_enterprise_category(self, name_or_id: str) -> Dict:
+    def from_enterprise_category(self, name_or_id: str) -> IoCDownloaderABC:
         return self.from_stream(self.user.enterprise_categories.get(name_or_id))
 
-    def from_enterprise_tag(self, name_or_id: str) -> Dict:
+    def from_enterprise_tag(self, name_or_id: str) -> IoCDownloaderABC:
         return self.from_stream(self.user.enterprise_tags.get(name_or_id))
 
-    def from_user_category(self, name_or_id: str) -> Dict:
+    def from_user_category(self, name_or_id: str) -> IoCDownloaderABC:
         return self.from_stream(self.user.user_categories.get(name_or_id))
 
-    def from_stream(self, stream: Streamable) -> Dict:
+    def from_stream(self, stream: Streamable) -> IoCDownloaderABC:
         return self.from_stream_id(stream.id)
 
-    def from_stream_id(self, stream_id: str) -> Dict:
+    def from_stream_id(self, stream_id: str) -> IoCDownloaderABC:
+        format2class = {IoCFormat.MISP: MispIoCDownloader, IoCFormat.STIX: StixIoCDownloader}
+        return format2class[self.format](session=self.session, newer_than=self.newer_than, stream_id=stream_id)
+
+
+class StixIoCDownloader(IoCDownloaderABC[Dict]):
+    FORMAT = "stix2.1"
+
+    def _merge(self, resp_jsons: List[Dict]) -> Dict:
         return {
-            "objects": self._download_ioc_objects(stream_id=stream_id),
+            "objects": list(chain.from_iterable(resp_json["objects"] for resp_json in resp_jsons)),
             "id": f"bundle--{str(uuid.uuid4())}",
             "type": "bundle",
         }
 
-    def _download_ioc_objects(self, stream_id: str) -> List[Dict]:
-        objects = []
-        continuation = None
-        while True:
-            resp = self.session.make_api_request(
-                f"{self.RELATIVE_URL}",
-                params={
-                    "newerThan": int(self.newer_than.timestamp()) if self.newer_than else None,
-                    "continuation": continuation,
-                    "streamId": stream_id,
-                },
-            )
-            objects += resp.json()["objects"]
-            if not self.newer_than:
-                return objects
-            if "link" not in resp.headers:
-                return objects
-            next_url = resp.headers["link"][1:].split(">")[0]
-            continuation = parse_qs(next_url)["continuation"][0]
+
+class MispIoCDownloader(IoCDownloaderABC[Dict]):
+    FORMAT = "misp"
+
+    def _merge(self, resp_jsons: Iterable[Dict]) -> Dict:
+        return {"response": list(chain.from_iterable(resp_json["response"] for resp_json in resp_jsons))}
diff --git a/feedly/api_client/enterprise/misp_exporter.py b/feedly/api_client/enterprise/misp_exporter.py
@@ -0,0 +1,36 @@
+import json
+import logging
+from typing import Iterable
+
+import requests
+
+
+class MispExporter:
+    def __init__(self, url: str, key: str, ignore_errors: bool = False, verify_certificate: bool = True):
+        self.url = url.rstrip("/")
+        self.key = key
+        self.ignore_errors = ignore_errors
+        self.verify_certificate = verify_certificate
+
+    def send_bundles(self, bundles: Iterable[dict]) -> None:
+        self.send_events(event["Event"] for bundle in bundles for event in bundle["response"])
+
+    def send_events(self, events: Iterable[dict]) -> None:
+        for event in events:
+            self.send_event(event)
+
+    def send_event(self, event: dict) -> None:
+        try:
+            resp = requests.post(
+                f"{self.url}/events/add",
+                headers={"Authorization": self.key, "Accept": f"application/json", "content-type": f"application/json"},
+                data=json.dumps(event),
+                verify=self.verify_certificate,
+            )
+            resp.raise_for_status()
+            logging.info(f"{self.url}/events/view/{resp.json()['Event']['id']}")
+        except:
+            if self.ignore_errors:
+                logging.exception(f"Failed to send event {event}")
+                return
+            raise
diff --git a/setup.py b/setup.py
@@ -18,7 +18,7 @@
 EMAIL = "ml@feedly.com"
 AUTHOR = "Feedly"
 REQUIRES_PYTHON = ">=3.6.0"
-VERSION = "0.23.2"
+VERSION = "0.24"
 
 # What packages are required for this module to be executed?
 with open("requirements.txt") as f: