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"