From e9fd1bae92f5934541f14b5d68ed24fe3929bedd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?V=C3=ADctor=20Mart=C3=ADnez?= Date: Tue, 22 Sep 2026 15:25:06 -0300 Subject: [PATCH 1/9] chore: bump version to 4.6.0 for the IoC refang surface --- pyproject.toml | 4 ++-- src/polyswarm_api/__init__.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) 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/src/polyswarm_api/__init__.py b/src/polyswarm_api/__init__.py index d9cf1695..52829e54 100644 --- a/src/polyswarm_api/__init__.py +++ b/src/polyswarm_api/__init__.py @@ -1,5 +1,5 @@ # 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 29db1655752f8fcfb4efe58698ed7ac5c7151de4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?V=C3=ADctor=20Mart=C3=ADnez?= Date: Tue, 22 Sep 2026 15:25:34 -0300 Subject: [PATCH 2/9] feat: refang defanged IoC inputs before building requests Threat-intel reports print indicators defanged (hxxps[:]//evil[.]com, 127[.]0[.]0[.]1). The server looks URLs up by an exact hash and stores a submitted URL verbatim, so a defanged value silently missed a search or became a broken URL artifact. Add polyswarm_api.refang (refang_text, is_network_ioc, refang_ioc) and a refang_iocs=True constructor flag. When on, the client refangs the URL/domain/IP inputs of search_url, search_by_metadata (ips/urls/domains, never the free-form query), search_by_ioc, check_known_hosts, the known-host writes, and every URL submission path. Live values are sent unchanged; refang_iocs=False sends inputs verbatim. The rules, gate and case table are shared with other PolySwarm clients; test/fixtures/refang_cases.json is kept byte-identical across them. --- src/polyswarm_api/__init__.py | 1 + src/polyswarm_api/aio/api.py | 38 ++++- src/polyswarm_api/api.py | 42 ++++- src/polyswarm_api/refang.py | 111 +++++++++++++ test/fixtures/refang_cases.json | 36 ++++ test/refang_test.py | 281 ++++++++++++++++++++++++++++++++ 6 files changed, 507 insertions(+), 2 deletions(-) create mode 100644 src/polyswarm_api/refang.py create mode 100644 test/fixtures/refang_cases.json create mode 100644 test/refang_test.py diff --git a/src/polyswarm_api/__init__.py b/src/polyswarm_api/__init__.py index 52829e54..e80be0e3 100644 --- a/src/polyswarm_api/__init__.py +++ b/src/polyswarm_api/__init__.py @@ -4,6 +4,7 @@ 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..c82d687d 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 = True, **httpx_kwargs, ): key_masked = '******' + (key[-4:] if key and len(key) > 16 else '') @@ -64,6 +65,10 @@ 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. 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 +93,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 +366,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 +392,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 +419,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 +432,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 +446,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 +459,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 +472,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)) @@ -1378,6 +1411,7 @@ async def submit( 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 +1514,7 @@ async def sandbox_file( self, artifact, artifact_type=artifact_type, artifact_name=artifact_name ) elif artifact_type == resources.ArtifactType.URL: + artifact = self._refang(artifact) artifact = resources.LocalArtifact.from_content( self, artifact, artifact_name=artifact_name or artifact, artifact_type=artifact_type, @@ -1542,6 +1577,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..8f5abc4b 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 = True, **httpx_kwargs, ): key_masked = "******" + (key[-4:] if key and len(key) > 16 else "") @@ -69,6 +70,10 @@ 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. 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 +100,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 +392,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 +422,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 +468,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 +487,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 +501,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 +514,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 +527,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) @@ -1708,6 +1745,7 @@ def submit( artifact_name=artifact_name, ) else: + artifact = self._refang(artifact) artifact = resources.LocalArtifact.from_content( self, artifact, @@ -1822,6 +1860,7 @@ def sandbox_file( artifact_name=artifact_name, ) elif artifact_type == resources.ArtifactType.URL: + artifact = self._refang(artifact) artifact = resources.LocalArtifact.from_content( self, artifact, @@ -1896,6 +1935,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..6609578b --- /dev/null +++ b/src/polyswarm_api/refang.py @@ -0,0 +1,111 @@ +"""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). The +regexes avoid ``\\b``, ``\\d`` and ``\\w`` on purpose so every implementation +matches the same characters. + +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'] + +# 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(r'[\[({]\s*:\s*/\s*/\s*[\])}]'), '://'), + (re.compile(r'[\[({]\s*:\s*[\])}]'), ':'), + (re.compile(r'[\[({]\s*/\s*[\])}]'), '/'), + (re.compile(r'[\[({]\s*\.\s*[\])}]'), '.'), + (re.compile(r'[\[({]\s*dot\s*[\])}]', re.IGNORECASE), '.'), + (re.compile(r'^h[x*]{2}ps(?=://)', re.IGNORECASE), 'https'), + (re.compile(r'^h[x*]{2}p(?=://)', re.IGNORECASE), 'http'), + (re.compile(r'^fxps(?=://)', re.IGNORECASE), 'ftps'), + (re.compile(r'^fxp(?=://)', re.IGNORECASE), '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-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?' +_TLD = r'(?:[a-z]{2,63}|xn--[a-z0-9-]{1,59})' +_DOMAIN = rf'(?:{_LABEL}\.)+{_TLD}' +_IPV6 = r'\[[0-9a-f:.]+\]' +_HOST = rf'(?:{_DOMAIN}|{_IPV4}|{_IPV6})' +_NETWORK_IOC = re.compile( + rf'^(?:(?:https?|ftps?)://)?(?:[^\s/?#@]+@)?{_HOST}(?::[0-9]{{1,5}})?(?:[/?#]\S*)?$', + re.IGNORECASE, +) +_LIVE_URL_HOST = re.compile(r'^(?:https?|ftps?)://([^/?#]*)', re.IGNORECASE) +_QUERY_SYNTAX = re.compile(r'[\s"]') + + +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.match(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 whitespace or a double quote -> unchanged: that is + a query or quoted data, never a single indicator; + * the input is already a live http(s)/ftp(s) URL whose host has no defang + token -> unchanged, so a legitimate ``[.]`` in a path 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. + """ + if not isinstance(value, str): + return value + trimmed = value.strip() + candidate = refang_text(trimmed) + if candidate == trimmed: + return value + if _QUERY_SYNTAX.search(candidate): + return value + live = _LIVE_URL_HOST.match(trimmed) + if live and refang_text(live.group(1)) == live.group(1): + 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..7ae7db21 --- /dev/null +++ b/test/fixtures/refang_cases.json @@ -0,0 +1,36 @@ +[ + { "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": "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": "", "expected": "", "why": "empty input" } +] diff --git a/test/refang_test.py b/test/refang_test.py new file mode 100644 index 00000000..08ed479d --- /dev/null +++ b/test/refang_test.py @@ -0,0 +1,281 @@ +"""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 request is +captured at the ``_paginate`` / ``_single`` boundary, before any transport. +""" +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 = json.loads( + (pathlib.Path(__file__).parent / 'fixtures' / 'refang_cases.json').read_text(encoding='utf-8') +) + + +@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', + ]) + 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): + 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): + 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'), +] + + +@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' + + +@pytest.mark.parametrize('method,args,kwargs', SUBMIT_CALLS) +def test_uploaded_url_content_is_refanged(monkeypatch, method, args, kwargs): + # The server stores the uploaded content as the URL artifact, so the + # content — not only the default artifact name — must be refanged. + 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)) + api, _ = _sync_client() + _run_sync(api, method, *args, **kwargs) + assert contents == ['https://evil.com/x'] + + +# (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' From cdabf8c3cb2b98bfa5e59e54c4e942ad4724e83b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?V=C3=ADctor=20Mart=C3=ADnez?= Date: Tue, 22 Sep 2026 15:25:34 -0300 Subject: [PATCH 3/9] docs: document IoC refanging in the endpoint and downstream-contract specs --- specs/03-endpoints.md | 4 ++++ specs/04-testing.md | 1 + specs/05-downstream-contract.md | 22 +++++++++++++++++++++- 3 files changed, 26 insertions(+), 1 deletion(-) diff --git a/specs/03-endpoints.md b/specs/03-endpoints.md index 4eb0c86d..6a3eb5c2 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's `refang_iocs` is on (the 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)). + `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..34683517 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=True, **httpx_kwargs): ... def close(self): ... def __enter__(self): ... def __exit__(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 = True, # refang defanged URL/domain/IP inputs — see "IoC refanging" **httpx_kwargs, # forwarded to httpx.Client when constructing default session ) ``` @@ -210,6 +212,24 @@ If `session=` is provided, the api client uses it as-is (no construction) and `k Same shape for `PolySwarmAsyncAPI` with `AsyncPolyswarmSession`. +## 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 refangs at its own edge (added in 4.6.0). + +**`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 input is not already a live URL with a clean host, 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 `refang_iocs=True`, the default): `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, ids and QR-code preprocessing paths are never touched. Pass `refang_iocs=False` to send inputs verbatim. + +Compatibility: the default changes behaviour only for inputs that were defanged network IoCs, which previously could never match or produced a broken URL artifact; any live value is sent byte-for-byte as before. + `**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. ## Customizing transport behaviour From add0f0814b8c4930008e56a690adcafb7c8a8c4b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?V=C3=ADctor=20Mart=C3=ADnez?= Date: Tue, 22 Sep 2026 15:28:29 -0300 Subject: [PATCH 4/9] docs: note refang_iocs is a public attribute consumers can read --- specs/05-downstream-contract.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/specs/05-downstream-contract.md b/specs/05-downstream-contract.md index 34683517..07ecb8ea 100644 --- a/specs/05-downstream-contract.md +++ b/specs/05-downstream-contract.md @@ -226,7 +226,7 @@ Threat-intel reports print indicators defanged (`hxxps[:]//evil[.]com`, `127[.]0 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 `refang_iocs=True`, the default): `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, ids and QR-code preprocessing paths are never touched. Pass `refang_iocs=False` to send inputs verbatim. +**Where the client applies it** (when `refang_iocs=True`, the default): `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, ids and QR-code preprocessing paths are never touched. Pass `refang_iocs=False` to send inputs verbatim. 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: the default changes behaviour only for inputs that were defanged network IoCs, which previously could never match or produced a broken URL artifact; any live value is sent byte-for-byte as before. From 58e061f6ab22ca389437571ceca759d449299837 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?V=C3=ADctor=20Mart=C3=ADnez?= Date: Tue, 22 Sep 2026 15:36:34 -0300 Subject: [PATCH 5/9] fix: make refang match the same characters in every regex engine Python's case-insensitive flag folds Unicode (U+212A KELVIN SIGN matched "k", so "evil[.]Kom" was refanged), \s/\S cover Unicode whitespace, and "$" matches before a trailing newline - all unlike the other clients that implement the same contract. Spell letters as explicit [aA] classes, use the ASCII whitespace set, trim only that set, and fullmatch the network-IoC pattern. The shared case table grows to 39 cases pinning these. --- src/polyswarm_api/refang.py | 58 +++++++++++++++----------- test/fixtures/refang_cases.json | 73 ++++++++++++++++++--------------- test/refang_test.py | 5 +++ 3 files changed, 79 insertions(+), 57 deletions(-) diff --git a/src/polyswarm_api/refang.py b/src/polyswarm_api/refang.py index 6609578b..ffd178a1 100644 --- a/src/polyswarm_api/refang.py +++ b/src/polyswarm_api/refang.py @@ -10,9 +10,16 @@ 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). The -regexes avoid ``\\b``, ``\\d`` and ``\\w`` on purpose so every implementation -matches the same characters. +(``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 @@ -23,36 +30,41 @@ __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(r'[\[({]\s*:\s*/\s*/\s*[\])}]'), '://'), - (re.compile(r'[\[({]\s*:\s*[\])}]'), ':'), - (re.compile(r'[\[({]\s*/\s*[\])}]'), '/'), - (re.compile(r'[\[({]\s*\.\s*[\])}]'), '.'), - (re.compile(r'[\[({]\s*dot\s*[\])}]', re.IGNORECASE), '.'), - (re.compile(r'^h[x*]{2}ps(?=://)', re.IGNORECASE), 'https'), - (re.compile(r'^h[x*]{2}p(?=://)', re.IGNORECASE), 'http'), - (re.compile(r'^fxps(?=://)', re.IGNORECASE), 'ftps'), - (re.compile(r'^fxp(?=://)', re.IGNORECASE), 'ftp'), + (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-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?' -_TLD = r'(?:[a-z]{2,63}|xn--[a-z0-9-]{1,59})' +_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-f:.]+\]' +_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'^(?:(?:https?|ftps?)://)?(?:[^\s/?#@]+@)?{_HOST}(?::[0-9]{{1,5}})?(?:[/?#]\S*)?$', - re.IGNORECASE, + rf'(?:{_SCHEME}://)?(?:[^ \t\n\r\f\v/?#@]+@)?{_HOST}(?::[0-9]{{1,5}})?(?:[/?#][^ \t\n\r\f\v]*)?' ) -_LIVE_URL_HOST = re.compile(r'^(?:https?|ftps?)://([^/?#]*)', re.IGNORECASE) -_QUERY_SYNTAX = re.compile(r'[\s"]') +_LIVE_URL_HOST = re.compile(rf'^{_SCHEME}://([^/?#]*)') +_QUERY_SYNTAX = re.compile(r'[ \t\n\r\f\v"]') def refang_text(text): @@ -72,7 +84,7 @@ def is_network_ioc(candidate): 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.match(candidate)) + return bool(_NETWORK_IOC.fullmatch(candidate)) def refang_ioc(value, accept=None): @@ -82,7 +94,7 @@ def refang_ioc(value, accept=None): run on every IoC-shaped input. The gate, in order: * nothing was defanged -> unchanged; - * the rewrite contains whitespace or a double quote -> unchanged: that is + * the rewrite contains ASCII whitespace or a double quote -> unchanged: that is a query or quoted data, never a single indicator; * the input is already a live http(s)/ftp(s) URL whose host has no defang token -> unchanged, so a legitimate ``[.]`` in a path survives; @@ -91,11 +103,11 @@ def refang_ioc(value, accept=None): to require their own routing to agree, e.g. "this would be searched as a URL". - A refanged value is returned trimmed. + A refanged value is returned trimmed of ASCII whitespace. """ if not isinstance(value, str): return value - trimmed = value.strip() + trimmed = value.strip(_WS_CHARS) candidate = refang_text(trimmed) if candidate == trimmed: return value diff --git a/test/fixtures/refang_cases.json b/test/fixtures/refang_cases.json index 7ae7db21..52d2b756 100644 --- a/test/fixtures/refang_cases.json +++ b/test/fixtures/refang_cases.json @@ -1,36 +1,41 @@ [ - { "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": "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": "", "expected": "", "why": "empty input" } + {"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": "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 index 08ed479d..0de2b011 100644 --- a/test/refang_test.py +++ b/test/refang_test.py @@ -71,6 +71,11 @@ def test_accepts(self, 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) From 96e5bb1a9258ca80fcbac2d75f3daab6f01abb07 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?V=C3=ADctor=20Mart=C3=ADnez?= Date: Tue, 22 Sep 2026 15:37:28 -0300 Subject: [PATCH 6/9] fix: never refang a QR-code submission's image path in sandbox_file submit already skipped the refang for preprocessing type qrcode, whose argument names an image file rather than a URL; sandbox_file did not, so a file called qr[.]png was sent as qr.png. Pin both paths on both transports, cover the async opt-out of the known-host writes and the bare-string form of the list arguments, and put the httpx_kwargs paragraph back under the constructor block in the contract spec. --- specs/03-endpoints.md | 2 +- specs/05-downstream-contract.md | 6 ++-- src/polyswarm_api/aio/api.py | 5 ++- src/polyswarm_api/api.py | 5 ++- test/refang_test.py | 61 +++++++++++++++++++++++++++++++++ 5 files changed, 73 insertions(+), 6 deletions(-) diff --git a/specs/03-endpoints.md b/specs/03-endpoints.md index 6a3eb5c2..f4245dd0 100644 --- a/specs/03-endpoints.md +++ b/specs/03-endpoints.md @@ -264,7 +264,7 @@ 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 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". diff --git a/specs/05-downstream-contract.md b/specs/05-downstream-contract.md index 07ecb8ea..7b6c3530 100644 --- a/specs/05-downstream-contract.md +++ b/specs/05-downstream-contract.md @@ -212,6 +212,8 @@ If `session=` is provided, the api client uses it as-is (no construction) and `k 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 refangs at its own edge (added in 4.6.0). @@ -226,12 +228,10 @@ Threat-intel reports print indicators defanged (`hxxps[:]//evil[.]com`, `127[.]0 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 `refang_iocs=True`, the default): `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, ids and QR-code preprocessing paths are never touched. Pass `refang_iocs=False` to send inputs verbatim. 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. +**Where the client applies it** (when `refang_iocs=True`, the default): `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`. Pass `refang_iocs=False` to send inputs verbatim. 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: the default changes behaviour only for inputs that were defanged network IoCs, which previously could never match or produced a broken URL artifact; any live value is sent byte-for-byte as before. -`**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. - ## Customizing transport behaviour Replaces the 3.x module-level monkey-patching pattern with subclassing. diff --git a/src/polyswarm_api/aio/api.py b/src/polyswarm_api/aio/api.py index c82d687d..714af3d6 100644 --- a/src/polyswarm_api/aio/api.py +++ b/src/polyswarm_api/aio/api.py @@ -1514,7 +1514,10 @@ async def sandbox_file( self, artifact, artifact_type=artifact_type, artifact_name=artifact_name ) elif artifact_type == resources.ArtifactType.URL: - artifact = self._refang(artifact) + # 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, diff --git a/src/polyswarm_api/api.py b/src/polyswarm_api/api.py index 8f5abc4b..3cd1aad5 100644 --- a/src/polyswarm_api/api.py +++ b/src/polyswarm_api/api.py @@ -1860,7 +1860,10 @@ def sandbox_file( artifact_name=artifact_name, ) elif artifact_type == resources.ArtifactType.URL: - artifact = self._refang(artifact) + # 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, diff --git a/test/refang_test.py b/test/refang_test.py index 0de2b011..6a6a18c9 100644 --- a/test/refang_test.py +++ b/test/refang_test.py @@ -166,6 +166,9 @@ async def _run_async(api, method, *args, **kwargs): ('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'), ] @@ -284,3 +287,61 @@ 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'] From 19d30dee18d151741ca548e5e07329a2d4320c62 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?V=C3=ADctor=20Mart=C3=ADnez?= Date: Tue, 22 Sep 2026 15:46:24 -0300 Subject: [PATCH 7/9] test: pin the shared refang case table by digest; place refang.py in the architecture specs --- specs/00-overview.md | 2 ++ specs/01-architecture.md | 1 + test/refang_test.py | 15 ++++++++++++--- 3 files changed, 15 insertions(+), 3 deletions(-) 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/test/refang_test.py b/test/refang_test.py index 6a6a18c9..c092df9b 100644 --- a/test/refang_test.py +++ b/test/refang_test.py @@ -8,6 +8,7 @@ when ``refang_iocs`` is on, and the raw value when it is off. The request is captured at the ``_paginate`` / ``_single`` boundary, before any transport. """ +import hashlib import json import pathlib @@ -20,9 +21,17 @@ # 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 = json.loads( - (pathlib.Path(__file__).parent / 'fixtures' / 'refang_cases.json').read_text(encoding='utf-8') -) +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 = '78359d1ac589055146fee033fd7908713dca768c6631aaf8dc41de6524995fa9' + + +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]) From 95682a7f57fecda6109c8a2356ac4231dcf4f83c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?V=C3=ADctor=20Mart=C3=ADnez?= Date: Tue, 22 Sep 2026 18:02:03 -0300 Subject: [PATCH 8/9] fix: never rewrite only the path of a scheme-less URL; document refang_iocs on the async constructor --- specs/05-downstream-contract.md | 4 ++-- src/polyswarm_api/refang.py | 12 +++++++----- test/fixtures/refang_cases.json | 3 +++ test/refang_test.py | 2 +- 4 files changed, 13 insertions(+), 8 deletions(-) diff --git a/specs/05-downstream-contract.md b/specs/05-downstream-contract.md index 7b6c3530..57ad36c0 100644 --- a/specs/05-downstream-contract.md +++ b/specs/05-downstream-contract.md @@ -63,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=True, **httpx_kwargs): ... async def aclose(self): ... async def __aenter__(self): ... async def __aexit__(self, *exc): ... @@ -224,7 +224,7 @@ Threat-intel reports print indicators defanged (`hxxps[:]//evil[.]com`, `127[.]0 |---|---| | `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 input is not already a live URL with a clean host, the result is a network IoC, and the optional `accept(candidate)` agrees. Otherwise returns `value` unchanged (untrimmed). Non-strings pass through. | +| `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. diff --git a/src/polyswarm_api/refang.py b/src/polyswarm_api/refang.py index ffd178a1..fb77205b 100644 --- a/src/polyswarm_api/refang.py +++ b/src/polyswarm_api/refang.py @@ -63,7 +63,9 @@ _NETWORK_IOC = re.compile( rf'(?:{_SCHEME}://)?(?:[^ \t\n\r\f\v/?#@]+@)?{_HOST}(?::[0-9]{{1,5}})?(?:[/?#][^ \t\n\r\f\v]*)?' ) -_LIVE_URL_HOST = re.compile(rf'^{_SCHEME}://([^/?#]*)') +# 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"]') @@ -96,8 +98,9 @@ def refang_ioc(value, accept=None): * 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 input is already a live http(s)/ftp(s) URL whose host has no defang - token -> unchanged, so a legitimate ``[.]`` in a path survives; + * 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 @@ -113,8 +116,7 @@ def refang_ioc(value, accept=None): return value if _QUERY_SYNTAX.search(candidate): return value - live = _LIVE_URL_HOST.match(trimmed) - if live and refang_text(live.group(1)) == live.group(1): + if candidate.startswith(_SCHEME_AND_HOST.match(trimmed).group(0)): return value if not is_network_ioc(candidate): return value diff --git a/test/fixtures/refang_cases.json b/test/fixtures/refang_cases.json index 52d2b756..9add92a3 100644 --- a/test/fixtures/refang_cases.json +++ b/test/fixtures/refang_cases.json @@ -21,6 +21,9 @@ {"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)"}, diff --git a/test/refang_test.py b/test/refang_test.py index c092df9b..22bed5be 100644 --- a/test/refang_test.py +++ b/test/refang_test.py @@ -27,7 +27,7 @@ # 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 = '78359d1ac589055146fee033fd7908713dca768c6631aaf8dc41de6524995fa9' +SHARED_CASES_SHA256 = '4bda4b2f8f0dacdd3fa80632fe2a6044b9d0ff5194402968cac9698605a72da8' def test_contract_table_is_byte_identical_to_the_pinned_copy(): From 1647bddb7689987d2bdb88dcfe8777a323d1acff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?V=C3=ADctor=20Mart=C3=ADnez?= Date: Tue, 22 Sep 2026 18:20:57 -0300 Subject: [PATCH 9/9] fix: make refanging opt-in so 4.6.0 keeps the 4.5.0 default behaviour A default-on refang changes what the write paths store (known-host rows, URL submissions), which the versioning table only allows in a major. With refang_iocs=False as the default the new keyword is purely additive; the CLI opts in through its own option. Also read the QR preprocessing type with .get() in submit, as sandbox_file already does, and cover the uploaded URL content on both transports with its opt-out twin. --- specs/03-endpoints.md | 2 +- specs/05-downstream-contract.md | 12 +++--- src/polyswarm_api/aio/api.py | 9 +++-- src/polyswarm_api/api.py | 9 +++-- test/refang_test.py | 69 ++++++++++++++++++++++++++++----- 5 files changed, 77 insertions(+), 24 deletions(-) diff --git a/specs/03-endpoints.md b/specs/03-endpoints.md index f4245dd0..12e4f857 100644 --- a/specs/03-endpoints.md +++ b/specs/03-endpoints.md @@ -205,7 +205,7 @@ 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's `refang_iocs` is on (the 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. +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 diff --git a/specs/05-downstream-contract.md b/specs/05-downstream-contract.md index 57ad36c0..271c6db6 100644 --- a/specs/05-downstream-contract.md +++ b/specs/05-downstream-contract.md @@ -49,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, refang_iocs=True, **httpx_kwargs): ... + verify=True, *, session=None, refang_iocs=False, **httpx_kwargs): ... def close(self): ... def __enter__(self): ... def __exit__(self, *exc): ... @@ -63,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, refang_iocs=True, **httpx_kwargs): ... + verify=True, *, session=None, refang_iocs=False, **httpx_kwargs): ... async def aclose(self): ... async def __aenter__(self): ... async def __aexit__(self, *exc): ... @@ -203,7 +203,7 @@ PolyswarmAPI( verify: bool = True, *, session: PolyswarmSession | None = None, # pre-built session; mutually exclusive with `**httpx_kwargs` - refang_iocs: bool = True, # refang defanged URL/domain/IP inputs — see "IoC refanging" + 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 ) ``` @@ -216,7 +216,7 @@ Same shape for `PolySwarmAsyncAPI` with `AsyncPolyswarmSession`. ## 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 refangs at its own edge (added in 4.6.0). +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): @@ -228,9 +228,9 @@ Threat-intel reports print indicators defanged (`hxxps[:]//evil[.]com`, `127[.]0 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 `refang_iocs=True`, the default): `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`. Pass `refang_iocs=False` to send inputs verbatim. 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. +**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: the default changes behaviour only for inputs that were defanged network IoCs, which previously could never match or produced a broken URL artifact; any live value is sent byte-for-byte as before. +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 diff --git a/src/polyswarm_api/aio/api.py b/src/polyswarm_api/aio/api.py index 714af3d6..7cbc1ab9 100644 --- a/src/polyswarm_api/aio/api.py +++ b/src/polyswarm_api/aio/api.py @@ -53,7 +53,7 @@ def __init__( verify: bool = True, *, session: AsyncPolyswarmSession | None = None, - refang_iocs: bool = True, + refang_iocs: bool = False, **httpx_kwargs, ): key_masked = '******' + (key[-4:] if key and len(key) > 16 else '') @@ -66,8 +66,9 @@ def __init__( self.timeout = timeout or settings.DEFAULT_HTTP_TIMEOUT self.verify = verify # Refang defanged URL / domain / IP inputs (``hxxps[:]//evil[.]com``) - # before building a request. See ``polyswarm_api.refang`` and - # ``_refang`` below for exactly which inputs are touched. + # 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 @@ -1406,7 +1407,7 @@ 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 ) diff --git a/src/polyswarm_api/api.py b/src/polyswarm_api/api.py index 3cd1aad5..3ac8beda 100644 --- a/src/polyswarm_api/api.py +++ b/src/polyswarm_api/api.py @@ -55,7 +55,7 @@ def __init__( verify: bool = True, *, session: PolyswarmSession | None = None, - refang_iocs: bool = True, + refang_iocs: bool = False, **httpx_kwargs, ): key_masked = "******" + (key[-4:] if key and len(key) > 16 else "") @@ -71,8 +71,9 @@ def __init__( self.timeout = timeout or settings.DEFAULT_HTTP_TIMEOUT self.verify = verify # Refang defanged URL / domain / IP inputs (``hxxps[:]//evil[.]com``) - # before building a request. See ``polyswarm_api.refang`` and - # ``_refang`` below for exactly which inputs are touched. + # 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 @@ -1737,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, diff --git a/test/refang_test.py b/test/refang_test.py index 22bed5be..bd324017 100644 --- a/test/refang_test.py +++ b/test/refang_test.py @@ -5,8 +5,13 @@ 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 request is -captured at the ``_paginate`` / ``_single`` boundary, before any transport. +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 @@ -98,6 +103,8 @@ class _Captured(Exception): 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 = [] @@ -115,6 +122,7 @@ def fake_single(request, *a, **kw): def _async_client(**kwargs): + kwargs.setdefault('refang_iocs', True) api = PolySwarmAsyncAPI(key='k' * 32, uri='https://api.example.test', community='gamma', **kwargs) captured = [] @@ -254,10 +262,7 @@ def test_explicit_artifact_name_is_kept(): assert captured[0].input_json['artifact_name'] == 'my label' -@pytest.mark.parametrize('method,args,kwargs', SUBMIT_CALLS) -def test_uploaded_url_content_is_refanged(monkeypatch, method, args, kwargs): - # The server stores the uploaded content as the URL artifact, so the - # content — not only the default artifact name — must be refanged. +def _spy_uploaded_content(monkeypatch): from polyswarm_api import resources contents = [] original = resources.LocalArtifact.from_content.__func__ @@ -267,9 +272,55 @@ def spy(cls, api, content, *a, **kw): return original(cls, api, content, *a, **kw) monkeypatch.setattr(resources.LocalArtifact, 'from_content', classmethod(spy)) - api, _ = _sync_client() - _run_sync(api, method, *args, **kwargs) - assert contents == ['https://evil.com/x'] + 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.