From 07cce5392e036d0abffc7a4de4111dffb572c0a3 Mon Sep 17 00:00:00 2001 From: serply Date: Sat, 19 Sep 2026 09:41:45 -0400 Subject: [PATCH 1/2] feat(enrichers): add org_to_news enricher backed by Serply Pivots an Organization to the news coverage that mentions it, emitting one Website per article and relating it back with MENTIONED_IN. The existing Organization enrichers are registry and infrastructure pivots (Whoxy, SIRENE, asnmap), so none of them surface what has been said about an organization publicly. Coverage names the people, partners and incidents that no registry records. Language and market are pinned by default, since the vertical otherwise infers them from where the call originates and answers an English query with articles in another language. Recency is exposed as a time_range param. --- docs/sources/available-enrichers.mdx | 3 + .../organization/to_news.py | 304 ++++++++++++++++++ .../tests/enrichers/test_org_to_news.py | 291 +++++++++++++++++ 3 files changed, 598 insertions(+) create mode 100644 flowsint-enrichers/src/flowsint_enrichers/organization/to_news.py create mode 100644 flowsint-enrichers/tests/enrichers/test_org_to_news.py diff --git a/docs/sources/available-enrichers.mdx b/docs/sources/available-enrichers.mdx index 93abe656..978a03cf 100644 --- a/docs/sources/available-enrichers.mdx +++ b/docs/sources/available-enrichers.mdx @@ -113,6 +113,9 @@ Tools/APIs: SIRENE (SireneTool) — see [INSEE Sirene API](https://api.insee.fr/ **org_to_asn**: Find ASNs associated with an organization name. Tools/Pivots: [asnmap](https://github.com/projectdiscovery/asnmap) (CLI), [jq](https://jqlang.github.io/jq/) (CLI) +**org_to_news**: Find news coverage of an organization, as Website nodes with the article title and snippet. +Tools/APIs: [Serply API](https://serply.io/docs) + ### Phone **phone_to_infos**: Probe phone footprint across services (demo modules) and normalize number. Tools/APIs: ignorant modules (Amazon, Snapchat, Instagram), [httpx](https://github.com/projectdiscovery/httpx) diff --git a/flowsint-enrichers/src/flowsint_enrichers/organization/to_news.py b/flowsint-enrichers/src/flowsint_enrichers/organization/to_news.py new file mode 100644 index 00000000..9d5cc026 --- /dev/null +++ b/flowsint-enrichers/src/flowsint_enrichers/organization/to_news.py @@ -0,0 +1,304 @@ +import os +from typing import Any, Dict, List, Optional + +import requests + +from flowsint_core.core.enricher_base import Enricher +from flowsint_core.core.logger import Logger +from flowsint_core.core.vault import VaultProtocol +from flowsint_enrichers.registry import flowsint_enricher +from flowsint_types.organization import Organization +from flowsint_types.website import Website + +SEARCH_ENDPOINT = "https://api.serply.io/v1/search/" +# `num` is an approximate cap of about ten rows per call, not an exact count, so +# the window size is fixed and surplus rows are trimmed against max_results. +PAGE_SIZE = 10 +DEFAULT_MAX_RESULTS = 20 +# The news vertical is selected with Google's own `tbm`, off the same endpoint. +NEWS_VERTICAL = "nws" +# Recency maps onto Google's `tbs=qdr:`. +RECENCY = {"d": "qdr:d", "w": "qdr:w", "m": "qdr:m", "y": "qdr:y"} +DEFAULT_LANGUAGE = "en" +DEFAULT_REGION = "us" + + +@flowsint_enricher +class OrgToNewsEnricher(Enricher): + """[Serply] Find news coverage of an organization. + + Runs the organization name as an exact-phrase query against the news + vertical and emits one Website per article. + Expects a `SERPLY_API_KEY` vault secret. + """ + + # Define types as class attributes - base class handles schema generation automatically + InputType = Organization + OutputType = Website + + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + vault: Optional[VaultProtocol] = None, + params: Optional[Dict[str, Any]] = None, + **kwargs: Any, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + **kwargs, + ) + + @classmethod + def name(cls) -> str: + return "org_to_news" + + @classmethod + def category(cls) -> str: + return "Organization" + + @classmethod + def key(cls) -> str: + return "name" + + @classmethod + def get_params_schema(cls) -> List[Dict[str, Any]]: + """Declare required parameters for this enricher""" + return [ + { + "name": "SERPLY_API_KEY", + "type": "vaultSecret", + "description": "Your Serply API key, from serply.io.", + "required": True, + }, + { + "name": "time_range", + "type": "select", + "description": "Only keep coverage published within this window.", + "required": False, + "default": "any", + "options": [ + {"label": "Any time", "value": "any"}, + {"label": "Past 24 hours", "value": "d"}, + {"label": "Past week", "value": "w"}, + {"label": "Past month", "value": "m"}, + {"label": "Past year", "value": "y"}, + ], + }, + { + "name": "language", + "type": "string", + "description": f"Language of the coverage, as a two-letter code. Default: {DEFAULT_LANGUAGE}", + "required": False, + "default": DEFAULT_LANGUAGE, + }, + { + "name": "region", + "type": "string", + "description": f"Market the coverage is ranked for, as a two-letter country code. Default: {DEFAULT_REGION}", + "required": False, + "default": DEFAULT_REGION, + }, + { + "name": "max_results", + "type": "number", + "description": f"Maximum articles per organization. Default: {DEFAULT_MAX_RESULTS}", + "required": False, + }, + ] + + def _max_results(self) -> int: + limit = self.params.get("max_results") + if limit is None: + return DEFAULT_MAX_RESULTS + try: + return max(1, int(limit)) + except (TypeError, ValueError): + Logger.warn( + self.sketch_id, + { + "message": f"(OrgToNews) Invalid max_results '{limit}', falling back to {DEFAULT_MAX_RESULTS}." + }, + ) + return DEFAULT_MAX_RESULTS + + def _base_query_params(self, query: str) -> Dict[str, str | int]: + """Build the parameters shared by every window of one scan. + + Language and market are pinned on purpose. Left unset, the index infers + them from where the call originates, which for a server means the + vertical answers an English query with articles in whatever language + that host looks like it wants. + """ + query_params: Dict[str, str | int] = { + "q": query, + "num": PAGE_SIZE, + "tbm": NEWS_VERTICAL, + "hl": (self.params.get("language") or DEFAULT_LANGUAGE).strip(), + "gl": (self.params.get("region") or DEFAULT_REGION).strip(), + } + + time_range = (self.params.get("time_range") or "").strip() + if time_range in RECENCY: + query_params["tbs"] = RECENCY[time_range] + elif time_range and time_range != "any": + Logger.warn( + self.sketch_id, + { + "message": f"(OrgToNews) Unknown time_range '{time_range}', searching all of time instead." + }, + ) + + return query_params + + def _fetch_window( + self, query: str, start: int, api_key: str + ) -> List[Dict[str, Any]]: + """Fetch one result window. `start` is the only offset the API honours.""" + query_params = self._base_query_params(query) + query_params["start"] = start + + api_request = requests.get( + SEARCH_ENDPOINT, + params=query_params, + headers={ + "X-Api-Key": api_key, + "Accept": "application/json", + "User-Agent": "FlowsInt-Enricher", + }, + timeout=30, + ) + + if api_request.status_code != 200: + Logger.error( + self.sketch_id, + { + "message": f"(OrgToNews) Search failed for '{query}' (HTTP {api_request.status_code}): {api_request.text}" + }, + ) + return [] + + return api_request.json().get("results") or [] + + async def scan(self, data: List[InputType]) -> List[OutputType]: + results: List[OutputType] = [] + + api_key = self.get_secret("SERPLY_API_KEY", os.getenv("SERPLY_API_KEY")) + max_results = self._max_results() + + for org in data: + org_name = str(org.name).strip() if org.name else "" + if not org_name: + Logger.warn( + self.sketch_id, + { + "message": "(OrgToNews) Skipping an organization with no name to search on." + }, + ) + continue + + try: + # Quoting keeps a multi-word name together, so a two-word + # company does not match every article carrying both words. + query = f'"{org_name}"' + articles: List[Website] = [] + seen_links: set[str] = set() + start = 0 + + while len(articles) < max_results: + rows = self._fetch_window(query, start, api_key) + if not rows: + break + + before = len(articles) + for row in rows: + link = row.get("link") + if not link or link in seen_links: + continue + seen_links.add(link) + + try: + article = Website( + url=link, + title=row.get("title"), + description=row.get("description"), + ) + except Exception as e: + # The vertical occasionally returns links that are + # not addressable as an HTTP URL; skip those rows. + Logger.warn( + self.sketch_id, + { + "message": f"(OrgToNews) Skipping unusable result '{link}': {e}" + }, + ) + continue + + # Carry the source name through to postprocess for graph + # wiring, the way domain_to_dns threads its source domain. + setattr(article, "_source_org_name", org_name) + articles.append(article) + + if len(articles) >= max_results: + break + + # A window that adds nothing new means the result set is + # exhausted (or the offset stopped moving), so stop paging + # rather than spend a credit per duplicate window. + if len(articles) == before: + break + start += PAGE_SIZE + + if not articles: + Logger.info( + self.sketch_id, + {"message": f"(OrgToNews) No coverage found for '{org_name}'."}, + ) + + results.extend(articles) + + except Exception as e: + Logger.error( + self.sketch_id, + { + "message": f"(OrgToNews) Exception while querying {org_name}: {e}" + }, + ) + + return results + + def postprocess( + self, results: List[OutputType], input_data: Optional[List[InputType]] = None + ) -> List[OutputType]: + if not self._graph_service: + return results + + for article in results: + source_org_name = getattr(article, "_source_org_name", None) + if not source_org_name: + continue + + org = Organization(name=source_org_name) + self.create_node(org) + self.create_node(article) + + # An article covering an organization mentions it; it is not a site + # the organization owns, so this is not the HAS_WEBSITE edge. + self.create_relationship(org, article, "MENTIONED_IN") + self.log_graph_message( + f"(OrgToNews) {source_org_name} -> {str(article.url)}" + ) + + # Clean up the temporary attribute used to thread context. + delattr(article, "_source_org_name") + + return results + + +# Make types available at module level for easy access +InputType = OrgToNewsEnricher.InputType +OutputType = OrgToNewsEnricher.OutputType diff --git a/flowsint-enrichers/tests/enrichers/test_org_to_news.py b/flowsint-enrichers/tests/enrichers/test_org_to_news.py new file mode 100644 index 00000000..68c7864b --- /dev/null +++ b/flowsint-enrichers/tests/enrichers/test_org_to_news.py @@ -0,0 +1,291 @@ +import pytest + +from flowsint_enrichers import ENRICHER_REGISTRY +from flowsint_enrichers.organization.to_news import ( + PAGE_SIZE, + OrgToNewsEnricher, +) +from flowsint_types.organization import Organization +from flowsint_types.website import Website + + +# --------------------------------------------------------------------------- +# Registry wiring +# --------------------------------------------------------------------------- +def test_org_to_news_is_registered(): + enricher = ENRICHER_REGISTRY.get_enricher("org_to_news", "123", "123") + assert enricher.name() == "org_to_news" + + +def test_org_to_news_metadata(): + assert OrgToNewsEnricher.category() == "Organization" + assert OrgToNewsEnricher.key() == "name" + assert OrgToNewsEnricher.input_schema()["type"] == "Organization" + assert OrgToNewsEnricher.output_schema()["type"] == "Website" + + +def test_api_key_is_declared_as_a_required_vault_secret(): + schema = {p["name"]: p for p in OrgToNewsEnricher.get_params_schema()} + assert schema["SERPLY_API_KEY"]["type"] == "vaultSecret" + assert schema["SERPLY_API_KEY"]["required"] is True + assert schema["time_range"]["required"] is False + assert [o["value"] for o in schema["time_range"]["options"]] == [ + "any", + "d", + "w", + "m", + "y", + ] + + +# --------------------------------------------------------------------------- +# scan() - HTTP layer mocked, no API key and no network needed +# --------------------------------------------------------------------------- +class _FakeResponse: + def __init__(self, payload, status_code=200): + self._payload = payload + self.status_code = status_code + self.text = str(payload) + + def json(self): + return self._payload + + +class _FakeRequests: + """Stands in for the `requests` module inside the enricher.""" + + def __init__(self, windows, status_code=200): + self._windows = windows + self._status_code = status_code + self.calls = [] + + def get(self, url, params=None, headers=None, timeout=None): + self.calls.append({"url": url, "params": params, "headers": headers}) + index = len(self.calls) - 1 + rows = self._windows[index] if index < len(self._windows) else [] + return _FakeResponse({"results": rows}, self._status_code) + + +def _row(position, link, title="Title", description="Description"): + return { + "position": position, + "title": title, + "link": link, + "description": description, + } + + +def _enricher(fake, monkeypatch, params=None): + monkeypatch.setattr( + "flowsint_enrichers.organization.to_news.requests", fake, raising=True + ) + return OrgToNewsEnricher( + sketch_id="s", + scan_id="t", + graph_service=None, + params={"SERPLY_API_KEY": "test-key", **(params or {})}, + ) + + +@pytest.mark.asyncio +async def test_scan_maps_articles_to_websites(monkeypatch): + fake = _FakeRequests( + [[_row(1, "https://news.example.com/acme-raises", "Acme raises", "Snippet")]] + ) + enricher = _enricher(fake, monkeypatch) + + results = await enricher.scan([Organization(name="Acme Corp")]) + + assert len(results) == 1 + article = results[0] + assert isinstance(article, Website) + assert str(article.url) == "https://news.example.com/acme-raises" + assert article.title == "Acme raises" + assert article.description == "Snippet" + + params = fake.calls[0]["params"] + # Quoting keeps a multi-word name together instead of matching both words + # anywhere on the page. + assert params["q"] == '"Acme Corp"' + # The news vertical is selected with Google's own `tbm`. + assert params["tbm"] == "nws" + assert fake.calls[0]["headers"]["X-Api-Key"] == "test-key" + + +@pytest.mark.asyncio +async def test_language_and_region_are_pinned_by_default(monkeypatch): + fake = _FakeRequests([[_row(1, "https://news.example.com/a")]]) + enricher = _enricher(fake, monkeypatch) + + await enricher.scan([Organization(name="Acme Corp")]) + + # Left unset these are inferred from where the call originates, which for a + # server answers an English query with articles in another language. + assert fake.calls[0]["params"]["hl"] == "en" + assert fake.calls[0]["params"]["gl"] == "us" + + +@pytest.mark.asyncio +async def test_language_and_region_are_overridable(monkeypatch): + fake = _FakeRequests([[_row(1, "https://news.example.com/a")]]) + enricher = _enricher(fake, monkeypatch, {"language": "fr", "region": "fr"}) + + await enricher.scan([Organization(name="Acme Corp")]) + + assert fake.calls[0]["params"]["hl"] == "fr" + assert fake.calls[0]["params"]["gl"] == "fr" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "time_range,expected", + [("d", "qdr:d"), ("w", "qdr:w"), ("m", "qdr:m"), ("y", "qdr:y")], +) +async def test_time_range_maps_onto_the_recency_filter( + monkeypatch, time_range, expected +): + fake = _FakeRequests([[_row(1, "https://news.example.com/a")]]) + enricher = _enricher(fake, monkeypatch, {"time_range": time_range}) + + await enricher.scan([Organization(name="Acme Corp")]) + + assert fake.calls[0]["params"]["tbs"] == expected + + +@pytest.mark.asyncio +@pytest.mark.parametrize("time_range", ["any", "", "since-tuesday"]) +async def test_unfiltered_time_ranges_send_no_recency_filter(monkeypatch, time_range): + fake = _FakeRequests([[_row(1, "https://news.example.com/a")]]) + enricher = _enricher(fake, monkeypatch, {"time_range": time_range}) + + await enricher.scan([Organization(name="Acme Corp")]) + + # An unrecognised value searches all of time rather than failing the scan. + assert "tbs" not in fake.calls[0]["params"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("name", ["", " ", None]) +async def test_scan_skips_an_organization_without_a_name(monkeypatch, name): + fake = _FakeRequests([[_row(1, "https://news.example.com/a")]]) + enricher = _enricher(fake, monkeypatch) + + # `Organization.name` is typed Any, so it can arrive empty or unset from an + # upstream pivot. Nothing to search on, so no credit is spent. + assert await enricher.scan([Organization(name=name)]) == [] + assert fake.calls == [] + + +@pytest.mark.asyncio +async def test_scan_pages_with_start_and_dedupes_links(monkeypatch): + first = [_row(1, "https://a.example.com/"), _row(2, "https://b.example.com/")] + second = [_row(1, "https://a.example.com/"), _row(2, "https://c.example.com/")] + fake = _FakeRequests([first, second]) + enricher = _enricher(fake, monkeypatch, {"max_results": "4"}) + + results = await enricher.scan([Organization(name="Acme Corp")]) + + assert [str(a.url) for a in results] == [ + "https://a.example.com/", + "https://b.example.com/", + "https://c.example.com/", + ] + # `start` is the only offset the API honours, so it is what advances. + assert [c["params"]["start"] for c in fake.calls[:2]] == [0, PAGE_SIZE] + # Every window stays on the news vertical, not just the first. + assert all(c["params"]["tbm"] == "nws" for c in fake.calls) + + +@pytest.mark.asyncio +async def test_scan_stops_when_a_window_adds_nothing_new(monkeypatch): + repeated = [_row(1, "https://a.example.com/")] + fake = _FakeRequests([repeated] * 10) + enricher = _enricher(fake, monkeypatch, {"max_results": "50"}) + + results = await enricher.scan([Organization(name="Acme Corp")]) + + # One window of duplicates ends the loop instead of refetching forever. + assert len(results) == 1 + assert len(fake.calls) == 2 + + +@pytest.mark.asyncio +async def test_scan_trims_surplus_rows_to_max_results(monkeypatch): + rows = [_row(i, f"https://p{i}.example.com/") for i in range(1, 6)] + fake = _FakeRequests([rows]) + enricher = _enricher(fake, monkeypatch, {"max_results": "2"}) + + results = await enricher.scan([Organization(name="Acme Corp")]) + assert len(results) == 2 + + +@pytest.mark.asyncio +async def test_scan_returns_nothing_on_api_error(monkeypatch): + fake = _FakeRequests([[]], status_code=403) + enricher = _enricher(fake, monkeypatch) + + assert await enricher.scan([Organization(name="Acme Corp")]) == [] + + +@pytest.mark.asyncio +async def test_scan_skips_rows_without_a_usable_link(monkeypatch): + rows = [_row(1, ""), _row(2, "not-a-url"), _row(3, "https://ok.example.com/")] + fake = _FakeRequests([rows]) + enricher = _enricher(fake, monkeypatch) + + results = await enricher.scan([Organization(name="Acme Corp")]) + assert [str(a.url) for a in results] == ["https://ok.example.com/"] + + +# --------------------------------------------------------------------------- +# postprocess() - graph service mocked +# --------------------------------------------------------------------------- +class _FakeGraphService: + def __init__(self): + self.nodes = [] + self.relationships = [] + self.messages = [] + + def create_node_from_flowsint_type(self, node_obj): + self.nodes.append(node_obj) + + def create_relationship(self, from_obj, to_obj, rel_label): + self.relationships.append((from_obj, to_obj, rel_label)) + + def log_graph_message(self, message): + self.messages.append(message) + + +def _article(url, org_name="Acme Corp"): + article = Website(url=url) + setattr(article, "_source_org_name", org_name) + return article + + +def test_postprocess_relates_each_article_back_to_the_organization(): + graph = _FakeGraphService() + enricher = OrgToNewsEnricher(sketch_id="s", scan_id="t", graph_service=graph) + article = _article("https://news.example.com/acme-raises") + + assert enricher.postprocess([article]) == [article] + org = Organization(name="Acme Corp") + assert graph.nodes == [org, article] + # Coverage is a mention, not a site the organization owns, so this is not + # the HAS_WEBSITE edge domain_to_website writes. + assert graph.relationships == [(org, article, "MENTIONED_IN")] + # The attribute used to thread context does not leak onto the node. + assert not hasattr(article, "_source_org_name") + + +def test_postprocess_keeps_each_article_with_its_own_organization(): + graph = _FakeGraphService() + enricher = OrgToNewsEnricher(sketch_id="s", scan_id="t", graph_service=graph) + first = _article("https://news.example.com/acme", "Acme Corp") + second = _article("https://news.example.com/globex", "Globex") + + enricher.postprocess([first, second]) + + assert [(rel[0].name, str(rel[1].url)) for rel in graph.relationships] == [ + ("Acme Corp", "https://news.example.com/acme"), + ("Globex", "https://news.example.com/globex"), + ] From 8c3b309c1bc0a43df3e3564eb34967b0bbf52081 Mon Sep 17 00:00:00 2001 From: serply Date: Sun, 20 Sep 2026 17:35:32 -0400 Subject: [PATCH 2/2] feat(enrichers): declare required_params on org_to_news The frontend needs a flag for whether an enricher takes params at all. Follows the existing override in ip/to_asn.py, email/to_leaks.py and six other enrichers, matching the fix already applied to #223. --- .../src/flowsint_enrichers/organization/to_news.py | 4 ++++ flowsint-enrichers/tests/enrichers/test_org_to_news.py | 4 ++++ 2 files changed, 8 insertions(+) diff --git a/flowsint-enrichers/src/flowsint_enrichers/organization/to_news.py b/flowsint-enrichers/src/flowsint_enrichers/organization/to_news.py index 9d5cc026..a402c3ee 100644 --- a/flowsint-enrichers/src/flowsint_enrichers/organization/to_news.py +++ b/flowsint-enrichers/src/flowsint_enrichers/organization/to_news.py @@ -65,6 +65,10 @@ def category(cls) -> str: def key(cls) -> str: return "name" + @classmethod + def required_params(cls) -> bool: + return True + @classmethod def get_params_schema(cls) -> List[Dict[str, Any]]: """Declare required parameters for this enricher""" diff --git a/flowsint-enrichers/tests/enrichers/test_org_to_news.py b/flowsint-enrichers/tests/enrichers/test_org_to_news.py index 68c7864b..bded399a 100644 --- a/flowsint-enrichers/tests/enrichers/test_org_to_news.py +++ b/flowsint-enrichers/tests/enrichers/test_org_to_news.py @@ -38,6 +38,10 @@ def test_api_key_is_declared_as_a_required_vault_secret(): ] +def test_enricher_declares_that_it_requires_params(): + assert OrgToNewsEnricher.required_params() is True + + # --------------------------------------------------------------------------- # scan() - HTTP layer mocked, no API key and no network needed # ---------------------------------------------------------------------------