diff --git a/pyproject.toml b/pyproject.toml index 03817648..682d705c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "polyswarm_api" -version = "4.5.0" +version = "4.6.0" description = "Client library to simplify interacting with the PolySwarm consumer API" readme = "README.md" requires-python = ">=3.10,<4" @@ -55,7 +55,7 @@ package-dir = { "" = "src" } where = ["src"] [tool.bumpversion] -current_version = "4.5.0" +current_version = "4.6.0" commit = true tag = false sign_tags = true diff --git a/specs/00-overview.md b/specs/00-overview.md index e42f02a5..4e79ba5f 100644 --- a/specs/00-overview.md +++ b/specs/00-overview.md @@ -83,6 +83,8 @@ polyswarm-api/ │ │ # return PolyswarmRequest descriptors. │ ├── exceptions.py # HAND-WRITTEN. Exception hierarchy. │ ├── settings.py # HAND-WRITTEN. Default URI, timeouts, etc. +│ ├── refang.py # HAND-WRITTEN. Pure IoC refanging helpers +│ │ # (refang_text / is_network_ioc / refang_ioc). │ ├── session.py # GENERATED from aio/session.py. │ │ # PolyswarmSession (httpx.Client wrapper). │ │ # .execute(request), .upload_file, .close diff --git a/specs/01-architecture.md b/specs/01-architecture.md index d9bea7a4..6ad231b9 100644 --- a/specs/01-architecture.md +++ b/specs/01-architecture.md @@ -23,6 +23,7 @@ How a call flows from the user's code through the SDK to the server and back. Co | `src/polyswarm_api/resources.py` | yes | Per-domain wrappers. Builders return `PolyswarmRequest`. | | `src/polyswarm_api/exceptions.py` | yes | Hierarchy. | | `src/polyswarm_api/settings.py` | yes | Default URI, timeouts, etc. | +| `src/polyswarm_api/refang.py` | yes | Pure IoC refanging (`refang_text`, `is_network_ioc`, `refang_ioc`). No I/O, shared by both transports, not unasync'd. The clients apply it at their edge when `refang_iocs` is on. Contract: `05-downstream-contract.md` §"IoC refanging". | | `src/polyswarm_api/aio/__init__.py` | yes | Re-exports. | | `src/polyswarm_api/aio/session.py` | yes (canonical async) | `AsyncPolyswarmSession`. | | `src/polyswarm_api/aio/api.py` | yes (canonical async) | `PolySwarmAsyncAPI` + endpoint methods. | diff --git a/specs/03-endpoints.md b/specs/03-endpoints.md index 4eb0c86d..12e4f857 100644 --- a/specs/03-endpoints.md +++ b/specs/03-endpoints.md @@ -205,6 +205,8 @@ refusal. | `notification_webhook_list()` | `Webhook.list` | | `report_template_list(is_default=None, **kwargs)` | `ReportTemplate.list` | +IoC inputs of `search_url`, `search_by_metadata` (`ips` / `urls` / `domains` only), `search_by_ioc` (`ip` / `domain`) and `check_known_hosts` are refanged before the builder runs when the client was constructed with `refang_iocs=True` (opt-in; off by default) — see [`05-downstream-contract.md`](./05-downstream-contract.md) §"IoC refanging". The builders themselves are unchanged and never refang. The known-host writes (`add_known_good_host` / `add_known_bad_host` / `update_known_good_host`) refang their `host` the same way. + ## Special methods | Method | Why it's special | @@ -262,6 +264,8 @@ async def submit(self, artifact, ...): Generated (`api.py`) is the same with `await`/`async` lowered. (`sandbox_file` / `sandbox_url` follow the same create → `upload_file` → finalize shape, finalizing via `_finalize_sandbox_task`.) +A URL passed as a string to `submit` / `sandbox_file` (with `artifact_type=URL`) or as `sandbox_url(url)` is refanged before `LocalArtifact.from_content` when `refang_iocs` is on, so the uploaded content and the default artifact name both carry the live URL (§"IoC refanging" in [`05-downstream-contract.md`](./05-downstream-contract.md)). A QR-code submission (`preprocessing={'type': 'qrcode'}`) is the exception on both `submit` and `sandbox_file`: its argument names an image file, not a URL, and is passed on unchanged. + `upload_file` is a method on the session class (`AsyncPolyswarmSession.upload_file` / `PolyswarmSession.upload_file`). Both strip the session-level `Authorization` header so the PolySwarm API key doesn't leak to the pre-signed S3 origin. Downstream consumers customize behaviour by subclassing the session — see [`05-downstream-contract.md`](./05-downstream-contract.md) §"Customizing transport behaviour". **Report-template logo upload is different**: `report_template_logo_upload` PUTs to the PolySwarm endpoint `/reports/templates/logo`, which is authenticated, *not* a pre-signed S3 URL. It builds a normal `PolyswarmRequest` descriptor via `ReportTemplate.upload_logo(...)` and dispatches it through `session.execute` like any other endpoint — the API key must ride along on the request. There is no `session.upload_logo` method. diff --git a/specs/04-testing.md b/specs/04-testing.md index 79dee474..19d7c6b2 100644 --- a/specs/04-testing.md +++ b/specs/04-testing.md @@ -30,6 +30,7 @@ How the test suite is organised. Three layers: pure unit tests (no HTTP at all - `test/vcr/*.vcr` — recorded cassettes. - `test/malicious` — fixture file for upload tests (`test/eicar.yara` was retired when the rules tests moved to per-test `uid_yara` bodies). - `test/hunt_tracking_builder_test.py` — pure-unit request-shape and parse tests for the hunt-page tracking builders/resources. +- `test/refang_test.py` — pure-unit tests for `polyswarm_api.refang` (driven by the shared case table `test/fixtures/refang_cases.json`, kept byte-identical with the other clients that implement the same contract) and request-shape tests for every client method that refangs its IoC inputs, captured at the `_paginate` / `_single` boundary for both transports. Pure-unit because this is client-side input normalization: the server contract is unchanged, so there is no new endpoint behaviour for a cassette to pin. - `test/ruleset_favorite_respx_test.py` — dual-transport (`ClientTestCase`) respx suite for the favorite toggle: the `FAVORITE_LIMIT` refusal envelope and the query/body split. ## Three test layers diff --git a/specs/05-downstream-contract.md b/specs/05-downstream-contract.md index 5bbc461a..271c6db6 100644 --- a/specs/05-downstream-contract.md +++ b/specs/05-downstream-contract.md @@ -37,6 +37,7 @@ __version__: str __release_url__: str api: module # contains PolyswarmAPI exceptions: module # contains the exception hierarchy +refang: module # IoC refanging helpers — see "IoC refanging" below PolyswarmAPI: class # sync client PolySwarmAsyncAPI: class # async client (re-exported from polyswarm_api.aio) ``` @@ -48,7 +49,7 @@ PolySwarmAsyncAPI: class # async client (re-exported from polyswarm_ ```python class PolyswarmAPI: def __init__(self, key=None, uri=None, community=None, timeout=None, - verify=True, *, session=None, **httpx_kwargs): ... + verify=True, *, session=None, refang_iocs=False, **httpx_kwargs): ... def close(self): ... def __enter__(self): ... def __exit__(self, *exc): ... @@ -62,7 +63,7 @@ Generated by `scripts/regenerate_sync.py` from `polyswarm_api/aio/api.py`. Publi ```python class PolySwarmAsyncAPI: def __init__(self, key=None, uri=None, community=None, timeout=None, - verify=True, *, session=None, **httpx_kwargs): ... + verify=True, *, session=None, refang_iocs=False, **httpx_kwargs): ... async def aclose(self): ... async def __aenter__(self): ... async def __aexit__(self, *exc): ... @@ -202,6 +203,7 @@ PolyswarmAPI( verify: bool = True, *, session: PolyswarmSession | None = None, # pre-built session; mutually exclusive with `**httpx_kwargs` + refang_iocs: bool = False, # opt in to refanging defanged URL/domain/IP inputs — see "IoC refanging" **httpx_kwargs, # forwarded to httpx.Client when constructing default session ) ``` @@ -212,6 +214,24 @@ Same shape for `PolySwarmAsyncAPI` with `AsyncPolyswarmSession`. `**httpx_kwargs` is forwarded to `httpx.{,Async}Client` (formerly to `requests.Session` in 3.x). Kwargs that worked on both (`timeout`, `verify`, `headers`) are unchanged. `requests`-only kwargs (e.g. `proxies` as a dict-of-protocol-strings) need translation to httpx's equivalent. +## IoC refanging + +Threat-intel reports print indicators defanged (`hxxps[:]//evil[.]com`, `127[.]0[.]0[.]1`). The server looks URLs up by an exact hash of the string and stores a submitted URL verbatim, so a defanged value silently misses a search or becomes a broken URL artifact — and the server deliberately does not guess. The SDK therefore offers to refang at its own edge (added in 4.6.0, opt-in). + +**`polyswarm_api.refang`** (public, pure, no I/O): + +| Function | Contract | +|---|---| +| `refang_text(text)` | Every rewrite, ungated and untrimmed: `[://]`→`://`, `[:]`→`:`, `[/]`→`/`, `[.]` `(.)` `{.}` `[dot]` `(dot)` `{dot}`→`.` (inner whitespace allowed), then the anchored schemes `hxxps`/`h**ps`→`https`, `hxxp`/`h**p`→`http`, `fxps`→`ftps`, `fxp`→`ftp`. | +| `is_network_ioc(candidate)` | URL (optional http(s)/ftp(s) scheme, userinfo, port, path), domain, IPv4, or bracketed IPv6 host. | +| `refang_ioc(value, accept=None)` | The gated form: returns the refanged value only when something was defanged, the result has no whitespace or `"`, the rewrite does not keep the input's scheme and host intact (if it does, only a path would change, so `example.com/a[.]b` and `https://example.com/a[.]b` both survive), the result is a network IoC, and the optional `accept(candidate)` agrees. Otherwise returns `value` unchanged (untrimmed). Non-strings pass through. | + +Invariants: the rules, their order, the gate and the case table (`test/fixtures/refang_cases.json`) are a contract other PolySwarm clients implement too — the table is kept byte-identical across them, so changing a rule means changing it everywhere. Out of scope: email `[at]`, bare-word ` dot `, `http__host` / `http:\\host`, bare-bracket stripping (IPv6 syntax), non-ASCII hosts. + +**Where the client applies it** (when constructed with `refang_iocs=True`; the default is `False`): `search_url(url)`; `search_by_metadata(ips=, urls=, domains=)` — **never** the free-form `query`; `search_by_ioc(ip=, domain=)`; `check_known_hosts(ips=, domains=)`; the `host` of `add_known_good_host` / `add_known_bad_host` / `update_known_good_host`; and the URL of every URL submission (`submit(..., artifact_type=URL)` and `sandbox_file(..., artifact_type=URL)` from a string, `sandbox_url(url)`) — both the uploaded content and the default artifact name. An explicit `artifact_name` is kept as given. Hashes and ids are never touched, and neither is the argument of a QR-code submission (`preprocessing={'type': 'qrcode'}` on `submit` or `sandbox_file`): it names an image file, not a URL, so a file called `qr[.]png` stays `qr[.]png`. With the default `refang_iocs=False` every input is sent verbatim, exactly as in 4.5.0. The setting is kept as the public attribute `api.refang_iocs`, so a consumer that handles a value outside the endpoint methods (its own validation, or a request it builds with `_single`) can apply `refang.refang_ioc` under the same switch — the CLI does exactly that. + +Compatibility: refanging is **opt-in**, so the default preserves the 4.5.0 behaviour exactly and 4.6.0 falls under the "new optional keyword argument … with a default that preserves the current behaviour" row of the versioning table (minor). The default matters beyond searches: turning refanging on changes what the write paths store — a known-host row written verbatim as `evil[.]com` by an earlier client is no longer matched by a lookup that now sends `evil.com`, and a URL submission creates a different artifact (content, name, sha) than the same defanged call did before. A consumer opts in knowing that; the CLI does so by default through its `--refang/--no-refang` option. + ## Customizing transport behaviour Replaces the 3.x module-level monkey-patching pattern with subclassing. diff --git a/src/polyswarm_api/__init__.py b/src/polyswarm_api/__init__.py index d9cf1695..e80be0e3 100644 --- a/src/polyswarm_api/__init__.py +++ b/src/polyswarm_api/__init__.py @@ -1,9 +1,10 @@ # https://www.python.org/dev/peps/pep-0008/#module-level-dunder-names -__version__ = '4.5.0' +__version__ = '4.6.0' __release_url__ = 'https://api.github.com/repos/polyswarm/polyswarm-api/releases/latest' from . import api from . import exceptions +from . import refang from .api import PolyswarmAPI from .session import PolyswarmSession from .aio import PolySwarmAsyncAPI, AsyncPolyswarmSession diff --git a/src/polyswarm_api/aio/api.py b/src/polyswarm_api/aio/api.py index d9377401..7cbc1ab9 100644 --- a/src/polyswarm_api/aio/api.py +++ b/src/polyswarm_api/aio/api.py @@ -17,7 +17,7 @@ import logging import time -from polyswarm_api import exceptions, resources, settings +from polyswarm_api import exceptions, refang, resources, settings from polyswarm_api.core import PolyswarmRequest, _as_result_bound from .session import AsyncPolyswarmSession @@ -53,6 +53,7 @@ def __init__( verify: bool = True, *, session: AsyncPolyswarmSession | None = None, + refang_iocs: bool = False, **httpx_kwargs, ): key_masked = '******' + (key[-4:] if key and len(key) > 16 else '') @@ -64,6 +65,11 @@ def __init__( self.community = community or settings.DEFAULT_COMMUNITY self.timeout = timeout or settings.DEFAULT_HTTP_TIMEOUT self.verify = verify + # Refang defanged URL / domain / IP inputs (``hxxps[:]//evil[.]com``) + # before building a request. Opt-in: the default preserves the 4.5.0 + # behaviour of sending every input verbatim. See ``polyswarm_api.refang`` + # and ``_refang`` below for exactly which inputs are touched. + self.refang_iocs = refang_iocs self._engines = None # Either accept a pre-built session (customization point) or # build the default from ``key``. Passing both is ambiguous. @@ -88,6 +94,27 @@ def __repr__(self): attrs = f'uri={self.uri!r}, community={self.community!r}, timeout={self.timeout!r}' return f'<{clsname}({attrs}) at 0x{id(self):x}>' + def _refang(self, value): + """Refang one URL / domain / IP input when ``refang_iocs`` is on. + + Applied only to arguments that name a network indicator (a URL to + search or submit, an ``ips=`` / ``urls=`` / ``domains=`` entry, an IoC + ``ip`` / ``domain``, a known-host ``host``) — never to a free-form + metadata query or a hash. + Anything that is not a defanged network IoC is returned unchanged. + """ + if not self.refang_iocs: + return value + return refang.refang_ioc(value) + + def _refang_all(self, values): + """``_refang`` over a list argument; ``None`` / empty pass through.""" + if not values or not self.refang_iocs: + return values + if isinstance(values, str): + return self._refang(values) + return [self._refang(v) for v in values] + async def aclose(self): """Close the underlying HTTP client.""" await self.session.aclose() @@ -340,6 +367,7 @@ async def search_url(self, url): :param url: A url to be searched by exact match :return: Generator of ArtifactInstance resources """ + url = self._refang(url) logger.info('Searching for url %s', url) async for item in self._paginate(resources.ArtifactInstance.search_url(self, url)): yield item @@ -365,6 +393,7 @@ async def search_by_metadata(self, query, include=None, exclude=None, ips=None, :param exclude: A list of fields to be excluded from the result (.* wildcards are accepted) :return: Generator of ArtifactInstance resources """ + ips, urls, domains = self._refang_all(ips), self._refang_all(urls), self._refang_all(domains) logger.info('Searching for metadata %s', query) async for item in self._paginate(resources.Metadata.get(self, query=query, community=self.community, include=include, exclude=exclude, ips=ips, urls=urls, domains=domains)): yield item @@ -391,6 +420,7 @@ async def search_by_ioc(self, ip=None, domain=None, ttp=None, imphash=None): :param imphash: ImpHash to search by :return: Generator of ArtifactInstance resources """ + ip, domain = self._refang(ip), self._refang(domain) logger.info('Searching by ioc %s', dict(ip=ip, domain=domain, ttp=ttp, imphash=imphash)) async for item in self._paginate(resources.IOC.ioc_search(self, ip=ip, domain=domain, ttp=ttp, imphash=imphash)): yield item @@ -403,6 +433,7 @@ async def check_known_hosts(self, ips=[], domains=[]): :param domains :return: Generator of IOC resources """ + ips, domains = self._refang_all(ips), self._refang_all(domains) logger.info('Checking known hosts ips: %s, domains: %s', ips, domains) async for item in self._paginate(resources.IOC.check_known_hosts(self, ips, domains)): yield item @@ -416,6 +447,7 @@ async def add_known_good_host(self, type, source, host): :param host :return: IOC resource """ + host = self._refang(host) logger.info('Creating known good ioc %s %s %s', type, host, source) return await self._single(resources.IOC.create_known_good(self, type, host, source)) @@ -428,6 +460,7 @@ async def add_known_bad_host(self, type, source, host): :param host :return: IOC resource """ + host = self._refang(host) logger.info('Creating known bad ioc %s %s %s', type, host, source) return await self._single(resources.IOC.create_known_bad(self, type, host, source)) @@ -440,6 +473,7 @@ async def update_known_good_host(self, id, type, source, host, good): :param host :return: IOC resource """ + host = self._refang(host) logger.info('Updating known good ioc %s %s %s %s', id, type, host, source) return await self._single(resources.IOC.update_known_good(self, id, type, host, source, good)) @@ -1373,11 +1407,12 @@ async def submit( self, artifact, artifact_type=artifact_type, artifact_name=artifact_name ) elif artifact_type == resources.ArtifactType.URL: - if preprocessing and preprocessing["type"] == "qrcode": + if preprocessing and preprocessing.get("type") == "qrcode": artifact = resources.LocalArtifact.from_path( self, artifact, artifact_type=artifact_type, artifact_name=artifact_name ) else: + artifact = self._refang(artifact) artifact = resources.LocalArtifact.from_content( self, artifact, artifact_name=artifact_name or artifact, artifact_type=artifact_type, @@ -1480,6 +1515,10 @@ async def sandbox_file( self, artifact, artifact_type=artifact_type, artifact_name=artifact_name ) elif artifact_type == resources.ArtifactType.URL: + # A QR-code submission's argument is an image path, not a URL + # (same rule as ``submit``), so it is never refanged. + if not (preprocessing and preprocessing.get("type") == "qrcode"): + artifact = self._refang(artifact) artifact = resources.LocalArtifact.from_content( self, artifact, artifact_name=artifact_name or artifact, artifact_type=artifact_type, @@ -1542,6 +1581,7 @@ async def sandbox_url( else: local = artifact else: + url = self._refang(url) local = resources.LocalArtifact.from_content( self, url, artifact_name=artifact_name or url, artifact_type=resources.ArtifactType.URL, diff --git a/src/polyswarm_api/api.py b/src/polyswarm_api/api.py index 3f98cc5d..3ac8beda 100644 --- a/src/polyswarm_api/api.py +++ b/src/polyswarm_api/api.py @@ -19,7 +19,7 @@ import io import logging -from polyswarm_api import exceptions, resources, settings +from polyswarm_api import exceptions, refang, resources, settings from polyswarm_api.core import PolyswarmRequest, _as_result_bound from .session import PolyswarmSession @@ -55,6 +55,7 @@ def __init__( verify: bool = True, *, session: PolyswarmSession | None = None, + refang_iocs: bool = False, **httpx_kwargs, ): key_masked = "******" + (key[-4:] if key and len(key) > 16 else "") @@ -69,6 +70,11 @@ def __init__( self.community = community or settings.DEFAULT_COMMUNITY self.timeout = timeout or settings.DEFAULT_HTTP_TIMEOUT self.verify = verify + # Refang defanged URL / domain / IP inputs (``hxxps[:]//evil[.]com``) + # before building a request. Opt-in: the default preserves the 4.5.0 + # behaviour of sending every input verbatim. See ``polyswarm_api.refang`` + # and ``_refang`` below for exactly which inputs are touched. + self.refang_iocs = refang_iocs self._engines = None # Either accept a pre-built session (customization point) or # build the default from ``key``. Passing both is ambiguous. @@ -95,6 +101,27 @@ def __repr__(self): ) return f"<{clsname}({attrs}) at 0x{id(self):x}>" + def _refang(self, value): + """Refang one URL / domain / IP input when ``refang_iocs`` is on. + + Applied only to arguments that name a network indicator (a URL to + search or submit, an ``ips=`` / ``urls=`` / ``domains=`` entry, an IoC + ``ip`` / ``domain``, a known-host ``host``) — never to a free-form + metadata query or a hash. + Anything that is not a defanged network IoC is returned unchanged. + """ + if not self.refang_iocs: + return value + return refang.refang_ioc(value) + + def _refang_all(self, values): + """``_refang`` over a list argument; ``None`` / empty pass through.""" + if not values or not self.refang_iocs: + return values + if isinstance(values, str): + return self._refang(values) + return [self._refang(v) for v in values] + def close(self): """Close the underlying HTTP client.""" self.session.close() @@ -366,6 +393,7 @@ def search_url(self, url): :param url: A url to be searched by exact match :return: Generator of ArtifactInstance resources """ + url = self._refang(url) logger.info("Searching for url %s", url) for item in self._paginate(resources.ArtifactInstance.search_url(self, url)): yield item @@ -395,6 +423,11 @@ def search_by_metadata( :param exclude: A list of fields to be excluded from the result (.* wildcards are accepted) :return: Generator of ArtifactInstance resources """ + ips, urls, domains = ( + self._refang_all(ips), + self._refang_all(urls), + self._refang_all(domains), + ) logger.info("Searching for metadata %s", query) for item in self._paginate( resources.Metadata.get( @@ -436,6 +469,7 @@ def search_by_ioc(self, ip=None, domain=None, ttp=None, imphash=None): :param imphash: ImpHash to search by :return: Generator of ArtifactInstance resources """ + ip, domain = self._refang(ip), self._refang(domain) logger.info( "Searching by ioc %s", dict(ip=ip, domain=domain, ttp=ttp, imphash=imphash) ) @@ -454,6 +488,7 @@ def check_known_hosts(self, ips=[], domains=[]): :param domains :return: Generator of IOC resources """ + ips, domains = self._refang_all(ips), self._refang_all(domains) logger.info("Checking known hosts ips: %s, domains: %s", ips, domains) for item in self._paginate(resources.IOC.check_known_hosts(self, ips, domains)): yield item @@ -467,6 +502,7 @@ def add_known_good_host(self, type, source, host): :param host :return: IOC resource """ + host = self._refang(host) logger.info("Creating known good ioc %s %s %s", type, host, source) return self._single(resources.IOC.create_known_good(self, type, host, source)) @@ -479,6 +515,7 @@ def add_known_bad_host(self, type, source, host): :param host :return: IOC resource """ + host = self._refang(host) logger.info("Creating known bad ioc %s %s %s", type, host, source) return self._single(resources.IOC.create_known_bad(self, type, host, source)) @@ -491,6 +528,7 @@ def update_known_good_host(self, id, type, source, host, good): :param host :return: IOC resource """ + host = self._refang(host) logger.info("Updating known good ioc %s %s %s %s", id, type, host, source) return self._single( resources.IOC.update_known_good(self, id, type, host, source, good) @@ -1700,7 +1738,7 @@ def submit( artifact_name=artifact_name, ) elif artifact_type == resources.ArtifactType.URL: - if preprocessing and preprocessing["type"] == "qrcode": + if preprocessing and preprocessing.get("type") == "qrcode": artifact = resources.LocalArtifact.from_path( self, artifact, @@ -1708,6 +1746,7 @@ def submit( artifact_name=artifact_name, ) else: + artifact = self._refang(artifact) artifact = resources.LocalArtifact.from_content( self, artifact, @@ -1822,6 +1861,10 @@ def sandbox_file( artifact_name=artifact_name, ) elif artifact_type == resources.ArtifactType.URL: + # A QR-code submission's argument is an image path, not a URL + # (same rule as ``submit``), so it is never refanged. + if not (preprocessing and preprocessing.get("type") == "qrcode"): + artifact = self._refang(artifact) artifact = resources.LocalArtifact.from_content( self, artifact, @@ -1896,6 +1939,7 @@ def sandbox_url( else: local = artifact else: + url = self._refang(url) local = resources.LocalArtifact.from_content( self, url, diff --git a/src/polyswarm_api/refang.py b/src/polyswarm_api/refang.py new file mode 100644 index 00000000..fb77205b --- /dev/null +++ b/src/polyswarm_api/refang.py @@ -0,0 +1,125 @@ +"""Refang defanged indicators of compromise (IoCs). + +Threat-intel reports print every indicator defanged — ``hxxps[:]//evil[.]com``, +``127[.]0[.]0[.]1`` — so that nobody clicks it by accident. Pasted as-is into a +search or a URL submission it never matches anything: the server looks URLs up +by an exact hash of the string and stores a submitted URL verbatim, so a +defanged value silently misses (search) or becomes a new, broken URL artifact +(submission). The server deliberately does not guess, so clients refang at their +own edge, before the request is built. + +This module is the SDK's implementation of a contract other PolySwarm clients +implement too: same rules, same order, same gate, same case table +(``test/fixtures/refang_cases.json``, kept byte-identical across clients). + +Portability is part of that contract, because the same pattern can match +different characters in different regex engines. So: no case-insensitive +flag (Python's folds Unicode, e.g. U+212A KELVIN SIGN matches ``k``) — letters +are spelled as explicit ``[aA]`` classes; no ``\\b``, ``\\d``, ``\\w``, ``\\s`` +or ``\\S`` (their Unicode sets differ between engines) — whitespace is the +explicit ASCII set ``[ \\t\\n\\r\\f\\v]``, and trimming strips only that set; +and full matches use ``fullmatch`` (Python's ``$`` also matches before a +trailing newline). + +Out of scope, everywhere: email ``[at]``, a bare-word `` dot ``, ``http__host`` +and ``http:\\\\host`` variants, stripping bare brackets (they are IPv6 literal +syntax), and non-ASCII (IDN) hosts. +""" + +import re + +__all__ = ['is_network_ioc', 'refang_ioc', 'refang_text'] + +# ASCII whitespace only — see the module docstring on portability. +_WS_CHARS = ' \t\n\r\f\v' +_WS = r'[ \t\n\r\f\v]' + +# Applied in order. The bracket rules run first so that ``hxxps[:]//`` has +# become ``hxxps://`` by the time the (anchored) scheme rules look for ``://``. +# ``hxxps`` is matched before ``hxxp``: the shorter rule would otherwise leave +# a stray ``s`` behind. Each scheme rule is a literal replacement — a capture +# group would carry the input's case (``HXXPS`` -> ``httpS``). +_RULES = ( + (re.compile(rf'[\[({{]{_WS}*:{_WS}*/{_WS}*/{_WS}*[\])}}]'), '://'), + (re.compile(rf'[\[({{]{_WS}*:{_WS}*[\])}}]'), ':'), + (re.compile(rf'[\[({{]{_WS}*/{_WS}*[\])}}]'), '/'), + (re.compile(rf'[\[({{]{_WS}*\.{_WS}*[\])}}]'), '.'), + (re.compile(rf'[\[({{]{_WS}*[dD][oO][tT]{_WS}*[\])}}]'), '.'), + (re.compile(r'^[hH][xX*]{2}[pP][sS](?=://)'), 'https'), + (re.compile(r'^[hH][xX*]{2}[pP](?=://)'), 'http'), + (re.compile(r'^[fF][xX][pP][sS](?=://)'), 'ftps'), + (re.compile(r'^[fF][xX][pP](?=://)'), 'ftp'), +) + +_OCTET = r'(?:25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])' +_IPV4 = rf'{_OCTET}(?:\.{_OCTET}){{3}}' +_LABEL = r'[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?' +_TLD = r'(?:[a-zA-Z]{2,63}|[xX][nN]--[a-zA-Z0-9-]{1,59})' +_DOMAIN = rf'(?:{_LABEL}\.)+{_TLD}' +_IPV6 = r'\[[0-9a-fA-F:.]+\]' +_HOST = rf'(?:{_DOMAIN}|{_IPV4}|{_IPV6})' +_SCHEME = r'(?:[hH][tT][tT][pP][sS]?|[fF][tT][pP][sS]?)' +# Used with ``fullmatch`` — never ``^…$``, whose ``$`` accepts a trailing ``\n``. +_NETWORK_IOC = re.compile( + rf'(?:{_SCHEME}://)?(?:[^ \t\n\r\f\v/?#@]+@)?{_HOST}(?::[0-9]{{1,5}})?(?:[/?#][^ \t\n\r\f\v]*)?' +) +# An optional scheme-like token (live or defanged, e.g. ``hxxps``) plus the +# host: everything before the first ``/``, ``?`` or ``#``. +_SCHEME_AND_HOST = re.compile(r'^(?:[a-zA-Z*]+://)?[^/?#]*') +_QUERY_SYNTAX = re.compile(r'[ \t\n\r\f\v"]') + + +def refang_text(text): + """Apply every refang rewrite to ``text``, with no trimming and no gate. + + Ungated: it rewrites anything that contains a defang token, IoC or not. + Callers that take user input want :func:`refang_ioc` instead. + """ + for pattern, replacement in _RULES: + text = pattern.sub(replacement, text) + return text + + +def is_network_ioc(candidate): + """Whether ``candidate`` is a URL, a domain, or an IP address. + + A URL may carry an http(s)/ftp(s) scheme, userinfo, a port and a path; an + IPv6 host is accepted only in its bracketed URL form. + """ + return bool(_NETWORK_IOC.fullmatch(candidate)) + + +def refang_ioc(value, accept=None): + """Return the refanged form of ``value`` if it is a defanged network IoC. + + Anything else comes back exactly as given (untrimmed), so this is safe to + run on every IoC-shaped input. The gate, in order: + + * nothing was defanged -> unchanged; + * the rewrite contains ASCII whitespace or a double quote -> unchanged: that is + a query or quoted data, never a single indicator; + * the rewrite keeps the input's scheme and host intact (with or without + a scheme) -> unchanged: only a path, query or fragment would change, + so a legitimate ``[.]`` in ``example.com/a[.]b`` survives; + * the rewrite is not a URL, domain or IP -> unchanged; + * ``accept`` (optional) rejects the rewrite -> unchanged. Callers use it + to require their own routing to agree, e.g. "this would be searched as + a URL". + + A refanged value is returned trimmed of ASCII whitespace. + """ + if not isinstance(value, str): + return value + trimmed = value.strip(_WS_CHARS) + candidate = refang_text(trimmed) + if candidate == trimmed: + return value + if _QUERY_SYNTAX.search(candidate): + return value + if candidate.startswith(_SCHEME_AND_HOST.match(trimmed).group(0)): + return value + if not is_network_ioc(candidate): + return value + if accept is not None and not accept(candidate): + return value + return candidate diff --git a/test/fixtures/refang_cases.json b/test/fixtures/refang_cases.json new file mode 100644 index 00000000..9add92a3 --- /dev/null +++ b/test/fixtures/refang_cases.json @@ -0,0 +1,44 @@ +[ + {"input": "hxxps[:]//evil[.]com", "expected": "https://evil.com", "why": "bracketed colon, bracketed dot, hxxps scheme"}, + {"input": "hxxp://evil.com", "expected": "http://evil.com", "why": "defanged scheme only"}, + {"input": "hXXps://evil.com/path?q=1", "expected": "https://evil.com/path?q=1", "why": "scheme rule is case-insensitive; path and query kept"}, + {"input": "h**p://evil.com", "expected": "http://evil.com", "why": "asterisk scheme variant"}, + {"input": "HXXPS[:]//EVIL[.]COM", "expected": "https://EVIL.COM", "why": "scheme is normalized, host case is kept"}, + {"input": "hxxps[://]evil[.]com", "expected": "https://evil.com", "why": "bracketed scheme separator"}, + {"input": "fxp://files[.]example[.]org", "expected": "ftp://files.example.org", "why": "fxp scheme"}, + {"input": "fxps://files.example.org", "expected": "ftps://files.example.org", "why": "fxps scheme"}, + {"input": "127[.]0[.]0[.]1", "expected": "127.0.0.1", "why": "defanged IPv4"}, + {"input": "192(.)168(.)100(.)200", "expected": "192.168.100.200", "why": "parenthesized dots"}, + {"input": "evil[.]com", "expected": "evil.com", "why": "defanged bare domain"}, + {"input": "evil[dot]com", "expected": "evil.com", "why": "bracketed dot word"}, + {"input": "evil(DOT)com", "expected": "evil.com", "why": "dot word is case-insensitive"}, + {"input": "evil{.}com", "expected": "evil.com", "why": "braced dot"}, + {"input": "evil[ . ]com", "expected": "evil.com", "why": "whitespace inside the brackets is part of the token"}, + {"input": "http://evil[.]com[/]payload", "expected": "http://evil.com/payload", "why": "live scheme but defanged host; bracketed slash"}, + {"input": "evil[.]com:8080/x", "expected": "evil.com:8080/x", "why": "port and path kept"}, + {"input": " hxxp://evil[.]com ", "expected": "http://evil.com", "why": "surrounding whitespace is trimmed when refanged"}, + {"input": "xn--80ak6aa92e[.]com", "expected": "xn--80ak6aa92e.com", "why": "punycode label"}, + {"input": "evil[.]xn--p1ai", "expected": "evil.xn--p1ai", "why": "punycode TLD"}, + {"input": "http://[2001:db8::1]:8080/", "expected": "http://[2001:db8::1]:8080/", "why": "IPv6 literal brackets are never touched"}, + {"input": "https://example.com/a[.]b", "expected": "https://example.com/a[.]b", "why": "a live URL with a clean host is never rewritten"}, + {"input": "example.com/a[.]b", "expected": "example.com/a[.]b", "why": "a scheme-less URL whose host is already live is never rewritten (only its path would change)"}, + {"input": "evil.com[/]payload", "expected": "evil.com/payload", "why": "a bracketed slash right after a live host is still a defang token"}, + {"input": "hxxp://example.com/a[.]b", "expected": "http://example.com/a.b", "why": "a defanged scheme makes the whole value a defanged IoC, path included"}, + {"input": "evil.com", "expected": "evil.com", "why": "nothing to refang"}, + {"input": "http://evil.com", "expected": "http://evil.com", "why": "nothing to refang"}, + {"input": "hash[.]sha256", "expected": "hash[.]sha256", "why": "result is not a network IoC (TLD has digits)"}, + {"input": "triage_sandbox_v0[.]extracted", "expected": "triage_sandbox_v0[.]extracted", "why": "result is not a network IoC (underscore in label)"}, + {"input": "field:\"evil[.]com\"", "expected": "field:\"evil[.]com\"", "why": "quoted values are data, never rewritten"}, + {"input": "evil[.]com OR bad[.]org", "expected": "evil[.]com OR bad[.]org", "why": "multi-token input is a query, never rewritten"}, + {"input": "user[at]example[.]com", "expected": "user[at]example[.]com", "why": "email refanging is out of scope"}, + {"input": "evil[.]", "expected": "evil[.]", "why": "result is not a network IoC"}, + {"input": "999[.]1[.]1[.]1", "expected": "999[.]1[.]1[.]1", "why": "result is not a valid IPv4 nor a domain"}, + {"input": "hxxp", "expected": "hxxp", "why": "a bare scheme word is not rewritten"}, + {"input": "d41d8cd98f00b204e9800998ecf8427e", "expected": "d41d8cd98f00b204e9800998ecf8427e", "why": "hashes pass through"}, + {"input": "evil[.]\u212aom", "expected": "evil[.]\u212aom", "why": "non-ASCII letters never match a letter class (U+212A KELVIN SIGN is not k)"}, + {"input": "evil[.]\u017fite", "expected": "evil[.]\u017fite", "why": "non-ASCII letters never match a letter class (U+017F LONG S is not s)"}, + {"input": "evil[.]com\n", "expected": "evil.com", "why": "a trailing ASCII newline is trimmed"}, + {"input": "evil[.]com\u00a0", "expected": "evil[.]com\u00a0", "why": "trim covers ASCII whitespace only; a trailing NBSP is not a host character"}, + {"input": "EVIL[DOT]COM", "expected": "EVIL.COM", "why": "dot word and host letters in upper case"}, + {"input": "", "expected": "", "why": "empty input"} +] diff --git a/test/refang_test.py b/test/refang_test.py new file mode 100644 index 00000000..bd324017 --- /dev/null +++ b/test/refang_test.py @@ -0,0 +1,407 @@ +"""Pure-unit tests for IoC refanging (``polyswarm_api.refang``) and for the +client methods that apply it before building a request. + +No HTTP at all (the pure-unit tier — see specs/04-testing.md). This is input +normalization, not a new endpoint: the server contract is unchanged, so what +needs pinning is (a) the refang function itself and (b) the request shape each +client method builds — that the outgoing params/body carry the refanged value +when ``refang_iocs`` is on, and the raw value when it is off (the default). +The request is captured at the ``_paginate`` / ``_single`` boundary, before any +transport, rather than through ``test/_client_harness.py``: what is under test +is the value an endpoint method hands its builder, and the fully built +``PolyswarmRequest`` (params *and* JSON body, for the search and the write +paths alike) is the most direct place to read it. The wire shape itself is +unchanged by refanging, so nothing below that boundary needs re-proving. +""" +import hashlib +import json +import pathlib + +import pytest + +from polyswarm_api import refang +from polyswarm_api.aio import PolySwarmAsyncAPI +from polyswarm_api.api import PolyswarmAPI + +# The case table is shared VERBATIM with the other PolySwarm clients that +# implement the same refang contract. Keep this file byte-identical across +# them: a change here is a change to the contract, and lands everywhere. +CASES_PATH = pathlib.Path(__file__).parent / 'fixtures' / 'refang_cases.json' +CASES = json.loads(CASES_PATH.read_text(encoding='utf-8')) + +# Drift guard: the web UI pins the same digest over its copy, so editing the +# table here fails this suite until the other copy -- and both pins -- change +# together. +SHARED_CASES_SHA256 = '4bda4b2f8f0dacdd3fa80632fe2a6044b9d0ff5194402968cac9698605a72da8' + + +def test_contract_table_is_byte_identical_to_the_pinned_copy(): + assert hashlib.sha256(CASES_PATH.read_bytes()).hexdigest() == SHARED_CASES_SHA256 + + +@pytest.mark.parametrize('case', CASES, ids=[c['why'] for c in CASES]) +def test_refang_ioc_contract_table(case): + assert refang.refang_ioc(case['input']) == case['expected'] + + +class TestRefangIoc: + def test_unchanged_input_is_returned_as_is_including_whitespace(self): + # Nothing to refang: the caller gets its own value back, untrimmed. + assert refang.refang_ioc(' evil.com ') == ' evil.com ' + + def test_accept_can_veto_a_candidate(self): + assert refang.refang_ioc('evil[.]com', accept=lambda c: False) == 'evil[.]com' + + def test_accept_sees_the_refanged_candidate(self): + seen = [] + assert refang.refang_ioc('evil[.]com', accept=lambda c: seen.append(c) or True) == 'evil.com' + assert seen == ['evil.com'] + + def test_accept_is_not_consulted_when_nothing_is_defanged(self): + def boom(_): + raise AssertionError('accept must not run') + assert refang.refang_ioc('evil.com', accept=boom) == 'evil.com' + + def test_non_string_values_pass_through(self): + assert refang.refang_ioc(None) is None + + +class TestRefangText: + def test_rewrites_without_gating_or_trimming(self): + # The raw rewrite is ungated: it rewrites a non-IoC too. + assert refang.refang_text(' hash[.]sha256 ') == ' hash.sha256 ' + + def test_scheme_rules_are_anchored_at_the_start(self): + assert refang.refang_text('see hxxp://evil.com') == 'see hxxp://evil.com' + + +class TestIsNetworkIoc: + @pytest.mark.parametrize('value', [ + 'evil.com', '127.0.0.1', 'https://evil.com/x?y#z', 'ftp://user@files.example.org:21/a', + 'http://[2001:db8::1]:8080/', 'xn--80ak6aa92e.com', + ]) + def test_accepts(self, value): + assert refang.is_network_ioc(value) + + @pytest.mark.parametrize('value', [ + '', 'evil', 'hash.sha256', '999.1.1.1', 'evil .com', 'a_b.com', 'hxxp://evil.com', + # Python's ``$`` also matches before a trailing newline; a full match + # must not, or this engine accepts what the others reject. + 'evil.com\n', + # A case-insensitive flag would fold U+212A KELVIN SIGN onto ``k``. + 'evil.\u212aom', + ]) + def test_rejects(self, value): + assert not refang.is_network_ioc(value) + + +# ── Client integration: request shape at the transport boundary ──────────── + +class _Captured(Exception): + """Raised by the fake boundary to stop a multi-step flow after its first + request, which is the one carrying the IoC.""" + + +def _sync_client(**kwargs): + # Refanging is opt-in; every test that is not about the default says so. + kwargs.setdefault('refang_iocs', True) + api = PolyswarmAPI(key='k' * 32, uri='https://api.example.test', community='gamma', **kwargs) + captured = [] + + def fake_paginate(request, *a, **kw): + captured.append(api._to_request(request)) + return iter(()) + + def fake_single(request, *a, **kw): + captured.append(api._to_request(request, *a, **kw)) + raise _Captured() + + api._paginate = fake_paginate + api._single = fake_single + return api, captured + + +def _async_client(**kwargs): + kwargs.setdefault('refang_iocs', True) + api = PolySwarmAsyncAPI(key='k' * 32, uri='https://api.example.test', community='gamma', **kwargs) + captured = [] + + async def fake_paginate(request, *a, **kw): + captured.append(api._to_request(request)) + return + yield # pragma: no cover - makes this an async generator + + async def fake_single(request, *a, **kw): + captured.append(api._to_request(request, *a, **kw)) + raise _Captured() + + api._paginate = fake_paginate + api._single = fake_single + return api, captured + + +def _params(request): + """Normalise a request's params to a list of (key, value) pairs.""" + params = request.params + if isinstance(params, dict): + pairs = [] + for key, value in params.items(): + if isinstance(value, (list, tuple)): + pairs.extend((key, v) for v in value) + else: + pairs.append((key, value)) + return pairs + return list(params) + + +def _run_sync(api, method, *args, **kwargs): + try: + list(getattr(api, method)(*args, **kwargs)) + except _Captured: + pass + + +async def _run_async(api, method, *args, **kwargs): + try: + result = getattr(api, method)(*args, **kwargs) + if hasattr(result, '__aiter__'): + async for _ in result: + pass + else: + await result + except _Captured: + pass + + +# (method, args, kwargs, param key, expected refanged value, raw value) +SEARCH_CALLS = [ + ('search_url', ('hxxps[:]//evil[.]com',), {}, 'url', 'https://evil.com', 'hxxps[:]//evil[.]com'), + ('search_by_metadata', ('strings.ips:*',), {'ips': ['127[.]0[.]0[.]1']}, 'ips', '127.0.0.1', '127[.]0[.]0[.]1'), + ('search_by_metadata', ('strings.urls:*',), {'urls': ['hxxp://evil[.]com/x']}, 'urls', 'http://evil.com/x', 'hxxp://evil[.]com/x'), + ('search_by_metadata', ('strings.domains:*',), {'domains': ['evil[dot]com']}, 'domains', 'evil.com', 'evil[dot]com'), + ('search_by_ioc', (), {'ip': '10[.]0[.]0[.]1'}, 'ip', '10.0.0.1', '10[.]0[.]0[.]1'), + ('search_by_ioc', (), {'domain': 'evil(.)com'}, 'domain', 'evil.com', 'evil(.)com'), + ('check_known_hosts', (), {'ips': ['8[.]8[.]8[.]8']}, 'ip', '8.8.8.8', '8[.]8[.]8[.]8'), + ('check_known_hosts', (), {'domains': ['good[.]example']}, 'domain', 'good.example', 'good[.]example'), + # A bare string where a list is expected (``_refang_all``'s str branch). + ('check_known_hosts', (), {'ips': '8[.]8[.]8[.]8'}, 'ip', '8.8.8.8', '8[.]8[.]8[.]8'), + ('check_known_hosts', (), {'domains': 'good[.]example'}, 'domain', 'good.example', 'good[.]example'), +] + + +@pytest.mark.parametrize('method,args,kwargs,key,refanged,raw', SEARCH_CALLS) +class TestSearchMethodsRefang: + def test_sync_sends_refanged_value(self, method, args, kwargs, key, refanged, raw): + api, captured = _sync_client() + _run_sync(api, method, *args, **kwargs) + assert (key, refanged) in _params(captured[0]) + + def test_sync_opt_out_sends_raw_value(self, method, args, kwargs, key, refanged, raw): + api, captured = _sync_client(refang_iocs=False) + _run_sync(api, method, *args, **kwargs) + assert (key, raw) in _params(captured[0]) + + async def test_async_sends_refanged_value(self, method, args, kwargs, key, refanged, raw): + api, captured = _async_client() + await _run_async(api, method, *args, **kwargs) + assert (key, refanged) in _params(captured[0]) + + async def test_async_opt_out_sends_raw_value(self, method, args, kwargs, key, refanged, raw): + api, captured = _async_client(refang_iocs=False) + await _run_async(api, method, *args, **kwargs) + assert (key, raw) in _params(captured[0]) + + +def test_metadata_free_form_query_is_never_refanged(): + api, captured = _sync_client() + _run_sync(api, 'search_by_metadata', 'strings.domains:"evil[.]com"') + assert ('query', 'strings.domains:"evil[.]com"') in _params(captured[0]) + + +def test_live_values_are_sent_unchanged(): + api, captured = _sync_client() + _run_sync(api, 'search_url', 'https://example.com/a[.]b') + assert ('url', 'https://example.com/a[.]b') in _params(captured[0]) + + +# (method, args, kwargs) — every URL submission path; the first request's JSON +# body carries the artifact name the server stores as the URL. +SUBMIT_CALLS = [ + ('submit', ('hxxps[:]//evil[.]com/x',), {'artifact_type': 'URL'}), + ('sandbox_file', ('hxxps[:]//evil[.]com/x', 'provider', 'vm'), {'artifact_type': 'URL'}), + ('sandbox_url', ('hxxps[:]//evil[.]com/x', 'provider', 'vm'), {}), +] + + +@pytest.mark.parametrize('method,args,kwargs', SUBMIT_CALLS) +class TestUrlSubmissionRefang: + def test_sync_submits_refanged_url(self, method, args, kwargs): + api, captured = _sync_client() + _run_sync(api, method, *args, **kwargs) + assert captured[0].input_json['artifact_name'] == 'https://evil.com/x' + + def test_sync_opt_out_submits_raw_url(self, method, args, kwargs): + api, captured = _sync_client(refang_iocs=False) + _run_sync(api, method, *args, **kwargs) + assert captured[0].input_json['artifact_name'] == 'hxxps[:]//evil[.]com/x' + + async def test_async_submits_refanged_url(self, method, args, kwargs): + api, captured = _async_client() + await _run_async(api, method, *args, **kwargs) + assert captured[0].input_json['artifact_name'] == 'https://evil.com/x' + + async def test_async_opt_out_submits_raw_url(self, method, args, kwargs): + api, captured = _async_client(refang_iocs=False) + await _run_async(api, method, *args, **kwargs) + assert captured[0].input_json['artifact_name'] == 'hxxps[:]//evil[.]com/x' + + +def test_explicit_artifact_name_is_kept(): + api, captured = _sync_client() + _run_sync(api, 'submit', 'hxxp://evil[.]com', artifact_type='URL', artifact_name='my label') + assert captured[0].input_json['artifact_name'] == 'my label' + + +def _spy_uploaded_content(monkeypatch): + from polyswarm_api import resources + contents = [] + original = resources.LocalArtifact.from_content.__func__ + + def spy(cls, api, content, *a, **kw): + contents.append(content) + return original(cls, api, content, *a, **kw) + + monkeypatch.setattr(resources.LocalArtifact, 'from_content', classmethod(spy)) + return contents + + +# The server stores the uploaded content as the URL artifact, so the content — +# not only the default artifact name — must be refanged, on both transports, +# and left raw when refanging is off. +@pytest.mark.parametrize('method,args,kwargs', SUBMIT_CALLS) +class TestUploadedUrlContent: + def test_sync_content_is_refanged(self, monkeypatch, method, args, kwargs): + contents = _spy_uploaded_content(monkeypatch) + api, _ = _sync_client() + _run_sync(api, method, *args, **kwargs) + assert contents == ['https://evil.com/x'] + + def test_sync_opt_out_uploads_raw_content(self, monkeypatch, method, args, kwargs): + contents = _spy_uploaded_content(monkeypatch) + api, _ = _sync_client(refang_iocs=False) + _run_sync(api, method, *args, **kwargs) + assert contents == [args[0]] + + async def test_async_content_is_refanged(self, monkeypatch, method, args, kwargs): + contents = _spy_uploaded_content(monkeypatch) + api, _ = _async_client() + await _run_async(api, method, *args, **kwargs) + assert contents == ['https://evil.com/x'] + + async def test_async_opt_out_uploads_raw_content(self, monkeypatch, method, args, kwargs): + contents = _spy_uploaded_content(monkeypatch) + api, _ = _async_client(refang_iocs=False) + await _run_async(api, method, *args, **kwargs) + assert contents == [args[0]] + + +# Refanging is OFF unless asked for: a default that preserves 4.5.0 behaviour is +# what makes this a minor release (specs/05-downstream-contract.md, Versioning). +class TestDefaultIsOff: + def test_sync_default_sends_raw(self): + api = PolyswarmAPI(key='k' * 32, uri='https://api.example.test', community='gamma') + assert api.refang_iocs is False + api, captured = _sync_client(refang_iocs=api.refang_iocs) + _run_sync(api, 'search_url', 'hxxps[:]//evil[.]com/x') + assert ('url', 'hxxps[:]//evil[.]com/x') in _params(captured[0]) + + async def test_async_default_sends_raw(self): + api = PolySwarmAsyncAPI(key='k' * 32, uri='https://api.example.test', community='gamma') + assert api.refang_iocs is False + api, captured = _async_client(refang_iocs=api.refang_iocs) + await _run_async(api, 'search_url', 'hxxps[:]//evil[.]com/x') + assert ('url', 'hxxps[:]//evil[.]com/x') in _params(captured[0]) + + +# (method, args) — known-host catalogue writes; the host rides the JSON body. +HOST_WRITE_CALLS = [ + ('add_known_good_host', ('domain', 'feed', 'good[.]example')), + ('add_known_bad_host', ('domain', 'feed', 'good[.]example')), + ('update_known_good_host', (7, 'domain', 'feed', 'good[.]example', True)), +] + + +@pytest.mark.parametrize('method,args', HOST_WRITE_CALLS) +class TestKnownHostWritesRefang: + def test_sync_writes_refanged_host(self, method, args): + api, captured = _sync_client() + _run_sync(api, method, *args) + assert captured[0].input_json['host'] == 'good.example' + + def test_sync_opt_out_writes_raw_host(self, method, args): + api, captured = _sync_client(refang_iocs=False) + _run_sync(api, method, *args) + assert captured[0].input_json['host'] == 'good[.]example' + + async def test_async_writes_refanged_host(self, method, args): + api, captured = _async_client() + await _run_async(api, method, *args) + assert captured[0].input_json['host'] == 'good.example' + + async def test_async_opt_out_writes_raw_host(self, method, args): + api, captured = _async_client(refang_iocs=False) + await _run_async(api, method, *args) + assert captured[0].input_json['host'] == 'good[.]example' + + +# ── QR-code submissions: the argument is an image PATH, never a URL ──────── +# +# ``qr[.]png`` refangs to ``qr.png``, which is shaped like a domain, so these +# fail if the QR branch ever runs the refang. ``submit`` reads the image with +# ``LocalArtifact.from_path``; ``sandbox_file`` hands the string to +# ``from_content`` (its URL branch has no path reader), so each is spied where +# the value actually lands. + +QR_PREPROCESSING = {'type': 'qrcode'} + + +def _spy(monkeypatch, name): + from polyswarm_api import resources + seen = [] + + def spy(cls, api, value, *a, **kw): + seen.append(value) + raise _Captured() + + monkeypatch.setattr(resources.LocalArtifact, name, classmethod(spy)) + return seen + + +def test_sync_submit_qrcode_path_is_never_refanged(monkeypatch): + seen = _spy(monkeypatch, 'from_path') + api, _ = _sync_client() + _run_sync(api, 'submit', 'qr[.]png', artifact_type='URL', preprocessing=QR_PREPROCESSING) + assert seen == ['qr[.]png'] + + +async def test_async_submit_qrcode_path_is_never_refanged(monkeypatch): + seen = _spy(monkeypatch, 'from_path') + api, _ = _async_client() + await _run_async(api, 'submit', 'qr[.]png', artifact_type='URL', preprocessing=QR_PREPROCESSING) + assert seen == ['qr[.]png'] + + +def test_sync_sandbox_file_qrcode_value_is_never_refanged(monkeypatch): + seen = _spy(monkeypatch, 'from_content') + api, _ = _sync_client() + _run_sync(api, 'sandbox_file', 'qr[.]png', 'provider', 'vm', artifact_type='URL', + preprocessing=QR_PREPROCESSING) + assert seen == ['qr[.]png'] + + +async def test_async_sandbox_file_qrcode_value_is_never_refanged(monkeypatch): + seen = _spy(monkeypatch, 'from_content') + api, _ = _async_client() + await _run_async(api, 'sandbox_file', 'qr[.]png', 'provider', 'vm', artifact_type='URL', + preprocessing=QR_PREPROCESSING) + assert seen == ['qr[.]png']