From 47a27c597b497b767d8f0722c02ae12419293101 Mon Sep 17 00:00:00 2001 From: Slava Olesik Date: Mon, 26 Oct 2026 13:57:39 +0200 Subject: [PATCH 1/3] Add NetAPI threat intelligence provider --- .../msticpy.context.tiproviders.netapi.rst | 7 + .../api/msticpy.context.tiproviders.rst | 1 + docs/source/data_acquisition/TIProviders.rst | 7 +- docs/source/getting_started/msticpyconfig.rst | 4 + msticpy/context/tiproviders/__init__.py | 1 + msticpy/context/tiproviders/netapi.py | 129 ++++++++++ msticpy/resources/mpconfig_defaults.yaml | 3 + tests/context/test_netapi.py | 226 ++++++++++++++++++ tests/context/test_tiproviders.py | 16 ++ tests/msticpyconfig-test.yaml | 3 + tests/testdata/msticpyconfig.yaml | 3 + 11 files changed, 399 insertions(+), 1 deletion(-) create mode 100644 docs/source/api/msticpy.context.tiproviders.netapi.rst create mode 100644 msticpy/context/tiproviders/netapi.py create mode 100644 tests/context/test_netapi.py diff --git a/docs/source/api/msticpy.context.tiproviders.netapi.rst b/docs/source/api/msticpy.context.tiproviders.netapi.rst new file mode 100644 index 00000000..a54b6b55 --- /dev/null +++ b/docs/source/api/msticpy.context.tiproviders.netapi.rst @@ -0,0 +1,7 @@ +msticpy.context.tiproviders.netapi module +========================================= + +.. automodule:: msticpy.context.tiproviders.netapi + :members: + :show-inheritance: + :undoc-members: diff --git a/docs/source/api/msticpy.context.tiproviders.rst b/docs/source/api/msticpy.context.tiproviders.rst index 34d285bd..cf7cec15 100644 --- a/docs/source/api/msticpy.context.tiproviders.rst +++ b/docs/source/api/msticpy.context.tiproviders.rst @@ -24,6 +24,7 @@ Submodules msticpy.context.tiproviders.ip_quality_score msticpy.context.tiproviders.kql_base msticpy.context.tiproviders.mblookup + msticpy.context.tiproviders.netapi msticpy.context.tiproviders.open_page_rank msticpy.context.tiproviders.pulsedive msticpy.context.tiproviders.result_severity diff --git a/docs/source/data_acquisition/TIProviders.rst b/docs/source/data_acquisition/TIProviders.rst index 1a6184c3..5bcab3cc 100644 --- a/docs/source/data_acquisition/TIProviders.rst +++ b/docs/source/data_acquisition/TIProviders.rst @@ -33,6 +33,7 @@ Features - **CrowdSec** - **AbuseIPDB** - **ScanMalware** + - **NetAPI** - Other pseudo-TI providers are also included: @@ -286,6 +287,9 @@ fictitious - the format of the keys may differ from what is shown ScanMalware: Primary: True Provider: "ScanMalware" + NetAPI: + Primary: True + Provider: "NetAPI" You need to tell `TILookup` to refresh its configuration. @@ -307,7 +311,8 @@ of providers loaded. 'AzSTI - Microsoft Sentinel TI provider class. (primary)', 'CrowdSec - CrowdSec CTI Smoke Lookup. (primary)', 'AbuseIPDB - AbuseIPDB Lookup. (primary)', - 'ScanMalware - ScanMalware Lookup. (primary)'] + 'ScanMalware - ScanMalware Lookup. (primary)', + 'NetAPI - NetAPI Lookup. (primary)'] .. warning:: Depending on the type of account that you have with a provider, they will typically impose a limit diff --git a/docs/source/getting_started/msticpyconfig.rst b/docs/source/getting_started/msticpyconfig.rst index 45340bc9..b81875b8 100644 --- a/docs/source/getting_started/msticpyconfig.rst +++ b/docs/source/getting_started/msticpyconfig.rst @@ -118,6 +118,7 @@ Currently supported provider names are: - CrowdSec - AbuseIPDB - ScanMalware +- NetAPI .. code:: yaml @@ -152,6 +153,9 @@ Currently supported provider names are: ScanMalware: Primary: True Provider: "ScanMalware" + NetAPI: + Primary: True + Provider: "NetAPI" .. note:: You store values in the ``Args`` section as simple strings, as names of environment variables containing the value, or diff --git a/msticpy/context/tiproviders/__init__.py b/msticpy/context/tiproviders/__init__.py index ea5370f2..a01c076d 100644 --- a/msticpy/context/tiproviders/__init__.py +++ b/msticpy/context/tiproviders/__init__.py @@ -31,4 +31,5 @@ "BinaryEdge": ("binaryedge", "BinaryEdge"), "Cyberint": ("cyberint", "Cyberint"), "ScanMalware": ("scanmalware", "ScanMalware"), + "NetAPI": ("netapi", "NetAPI"), } diff --git a/msticpy/context/tiproviders/netapi.py b/msticpy/context/tiproviders/netapi.py new file mode 100644 index 00000000..edd6316f --- /dev/null +++ b/msticpy/context/tiproviders/netapi.py @@ -0,0 +1,129 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +""" +NetAPI Provider. + +Looks up domains, hostnames, URLs and IP addresses in the NetAPI compromised +feed: the hosts that public threat feeds (URLhaus, CERT Polska, blackbook and +others) report for malware or phishing. The default query answers whether the +observable is listed now (seen within the last 24 hours), was listed in the +past only, or was never listed. The ``rank`` query subtype returns the +position of a domain in the NetAPI Top 1M, a daily popularity ranking. + +The API needs no key. Anonymous use is limited to 30 requests a minute and +1,000 a day per IP address. + +Input can be a single IoC observable or a pandas DataFrame containing +multiple observables. + +""" + +from __future__ import annotations + +from typing import Any, ClassVar + +from typing_extensions import Self + +from ..._version import VERSION +from ..http_provider import APILookupParams +from .ti_http_provider import HttpTIProvider +from .ti_provider_base import ResultSeverity + +__version__ = VERSION +__author__ = "Slava Olesik" + +_HEADERS: dict[str, str] = {"User-Agent": "netapi-msticpy/1.0"} + +_CHECK = APILookupParams( + path="/api-json/", + headers=_HEADERS, + params={"method": "compromised-check", "value": "{observable}"}, +) +_RANK = APILookupParams( + path="/api-json/", + headers=_HEADERS, + params={"method": "domain-rank", "domain": "{observable}"}, +) + + +class NetAPI(HttpTIProvider): + """NetAPI Lookup.""" + + _BASE_URL: ClassVar[str] = "https://netapi.com" + + # The feed matches a hostname or a URL through its registered domain, so + # every name type uses the same query. Addresses have no popularity rank. + _QUERIES: ClassVar[dict[str, APILookupParams]] = { + "dns": _CHECK, + "hostname": _CHECK, + "url": _CHECK, + "ipv4": _CHECK, + "ipv6": _CHECK, + "dns-rank": _RANK, + "hostname-rank": _RANK, + } + + def parse_results(self: Self, response: dict) -> tuple[bool, ResultSeverity, Any]: + """ + Return the details of the response. + + Parameters + ---------- + response : Dict + The returned data response + + Returns + ------- + tuple[bool, ResultSeverity, Any] + bool = positive or negative hit + ResultSeverity = enumeration of severity + Object with match details + + """ + if self._failed_response(response): + return False, ResultSeverity.information, "Not found." + + raw: dict[str, Any] = response["RawResult"] + if "in_top_1m" in raw: + return self._parse_rank(raw) + return self._parse_listing(raw) + + @staticmethod + def _parse_listing(raw: dict[str, Any]) -> tuple[bool, ResultSeverity, Any]: + """Return the compromised feed listing of the observable.""" + if not raw.get("listed_ever"): + return False, ResultSeverity.information, "Not listed, now or in the past." + + # A past listing is mostly an incident that is over, often a legitimate + # site that was compromised and cleaned up, so only a current listing + # carries a severity. + severity = ResultSeverity.high if raw.get("listed_now") else ResultSeverity.information + registrar = raw.get("registrar") + details: dict[str, Any] = { + "Listed Now": bool(raw.get("listed_now")), + "Matched Value": raw.get("matched_value"), + "First Seen": raw.get("first_seen"), + "Last Seen": raw.get("last_seen"), + "Reported By": raw.get("block_reason"), + "Registrar": ( + registrar.get("brand_name") or registrar.get("name") + if isinstance(registrar, dict) + else None + ), + } + return True, severity, details + + @staticmethod + def _parse_rank(raw: dict[str, Any]) -> tuple[bool, ResultSeverity, Any]: + """Return the Top 1M position of the registered domain.""" + details: dict[str, Any] = { + "Domain": raw.get("domain"), + "Rank": raw.get("rank"), + "Snapshot Date": raw.get("snapshot_date"), + } + if not raw.get("in_top_1m"): + return False, ResultSeverity.information, details + return True, ResultSeverity.information, details diff --git a/msticpy/resources/mpconfig_defaults.yaml b/msticpy/resources/mpconfig_defaults.yaml index 75f0bfbb..bce3a0c5 100644 --- a/msticpy/resources/mpconfig_defaults.yaml +++ b/msticpy/resources/mpconfig_defaults.yaml @@ -104,6 +104,9 @@ TIProviders: ScanMalware: Primary: bool(default=True) Provider: "ScanMalware" + NetAPI: + Primary: bool(default=True) + Provider: "NetAPI" RiskIQ: Primary: bool(default=True) Provider: "RiskIQ" diff --git a/tests/context/test_netapi.py b/tests/context/test_netapi.py new file mode 100644 index 00000000..83a93db2 --- /dev/null +++ b/tests/context/test_netapi.py @@ -0,0 +1,226 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- +"""NetAPI TI provider tests. + +These cover what the shared parametrized harness in test_tiproviders.py +cannot reach: past and never-listed results, the registered-domain match, +the `rank` query subtype and the request parameters. + +No network access. The provider's httpx client is replaced throughout. +""" + +from __future__ import annotations + +from typing import Any + +import httpx +import pytest +import pytest_check as check + +from msticpy.context.tiproviders.netapi import NetAPI +from msticpy.context.tiproviders.result_severity import ResultSeverity + +__author__ = "Slava Olesik" + +pytestmark = pytest.mark.filterwarnings("ignore::DeprecationWarning") + +# Answers of the live API on 2026-10-26, trimmed to the fields the provider reads. +_LISTED_NOW: dict[str, Any] = { + "value": "www.000359.xyz", + "type": "domain", + "listed_now": True, + "listed_ever": True, + "matched_value": "000359.xyz", + "first_seen": "2023-02-25", + "last_seen": "2026-10-26 01:05:27 UTC", + "block_reason": "blackbook", + "registrar": { + "id": 1599, + "name": "Alibaba Cloud Computing Ltd.", + "brand_name": "Alibaba Cloud", + }, +} +_LISTED_PAST: dict[str, Any] = { + "value": "1.1.1.1", + "type": "ip", + "listed_now": False, + "listed_ever": True, + "matched_value": "1.1.1.1", + "first_seen": "2026-06-14", + "last_seen": "2026-06-17 01:05:02 UTC", + "block_reason": "urlhaus.abuse.ch", + "registrar": None, +} +_NOT_LISTED: dict[str, Any] = { + "value": "github.com", + "type": "domain", + "listed_now": False, + "listed_ever": False, +} +_RANK_IN: dict[str, Any] = { + "domain": "github.com", + "in_top_1m": True, + "rank": 11, + "snapshot_date": "2026-10-26", +} +_RANK_OUT: dict[str, Any] = { + "domain": "000359.xyz", + "in_top_1m": False, + "rank": None, + "snapshot_date": "2026-10-26", +} + + +class _Response(httpx.Response): + """Mocked httpx response.""" + + def __init__(self, json_data: Any, status_code: int = 200) -> None: + self.json_data = json_data + self.status_code = status_code + # str(response) reads the reason phrase from here on a failed lookup. + self.extensions = {} + + def json(self) -> Any: # noqa: D102 + return self.json_data + + +class _Client: + """Mock httpx client that records every request it is given.""" + + def __init__(self, data: Any, status_code: int = 200) -> None: + self.data = data + self.status_code = status_code + self.requests: list[dict[str, Any]] = [] + + def get(self, *args, **kwargs) -> _Response: + """Return the canned response and record the call.""" + self.requests.append( + { + "url": kwargs.get("url") or (args[0] if args else ""), + "params": kwargs.get("params", {}), + "headers": kwargs.get("headers", {}), + }, + ) + return _Response(self.data, self.status_code) + + +def _lookup( + data: Any, + ioc: str, + ioc_type: str, + query_type: str | None = None, + status_code: int = 200, +) -> tuple[dict[str, Any], _Client]: + """Run one lookup against a mock client and return the result row.""" + provider = NetAPI() + client = _Client(data, status_code) + provider._httpx_client = client # type: ignore[assignment] + result = provider.lookup_ioc(ioc, ioc_type, query_type) + return result.to_dict(orient="records")[0], client + + +def test_current_listing_is_high_severity(): + """A host reported within the last 24 hours is a current listing.""" + row, client = _lookup(_LISTED_NOW, "www.000359.xyz", "hostname") + + check.is_true(row["Result"]) + check.equal(row["Severity"], ResultSeverity.high.name) + check.equal(row["Details"]["Matched Value"], "000359.xyz") + check.equal(row["Details"]["Reported By"], "blackbook") + check.equal(row["Details"]["Registrar"], "Alibaba Cloud") + check.equal( + client.requests[0]["params"], + {"method": "compromised-check", "value": "www.000359.xyz"}, + ) + check.equal(client.requests[0]["url"], "https://netapi.com/api-json/") + + +def test_past_listing_is_reported_but_not_rated(): + """ + A past listing is mostly an incident that is over. + + It is a hit with its dates, but it must not carry the severity of a + current listing. + """ + row, _ = _lookup(_LISTED_PAST, "1.1.1.1", "ipv4") + + check.is_true(row["Result"]) + check.equal(row["Severity"], ResultSeverity.information.name) + check.is_false(row["Details"]["Listed Now"]) + check.equal(row["Details"]["Last Seen"], "2026-06-17 01:05:02 UTC") + + +def test_never_listed_is_a_negative_result(): + """A value that was never in the feed is not a hit.""" + row, _ = _lookup(_NOT_LISTED, "github.com", "dns") + + check.is_false(row["Result"]) + check.equal(row["Severity"], ResultSeverity.information.name) + check.equal(row["Details"], "Not listed, now or in the past.") + + +def test_url_is_sent_to_the_compromised_check(): + """The API reduces a URL to its host itself.""" + _, client = _lookup(_LISTED_NOW, "https://www.000359.xyz/login", "url") + + check.equal(client.requests[0]["params"]["method"], "compromised-check") + check.is_in("000359.xyz", client.requests[0]["params"]["value"]) + + +@pytest.mark.parametrize("ioc_type", ["dns", "hostname"]) +def test_rank_subtype_returns_the_top_1m_position(ioc_type): + """The `rank` subtype asks domain-rank and reports the position.""" + row, client = _lookup(_RANK_IN, "github.com", ioc_type, "rank") + + check.is_true(row["Result"]) + check.equal(row["Details"]["Rank"], 11) + check.equal(row["Details"]["Snapshot Date"], "2026-10-26") + check.equal( + client.requests[0]["params"], + {"method": "domain-rank", "domain": "github.com"}, + ) + + +def test_domain_outside_the_top_1m_is_not_a_hit(): + """A domain outside the ranking is a negative rank result.""" + row, _ = _lookup(_RANK_OUT, "000359.xyz", "dns", "rank") + + check.is_false(row["Result"]) + check.is_none(row["Details"]["Rank"]) + + +def test_rate_limit_is_not_a_hit(): + """An API error is reported as a failed lookup, never as a listing.""" + error = {"error": {"code": "rate_limited", "message": "Too many requests."}} + row, _ = _lookup(error, "github.com", "dns", status_code=429) + + check.is_false(row["Result"]) + check.equal(row["Status"], 429) + + +def test_no_key_and_own_user_agent(): + """The API is keyless, so no auth header is ever sent.""" + _, client = _lookup(_NOT_LISTED, "github.com", "dns") + + headers = client.requests[0]["headers"] + check.equal(headers.get("User-Agent"), "netapi-msticpy/1.0") + check.is_false(any(key.lower() == "authorization" for key in headers)) + + +def test_file_hashes_are_not_supported(): + """Hashes are not looked up and no request is made.""" + row, client = _lookup(_NOT_LISTED, "7657fcb7d772448a6d8504e4b20168b7", "md5_hash") + + check.is_false(row["Result"]) + check.equal(client.requests, []) + + +def test_empty_answer_is_not_a_hit(): + """A 200 with an empty body is reported as not found, not as a listing.""" + row, _ = _lookup({}, "github.com", "dns") + + check.is_false(row["Result"]) + check.equal(row["Details"], "Not found.") diff --git a/tests/context/test_tiproviders.py b/tests/context/test_tiproviders.py index 9173dc9b..15cd273b 100644 --- a/tests/context/test_tiproviders.py +++ b/tests/context/test_tiproviders.py @@ -183,6 +183,7 @@ class TiTestCase: TiTestCase("IPQualityScore", exp_responses=20), TiTestCase("BinaryEdge"), TiTestCase("ScanMalware"), + TiTestCase("NetAPI"), ] @@ -257,6 +258,7 @@ def verify_result(result, ti_lookup): "IPQualityScore", "BinaryEdge", "ScanMalware", + "NetAPI", ], ) check.is_not_none(lu_result["Ioc"]) @@ -697,6 +699,20 @@ def _scanmalware_scans(**kwargs): _PROVIDER_RESPONSES = { + "https://netapi.com/api-json/": { + "ioc_param": "params.value", + "response": { + "value": "badplace.net", + "type": "domain", + "listed_now": True, + "listed_ever": True, + "matched_value": "badplace.net", + "first_seen": "2026-09-01", + "last_seen": "2026-10-26 01:05:27 UTC", + "block_reason": "urlhaus.abuse.ch", + "registrar": None, + }, + }, "https://scanmalware.com/api/v1/domains": { "ioc_param": "url", "params": {}, diff --git a/tests/msticpyconfig-test.yaml b/tests/msticpyconfig-test.yaml index e62fb909..a19323b7 100644 --- a/tests/msticpyconfig-test.yaml +++ b/tests/msticpyconfig-test.yaml @@ -71,6 +71,9 @@ TIProviders: ScanMalware: Primary: True Provider: "ScanMalware" + NetAPI: + Primary: True + Provider: "NetAPI" OpenPageRank: Args: AuthKey: diff --git a/tests/testdata/msticpyconfig.yaml b/tests/testdata/msticpyconfig.yaml index bf7e53a4..ef729b81 100644 --- a/tests/testdata/msticpyconfig.yaml +++ b/tests/testdata/msticpyconfig.yaml @@ -43,6 +43,9 @@ TIProviders: ScanMalware: Primary: True Provider: "ScanMalware" + NetAPI: + Primary: True + Provider: "NetAPI" AzureSentinel: Args: WorkspaceID: "a927809c-8142-43e1-96b3-4ad87cfe95a3" From 37f5799339c35a48bdfa2a43fdf1f90716c169e2 Mon Sep 17 00:00:00 2001 From: Slava Olesik Date: Tue, 27 Oct 2026 12:42:58 +0200 Subject: [PATCH 2/3] NetAPI provider: look up domains, hostnames and URLs only --- msticpy/context/tiproviders/netapi.py | 10 ++-- tests/context/test_netapi.py | 66 +++++++++++++++++++++------ tests/context/test_tiproviders.py | 16 ------- 3 files changed, 55 insertions(+), 37 deletions(-) diff --git a/msticpy/context/tiproviders/netapi.py b/msticpy/context/tiproviders/netapi.py index edd6316f..9ee3248e 100644 --- a/msticpy/context/tiproviders/netapi.py +++ b/msticpy/context/tiproviders/netapi.py @@ -6,9 +6,9 @@ """ NetAPI Provider. -Looks up domains, hostnames, URLs and IP addresses in the NetAPI compromised -feed: the hosts that public threat feeds (URLhaus, CERT Polska, blackbook and -others) report for malware or phishing. The default query answers whether the +Looks up domains, hostnames and URLs in the NetAPI compromised feed: the +hosts that public threat feeds (CERT Polska, blackbook and others) report for +malware or phishing. The default query answers whether the observable is listed now (seen within the last 24 hours), was listed in the past only, or was never listed. The ``rank`` query subtype returns the position of a domain in the NetAPI Top 1M, a daily popularity ranking. @@ -55,13 +55,11 @@ class NetAPI(HttpTIProvider): _BASE_URL: ClassVar[str] = "https://netapi.com" # The feed matches a hostname or a URL through its registered domain, so - # every name type uses the same query. Addresses have no popularity rank. + # every name type uses the same query. _QUERIES: ClassVar[dict[str, APILookupParams]] = { "dns": _CHECK, "hostname": _CHECK, "url": _CHECK, - "ipv4": _CHECK, - "ipv6": _CHECK, "dns-rank": _RANK, "hostname-rank": _RANK, } diff --git a/tests/context/test_netapi.py b/tests/context/test_netapi.py index 83a93db2..bf6cde7f 100644 --- a/tests/context/test_netapi.py +++ b/tests/context/test_netapi.py @@ -5,29 +5,38 @@ # -------------------------------------------------------------------------- """NetAPI TI provider tests. -These cover what the shared parametrized harness in test_tiproviders.py -cannot reach: past and never-listed results, the registered-domain match, -the `rank` query subtype and the request parameters. +The provider looks up domains, hostnames and URLs only. The shared +parametrized harness in test_tiproviders.py looks up IP addresses in bulk, +so this provider is tested here: loading through TILookup, current, past +and never-listed results, the registered-domain match, the `rank` query +subtype and the request parameters. No network access. The provider's httpx client is replaced throughout. """ from __future__ import annotations +import warnings +from pathlib import Path from typing import Any import httpx import pytest import pytest_check as check +from msticpy.common import pkg_config +from msticpy.context.tilookup import TILookup from msticpy.context.tiproviders.netapi import NetAPI from msticpy.context.tiproviders.result_severity import ResultSeverity +from ..unit_test_lib import custom_mp_config, get_test_data_path + __author__ = "Slava Olesik" pytestmark = pytest.mark.filterwarnings("ignore::DeprecationWarning") -# Answers of the live API on 2026-10-26, trimmed to the fields the provider reads. +# Answers of the live API on 2026-10-26 and 2026-10-27, trimmed to the fields +# the provider reads. _LISTED_NOW: dict[str, Any] = { "value": "www.000359.xyz", "type": "domain", @@ -44,14 +53,14 @@ }, } _LISTED_PAST: dict[str, Any] = { - "value": "1.1.1.1", - "type": "ip", + "value": "0-07.ru", + "type": "domain", "listed_now": False, "listed_ever": True, - "matched_value": "1.1.1.1", - "first_seen": "2026-06-14", - "last_seen": "2026-06-17 01:05:02 UTC", - "block_reason": "urlhaus.abuse.ch", + "matched_value": "0-07.ru", + "first_seen": "2019-12-18", + "last_seen": "2019-12-25 00:00:00 UTC", + "block_reason": "suspicious", "registrar": None, } _NOT_LISTED: dict[str, Any] = { @@ -122,6 +131,25 @@ def _lookup( return result.to_dict(orient="records")[0], client +def test_lookup_through_tilookup(): + """The provider loads from the test config and answers through TILookup.""" + config_path = Path(get_test_data_path()).joinpath(pkg_config._CONFIG_FILE) + with custom_mp_config(config_path): + with warnings.catch_warnings(): + warnings.simplefilter("ignore", category=UserWarning) + ti_lookup = TILookup() + client = _Client(_LISTED_NOW) + ti_lookup.loaded_providers["NetAPI"]._httpx_client = client + + result = ti_lookup.lookup_ioc("www.000359.xyz", "hostname", providers=["NetAPI"]) + row = result.to_dict(orient="records")[0] + + check.equal(row["Provider"], "NetAPI") + check.is_true(row["Result"]) + check.equal(row["Severity"], ResultSeverity.high.name) + check.equal(len(client.requests), 1) + + def test_current_listing_is_high_severity(): """A host reported within the last 24 hours is a current listing.""" row, client = _lookup(_LISTED_NOW, "www.000359.xyz", "hostname") @@ -145,12 +173,12 @@ def test_past_listing_is_reported_but_not_rated(): It is a hit with its dates, but it must not carry the severity of a current listing. """ - row, _ = _lookup(_LISTED_PAST, "1.1.1.1", "ipv4") + row, _ = _lookup(_LISTED_PAST, "0-07.ru", "dns") check.is_true(row["Result"]) check.equal(row["Severity"], ResultSeverity.information.name) check.is_false(row["Details"]["Listed Now"]) - check.equal(row["Details"]["Last Seen"], "2026-06-17 01:05:02 UTC") + check.equal(row["Details"]["Last Seen"], "2019-12-25 00:00:00 UTC") def test_never_listed_is_a_negative_result(): @@ -210,9 +238,17 @@ def test_no_key_and_own_user_agent(): check.is_false(any(key.lower() == "authorization" for key in headers)) -def test_file_hashes_are_not_supported(): - """Hashes are not looked up and no request is made.""" - row, client = _lookup(_NOT_LISTED, "7657fcb7d772448a6d8504e4b20168b7", "md5_hash") +@pytest.mark.parametrize( + ("ioc", "ioc_type"), + [ + ("7657fcb7d772448a6d8504e4b20168b7", "md5_hash"), + ("185.92.220.35", "ipv4"), + ("2345:0425:2CA1:0000:0000:0567:5673:23b5", "ipv6"), + ], +) +def test_hashes_and_ip_addresses_are_not_supported(ioc, ioc_type): + """Hashes and IP addresses are not looked up and no request is made.""" + row, client = _lookup(_NOT_LISTED, ioc, ioc_type) check.is_false(row["Result"]) check.equal(client.requests, []) diff --git a/tests/context/test_tiproviders.py b/tests/context/test_tiproviders.py index 15cd273b..9173dc9b 100644 --- a/tests/context/test_tiproviders.py +++ b/tests/context/test_tiproviders.py @@ -183,7 +183,6 @@ class TiTestCase: TiTestCase("IPQualityScore", exp_responses=20), TiTestCase("BinaryEdge"), TiTestCase("ScanMalware"), - TiTestCase("NetAPI"), ] @@ -258,7 +257,6 @@ def verify_result(result, ti_lookup): "IPQualityScore", "BinaryEdge", "ScanMalware", - "NetAPI", ], ) check.is_not_none(lu_result["Ioc"]) @@ -699,20 +697,6 @@ def _scanmalware_scans(**kwargs): _PROVIDER_RESPONSES = { - "https://netapi.com/api-json/": { - "ioc_param": "params.value", - "response": { - "value": "badplace.net", - "type": "domain", - "listed_now": True, - "listed_ever": True, - "matched_value": "badplace.net", - "first_seen": "2026-09-01", - "last_seen": "2026-10-26 01:05:27 UTC", - "block_reason": "urlhaus.abuse.ch", - "registrar": None, - }, - }, "https://scanmalware.com/api/v1/domains": { "ioc_param": "url", "params": {}, From 5a736e7c7955dbac13da6495091767f3b9ab402f Mon Sep 17 00:00:00 2001 From: Slava Olesik Date: Wed, 28 Oct 2026 08:42:23 +0200 Subject: [PATCH 3/3] Put the NetAPI IP feed back: NetAPI keeps publishing it --- msticpy/context/tiproviders/netapi.py | 10 ++-- tests/context/test_netapi.py | 66 ++++++--------------------- tests/context/test_tiproviders.py | 16 +++++++ 3 files changed, 37 insertions(+), 55 deletions(-) diff --git a/msticpy/context/tiproviders/netapi.py b/msticpy/context/tiproviders/netapi.py index 9ee3248e..edd6316f 100644 --- a/msticpy/context/tiproviders/netapi.py +++ b/msticpy/context/tiproviders/netapi.py @@ -6,9 +6,9 @@ """ NetAPI Provider. -Looks up domains, hostnames and URLs in the NetAPI compromised feed: the -hosts that public threat feeds (CERT Polska, blackbook and others) report for -malware or phishing. The default query answers whether the +Looks up domains, hostnames, URLs and IP addresses in the NetAPI compromised +feed: the hosts that public threat feeds (URLhaus, CERT Polska, blackbook and +others) report for malware or phishing. The default query answers whether the observable is listed now (seen within the last 24 hours), was listed in the past only, or was never listed. The ``rank`` query subtype returns the position of a domain in the NetAPI Top 1M, a daily popularity ranking. @@ -55,11 +55,13 @@ class NetAPI(HttpTIProvider): _BASE_URL: ClassVar[str] = "https://netapi.com" # The feed matches a hostname or a URL through its registered domain, so - # every name type uses the same query. + # every name type uses the same query. Addresses have no popularity rank. _QUERIES: ClassVar[dict[str, APILookupParams]] = { "dns": _CHECK, "hostname": _CHECK, "url": _CHECK, + "ipv4": _CHECK, + "ipv6": _CHECK, "dns-rank": _RANK, "hostname-rank": _RANK, } diff --git a/tests/context/test_netapi.py b/tests/context/test_netapi.py index bf6cde7f..83a93db2 100644 --- a/tests/context/test_netapi.py +++ b/tests/context/test_netapi.py @@ -5,38 +5,29 @@ # -------------------------------------------------------------------------- """NetAPI TI provider tests. -The provider looks up domains, hostnames and URLs only. The shared -parametrized harness in test_tiproviders.py looks up IP addresses in bulk, -so this provider is tested here: loading through TILookup, current, past -and never-listed results, the registered-domain match, the `rank` query -subtype and the request parameters. +These cover what the shared parametrized harness in test_tiproviders.py +cannot reach: past and never-listed results, the registered-domain match, +the `rank` query subtype and the request parameters. No network access. The provider's httpx client is replaced throughout. """ from __future__ import annotations -import warnings -from pathlib import Path from typing import Any import httpx import pytest import pytest_check as check -from msticpy.common import pkg_config -from msticpy.context.tilookup import TILookup from msticpy.context.tiproviders.netapi import NetAPI from msticpy.context.tiproviders.result_severity import ResultSeverity -from ..unit_test_lib import custom_mp_config, get_test_data_path - __author__ = "Slava Olesik" pytestmark = pytest.mark.filterwarnings("ignore::DeprecationWarning") -# Answers of the live API on 2026-10-26 and 2026-10-27, trimmed to the fields -# the provider reads. +# Answers of the live API on 2026-10-26, trimmed to the fields the provider reads. _LISTED_NOW: dict[str, Any] = { "value": "www.000359.xyz", "type": "domain", @@ -53,14 +44,14 @@ }, } _LISTED_PAST: dict[str, Any] = { - "value": "0-07.ru", - "type": "domain", + "value": "1.1.1.1", + "type": "ip", "listed_now": False, "listed_ever": True, - "matched_value": "0-07.ru", - "first_seen": "2019-12-18", - "last_seen": "2019-12-25 00:00:00 UTC", - "block_reason": "suspicious", + "matched_value": "1.1.1.1", + "first_seen": "2026-06-14", + "last_seen": "2026-06-17 01:05:02 UTC", + "block_reason": "urlhaus.abuse.ch", "registrar": None, } _NOT_LISTED: dict[str, Any] = { @@ -131,25 +122,6 @@ def _lookup( return result.to_dict(orient="records")[0], client -def test_lookup_through_tilookup(): - """The provider loads from the test config and answers through TILookup.""" - config_path = Path(get_test_data_path()).joinpath(pkg_config._CONFIG_FILE) - with custom_mp_config(config_path): - with warnings.catch_warnings(): - warnings.simplefilter("ignore", category=UserWarning) - ti_lookup = TILookup() - client = _Client(_LISTED_NOW) - ti_lookup.loaded_providers["NetAPI"]._httpx_client = client - - result = ti_lookup.lookup_ioc("www.000359.xyz", "hostname", providers=["NetAPI"]) - row = result.to_dict(orient="records")[0] - - check.equal(row["Provider"], "NetAPI") - check.is_true(row["Result"]) - check.equal(row["Severity"], ResultSeverity.high.name) - check.equal(len(client.requests), 1) - - def test_current_listing_is_high_severity(): """A host reported within the last 24 hours is a current listing.""" row, client = _lookup(_LISTED_NOW, "www.000359.xyz", "hostname") @@ -173,12 +145,12 @@ def test_past_listing_is_reported_but_not_rated(): It is a hit with its dates, but it must not carry the severity of a current listing. """ - row, _ = _lookup(_LISTED_PAST, "0-07.ru", "dns") + row, _ = _lookup(_LISTED_PAST, "1.1.1.1", "ipv4") check.is_true(row["Result"]) check.equal(row["Severity"], ResultSeverity.information.name) check.is_false(row["Details"]["Listed Now"]) - check.equal(row["Details"]["Last Seen"], "2019-12-25 00:00:00 UTC") + check.equal(row["Details"]["Last Seen"], "2026-06-17 01:05:02 UTC") def test_never_listed_is_a_negative_result(): @@ -238,17 +210,9 @@ def test_no_key_and_own_user_agent(): check.is_false(any(key.lower() == "authorization" for key in headers)) -@pytest.mark.parametrize( - ("ioc", "ioc_type"), - [ - ("7657fcb7d772448a6d8504e4b20168b7", "md5_hash"), - ("185.92.220.35", "ipv4"), - ("2345:0425:2CA1:0000:0000:0567:5673:23b5", "ipv6"), - ], -) -def test_hashes_and_ip_addresses_are_not_supported(ioc, ioc_type): - """Hashes and IP addresses are not looked up and no request is made.""" - row, client = _lookup(_NOT_LISTED, ioc, ioc_type) +def test_file_hashes_are_not_supported(): + """Hashes are not looked up and no request is made.""" + row, client = _lookup(_NOT_LISTED, "7657fcb7d772448a6d8504e4b20168b7", "md5_hash") check.is_false(row["Result"]) check.equal(client.requests, []) diff --git a/tests/context/test_tiproviders.py b/tests/context/test_tiproviders.py index 9173dc9b..15cd273b 100644 --- a/tests/context/test_tiproviders.py +++ b/tests/context/test_tiproviders.py @@ -183,6 +183,7 @@ class TiTestCase: TiTestCase("IPQualityScore", exp_responses=20), TiTestCase("BinaryEdge"), TiTestCase("ScanMalware"), + TiTestCase("NetAPI"), ] @@ -257,6 +258,7 @@ def verify_result(result, ti_lookup): "IPQualityScore", "BinaryEdge", "ScanMalware", + "NetAPI", ], ) check.is_not_none(lu_result["Ioc"]) @@ -697,6 +699,20 @@ def _scanmalware_scans(**kwargs): _PROVIDER_RESPONSES = { + "https://netapi.com/api-json/": { + "ioc_param": "params.value", + "response": { + "value": "badplace.net", + "type": "domain", + "listed_now": True, + "listed_ever": True, + "matched_value": "badplace.net", + "first_seen": "2026-09-01", + "last_seen": "2026-10-26 01:05:27 UTC", + "block_reason": "urlhaus.abuse.ch", + "registrar": None, + }, + }, "https://scanmalware.com/api/v1/domains": { "ioc_param": "url", "params": {},