Skip to content

IoCDownloader: older_than param #30

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Edouard360 merged 5 commits into from
Oct 10, 2025
Merged

IoCDownloader: older_than param #30

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a5b1ce4
older than iocs downloader
Edouard360 Oct 9, 2025
63cfaf6
formatting
Edouard360 Oct 9, 2025
b3f365b
bump version
Edouard360 Oct 9, 2025
ecec168
add max batches as well
Edouard360 Oct 9, 2025
25a9ccb
pr review
Edouard360 Oct 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 56 additions & 13 deletions feedly/api_client/enterprise/indicators_of_compromise.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,9 @@
from datetime import datetime
from enum import Enum
from itertools import chain
from requests import Response
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
Expand All @@ -24,38 +24,61 @@ class IoCDownloaderABC(ABC, Generic[T]):
RELATIVE_URL = "/v3/enterprise/ioc"
FORMAT: ClassVar[str]

def __init__(self, session: FeedlySession, newer_than: Optional[datetime], stream_id: str):
def __init__(
self,
session: FeedlySession,
newer_than: Optional[datetime],
older_than: Optional[datetime],
stream_id: str,
max_batches: Optional[int] = 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.
Use the older_than parameter to filter articles that are older than your last call.
Use the max_batches parameter to limit the number of batches/pages to retrieve.

: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
:param older_than: Only articles older than this parameter will be used. If None only one call will be make,

@EmmaDemarecaux EmmaDemarecaux Oct 13, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
:param older_than: Only articles older than this parameter will be used. If None only one call will be make,
:param older_than: Only articles older than this parameter will be used. If None only one call will be made,

and the continuation will be ignored
:param max_batches: Maximum number of batches to retrieve. If None, will continue until no more data is available
"""
self.newer_than = newer_than
self.older_than = older_than
self.session = session
self.stream_id = stream_id
self.max_batches = max_batches

def download_all(self) -> List[T]:
return self._merge(self.stream_bundles())

def stream_bundles(self) -> Iterable[T]:
continuation = None
batch_count = 0
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,
},
)
params = {
"continuation": continuation,
"streamId": self.stream_id,
"format": self.FORMAT,
}
if self.newer_than:
params["newerThan"] = int(self.newer_than.timestamp())
if self.older_than:
params["olderThan"] = int(self.older_than.timestamp())

resp = self.session.make_api_request(f"{self.RELATIVE_URL}", params=params)

@EmmaDemarecaux EmmaDemarecaux Oct 13, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
resp = self.session.make_api_request(f"{self.RELATIVE_URL}", params=params)
resp = self.session.make_api_request(self.RELATIVE_URL, params=params)

yield self._parse_response(resp)
batch_count += 1

# Check if we've reached max_batches limit
if self.max_batches and batch_count >= self.max_batches:
return

if not self.newer_than or "link" not in resp.headers:
return
next_url = resp.headers["link"][1:].split(">")[0]
Expand All @@ -70,21 +93,35 @@ def _merge(self, resp_jsons: Iterable[T]) -> T:


class IoCDownloaderBuilder:
def __init__(self, session: FeedlySession, format: IoCFormat, newer_than: Optional[datetime] = None):
def __init__(
self,
session: FeedlySession,
format: IoCFormat,
newer_than: Optional[datetime] = None,
older_than: Optional[datetime] = None,
max_batches: Optional[int] = 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.
Use the older_than parameter to filter articles that are older than your last call.
Use the max_batches parameter to limit the number of batches/pages to retrieve.

: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
:param older_than: Only articles older than this parameter will be used. If None only one call will be make,
and the continuation will be ignored
:param max_batches: Maximum number of batches to retrieve. If None, will continue until no more data is available
"""
self.session = session
self.format = format
self.newer_than = newer_than
self.older_than = older_than
self.max_batches = max_batches

self.session.api_host = "https://cloud.feedly.com"
self.user = self.session.user
Expand Down Expand Up @@ -113,7 +150,13 @@ def from_stream_id(self, stream_id: str) -> IoCDownloaderABC:
IoCFormat.STIX: StixIoCDownloader,
IoCFormat.CSV: CsvIoCDownloader,
}
return format2class[self.format](session=self.session, newer_than=self.newer_than, stream_id=stream_id)
return format2class[self.format](
session=self.session,
newer_than=self.newer_than,
older_than=self.older_than,
stream_id=stream_id,
max_batches=self.max_batches,
)


class StixIoCDownloader(IoCDownloaderABC[Dict]):
Expand Down
2 changes: 1 addition & 1 deletion setup.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
EMAIL = "ml@feedly.com"
AUTHOR = "Feedly"
REQUIRES_PYTHON = ">=3.6.0"
VERSION = "0.26"
VERSION = "0.27"

# What packages are required for this module to be executed?
with open("requirements.txt") as f:
Expand Down