diff --git a/examples/enterprise/export_indicators_of_compromise_from_a_stream_as_csv.py b/examples/enterprise/export_indicators_of_compromise_from_a_stream_as_csv.py new file mode 100644 index 0000000..1e28680 --- /dev/null +++ b/examples/enterprise/export_indicators_of_compromise_from_a_stream_as_csv.py @@ -0,0 +1,44 @@ +import json +from csv import DictWriter +from datetime import datetime, timedelta +from pathlib import Path +from pprint import pprint + +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_as_csv(): + """ + This example will save a CSV file 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 CSV IoC downloader builder object, and limit it to 12 hours + # Usually newer_than will be the datetime of the last fetch + downloader_builder = IoCDownloaderBuilder( + session=session, newer_than=datetime.now() - timedelta(hours=12), format=IoCFormat.CSV + ) + + # Fetch the IoC from all the enterprise categories + # 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() + iocs = downloader.download_all() + + # Save the IoCs in a CSV + with (RESULTS_DIR / "ioc_example.csv").open("w") as f: + writer = DictWriter(f, fieldnames=list(iocs[0].keys())) + writer.writerows(iocs) + + +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_as_csv) diff --git a/feedly/api_client/enterprise/indicators_of_compromise.py b/feedly/api_client/enterprise/indicators_of_compromise.py index 94b3b49..3526844 100644 --- a/feedly/api_client/enterprise/indicators_of_compromise.py +++ b/feedly/api_client/enterprise/indicators_of_compromise.py @@ -1,11 +1,11 @@ import uuid from abc import ABC, abstractmethod +from csv import DictReader from datetime import datetime 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 @@ -17,6 +17,7 @@ class IoCFormat(Enum): MISP = "misp" STIX = "stix2.1" + CSV = "csv" class IoCDownloaderABC(ABC, Generic[T]): @@ -107,7 +108,11 @@ def from_stream(self, stream: Streamable) -> IoCDownloaderABC: return self.from_stream_id(stream.id) def from_stream_id(self, stream_id: str) -> IoCDownloaderABC: - format2class = {IoCFormat.MISP: MispIoCDownloader, IoCFormat.STIX: StixIoCDownloader} + format2class = { + IoCFormat.MISP: MispIoCDownloader, + IoCFormat.STIX: StixIoCDownloader, + IoCFormat.CSV: CsvIoCDownloader, + } return format2class[self.format](session=self.session, newer_than=self.newer_than, stream_id=stream_id) @@ -127,3 +132,13 @@ class MispIoCDownloader(IoCDownloaderABC[Dict]): def _merge(self, resp_jsons: Iterable[Dict]) -> Dict: return {"response": list(chain.from_iterable(resp_json["response"] for resp_json in resp_jsons))} + + +class CsvIoCDownloader(IoCDownloaderABC[List[Dict]]): + FORMAT = "csv" + + def _merge(self, resp_jsons: Iterable[List[Dict]]) -> List[Dict]: + return list(chain.from_iterable(resp_jsons)) + + def _parse_response(self, resp: Response) -> List[Dict]: + return list(DictReader(resp.text.splitlines())) \ No newline at end of file diff --git a/setup.py b/setup.py index 13d040b..1fa7f3f 100644 --- a/setup.py +++ b/setup.py @@ -18,7 +18,7 @@ EMAIL = "ml@feedly.com" AUTHOR = "Feedly" REQUIRES_PYTHON = ">=3.6.0" -VERSION = "0.25" +VERSION = "0.26" # What packages are required for this module to be executed? with open("requirements.txt") as f: