feat: refang defanged IoC inputs before building requests

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

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

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. |

specs/03-endpoints.md

@@ -205,6 +205,8 @@ refusal.
 | `notification_webhook_list()` | `Webhook.list` |
 | `report_template_list(is_default=None, **kwargs)` | `ReportTemplate.list` |
 
+IoC inputs of `search_url`, `search_by_metadata` (`ips` / `urls` / `domains` only), `search_by_ioc` (`ip` / `domain`) and `check_known_hosts` are refanged before the builder runs when the client was constructed with `refang_iocs=True` (opt-in; off by default) — see [`05-downstream-contract.md`](./05-downstream-contract.md) §"IoC refanging". The builders themselves are unchanged and never refang. The known-host writes (`add_known_good_host` / `add_known_bad_host` / `update_known_good_host`) refang their `host` the same way.
+
 ## Special methods
 
 | Method | Why it's special |
@@ -262,6 +264,8 @@ async def submit(self, artifact, ...):
 
 Generated (`api.py`) is the same with `await`/`async` lowered. (`sandbox_file` / `sandbox_url` follow the same create → `upload_file` → finalize shape, finalizing via `_finalize_sandbox_task`.)
 
+A URL passed as a string to `submit` / `sandbox_file` (with `artifact_type=URL`) or as `sandbox_url(url)` is refanged before `LocalArtifact.from_content` when `refang_iocs` is on, so the uploaded content and the default artifact name both carry the live URL (§"IoC refanging" in [`05-downstream-contract.md`](./05-downstream-contract.md)). A QR-code submission (`preprocessing={'type': 'qrcode'}`) is the exception on both `submit` and `sandbox_file`: its argument names an image file, not a URL, and is passed on unchanged.
+
 `upload_file` is a method on the session class (`AsyncPolyswarmSession.upload_file` / `PolyswarmSession.upload_file`). Both strip the session-level `Authorization` header so the PolySwarm API key doesn't leak to the pre-signed S3 origin. Downstream consumers customize behaviour by subclassing the session — see [`05-downstream-contract.md`](./05-downstream-contract.md) §"Customizing transport behaviour".
 
 **Report-template logo upload is different**: `report_template_logo_upload` PUTs to the PolySwarm endpoint `/reports/templates/logo`, which is authenticated, *not* a pre-signed S3 URL. It builds a normal `PolyswarmRequest` descriptor via `ReportTemplate.upload_logo(...)` and dispatches it through `session.execute` like any other endpoint — the API key must ride along on the request. There is no `session.upload_logo` method.

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

specs/05-downstream-contract.md

@@ -37,6 +37,7 @@ __version__: str
 __release_url__: str
 api: module                          # contains PolyswarmAPI
 exceptions: module                   # contains the exception hierarchy
+refang: module                       # IoC refanging helpers — see "IoC refanging" below
 PolyswarmAPI: class                  # sync client
 PolySwarmAsyncAPI: class             # async client (re-exported from polyswarm_api.aio)
 ```
@@ -48,7 +49,7 @@ PolySwarmAsyncAPI: class             # async client (re-exported from polyswarm_
 ```python
 class PolyswarmAPI:
     def __init__(self, key=None, uri=None, community=None, timeout=None,
-                 verify=True, *, session=None, **httpx_kwargs): ...
+                 verify=True, *, session=None, refang_iocs=False, **httpx_kwargs): ...
     def close(self): ...
     def __enter__(self): ...
     def __exit__(self, *exc): ...
@@ -62,7 +63,7 @@ Generated by `scripts/regenerate_sync.py` from `polyswarm_api/aio/api.py`. Publi
 ```python
 class PolySwarmAsyncAPI:
     def __init__(self, key=None, uri=None, community=None, timeout=None,
-                 verify=True, *, session=None, **httpx_kwargs): ...
+                 verify=True, *, session=None, refang_iocs=False, **httpx_kwargs): ...
     async def aclose(self): ...
     async def __aenter__(self): ...
     async def __aexit__(self, *exc): ...
@@ -202,6 +203,7 @@ PolyswarmAPI(
     verify: bool = True,
     *,
     session: PolyswarmSession | None = None,    # pre-built session; mutually exclusive with `**httpx_kwargs`
+    refang_iocs: bool = False,                  # opt in to refanging defanged URL/domain/IP inputs — see "IoC refanging"
     **httpx_kwargs,                             # forwarded to httpx.Client when constructing default session
 )
 ```
@@ -212,6 +214,24 @@ Same shape for `PolySwarmAsyncAPI` with `AsyncPolyswarmSession`.
 
 `**httpx_kwargs` is forwarded to `httpx.{,Async}Client` (formerly to `requests.Session` in 3.x). Kwargs that worked on both (`timeout`, `verify`, `headers`) are unchanged. `requests`-only kwargs (e.g. `proxies` as a dict-of-protocol-strings) need translation to httpx's equivalent.
 
+## IoC refanging
+
+Threat-intel reports print indicators defanged (`hxxps[:]//evil[.]com`, `127[.]0[.]0[.]1`). The server looks URLs up by an exact hash of the string and stores a submitted URL verbatim, so a defanged value silently misses a search or becomes a broken URL artifact — and the server deliberately does not guess. The SDK therefore offers to refang at its own edge (added in 4.6.0, opt-in).
+
+**`polyswarm_api.refang`** (public, pure, no I/O):
+
+| Function | Contract |
+|---|---|
+| `refang_text(text)` | Every rewrite, ungated and untrimmed: `[://]`→`://`, `[:]`→`:`, `[/]`→`/`, `[.]` `(.)` `{.}` `[dot]` `(dot)` `{dot}`→`.` (inner whitespace allowed), then the anchored schemes `hxxps`/`h**ps`→`https`, `hxxp`/`h**p`→`http`, `fxps`→`ftps`, `fxp`→`ftp`. |
+| `is_network_ioc(candidate)` | URL (optional http(s)/ftp(s) scheme, userinfo, port, path), domain, IPv4, or bracketed IPv6 host. |
+| `refang_ioc(value, accept=None)` | The gated form: returns the refanged value only when something was defanged, the result has no whitespace or `"`, the rewrite does not keep the input's scheme and host intact (if it does, only a path would change, so `example.com/a[.]b` and `https://example.com/a[.]b` both survive), the result is a network IoC, and the optional `accept(candidate)` agrees. Otherwise returns `value` unchanged (untrimmed). Non-strings pass through. |
+
+Invariants: the rules, their order, the gate and the case table (`test/fixtures/refang_cases.json`) are a contract other PolySwarm clients implement too — the table is kept byte-identical across them, so changing a rule means changing it everywhere. Out of scope: email `[at]`, bare-word ` dot `, `http__host` / `http:\\host`, bare-bracket stripping (IPv6 syntax), non-ASCII hosts.
+
+**Where the client applies it** (when constructed with `refang_iocs=True`; the default is `False`): `search_url(url)`; `search_by_metadata(ips=, urls=, domains=)` — **never** the free-form `query`; `search_by_ioc(ip=, domain=)`; `check_known_hosts(ips=, domains=)`; the `host` of `add_known_good_host` / `add_known_bad_host` / `update_known_good_host`; and the URL of every URL submission (`submit(..., artifact_type=URL)` and `sandbox_file(..., artifact_type=URL)` from a string, `sandbox_url(url)`) — both the uploaded content and the default artifact name. An explicit `artifact_name` is kept as given. Hashes and ids are never touched, and neither is the argument of a QR-code submission (`preprocessing={'type': 'qrcode'}` on `submit` or `sandbox_file`): it names an image file, not a URL, so a file called `qr[.]png` stays `qr[.]png`. With the default `refang_iocs=False` every input is sent verbatim, exactly as in 4.5.0. The setting is kept as the public attribute `api.refang_iocs`, so a consumer that handles a value outside the endpoint methods (its own validation, or a request it builds with `_single`) can apply `refang.refang_ioc` under the same switch — the CLI does exactly that.
+
+Compatibility: refanging is **opt-in**, so the default preserves the 4.5.0 behaviour exactly and 4.6.0 falls under the "new optional keyword argument … with a default that preserves the current behaviour" row of the versioning table (minor). The default matters beyond searches: turning refanging on changes what the write paths store — a known-host row written verbatim as `evil[.]com` by an earlier client is no longer matched by a lookup that now sends `evil.com`, and a URL submission creates a different artifact (content, name, sha) than the same defanged call did before. A consumer opts in knowing that; the CLI does so by default through its `--refang/--no-refang` option.
+
 ## Customizing transport behaviour
 
 Replaces the 3.x module-level monkey-patching pattern with subclassing.

src/polyswarm_api/__init__.py

@@ -1,9 +1,10 @@
 # https://www.python.org/dev/peps/pep-0008/#module-level-dunder-names
-__version__ = '4.5.0'
+__version__ = '4.6.0'
 __release_url__ = 'https://api.github.com/repos/polyswarm/polyswarm-api/releases/latest'
 
 from . import api
 from . import exceptions
+from . import refang
 from .api import PolyswarmAPI
 from .session import PolyswarmSession
 from .aio import PolySwarmAsyncAPI, AsyncPolyswarmSession

src/polyswarm_api/aio/api.py

@@ -17,7 +17,7 @@
 import logging
 import time
 
-from polyswarm_api import exceptions, resources, settings
+from polyswarm_api import exceptions, refang, resources, settings
 from polyswarm_api.core import PolyswarmRequest, _as_result_bound
 
 from .session import AsyncPolyswarmSession
@@ -53,6 +53,7 @@ def __init__(
         verify: bool = True,
         *,
         session: AsyncPolyswarmSession | None = None,
+        refang_iocs: bool = False,
         **httpx_kwargs,
     ):
         key_masked = '******' + (key[-4:] if key and len(key) > 16 else '')
@@ -64,6 +65,11 @@ def __init__(
         self.community = community or settings.DEFAULT_COMMUNITY
         self.timeout = timeout or settings.DEFAULT_HTTP_TIMEOUT
         self.verify = verify
+        # Refang defanged URL / domain / IP inputs (``hxxps[:]//evil[.]com``)
+        # before building a request. Opt-in: the default preserves the 4.5.0
+        # behaviour of sending every input verbatim. See ``polyswarm_api.refang``
+        # and ``_refang`` below for exactly which inputs are touched.
+        self.refang_iocs = refang_iocs
         self._engines = None
         # Either accept a pre-built session (customization point) or
         # build the default from ``key``. Passing both is ambiguous.
@@ -88,6 +94,27 @@ def __repr__(self):
         attrs = f'uri={self.uri!r}, community={self.community!r}, timeout={self.timeout!r}'
         return f'<{clsname}({attrs}) at 0x{id(self):x}>'
 
+    def _refang(self, value):
+        """Refang one URL / domain / IP input when ``refang_iocs`` is on.
+
+        Applied only to arguments that name a network indicator (a URL to
+        search or submit, an ``ips=`` / ``urls=`` / ``domains=`` entry, an IoC
+        ``ip`` / ``domain``, a known-host ``host``) — never to a free-form
+        metadata query or a hash.
+        Anything that is not a defanged network IoC is returned unchanged.
+        """
+        if not self.refang_iocs:
+            return value
+        return refang.refang_ioc(value)
+
+    def _refang_all(self, values):
+        """``_refang`` over a list argument; ``None`` / empty pass through."""
+        if not values or not self.refang_iocs:
+            return values
+        if isinstance(values, str):
+            return self._refang(values)
+        return [self._refang(v) for v in values]
+
     async def aclose(self):
         """Close the underlying HTTP client."""
         await self.session.aclose()
@@ -340,6 +367,7 @@ async def search_url(self, url):
         :param url: A url to be searched by exact match
         :return: Generator of ArtifactInstance resources
         """
+        url = self._refang(url)
         logger.info('Searching for url %s', url)
         async for item in self._paginate(resources.ArtifactInstance.search_url(self, url)):
             yield item
@@ -365,6 +393,7 @@ async def search_by_metadata(self, query, include=None, exclude=None, ips=None,
         :param exclude: A list of fields to be excluded from the result (.* wildcards are accepted)
         :return: Generator of ArtifactInstance resources
         """
+        ips, urls, domains = self._refang_all(ips), self._refang_all(urls), self._refang_all(domains)
         logger.info('Searching for metadata %s', query)
         async for item in self._paginate(resources.Metadata.get(self, query=query, community=self.community, include=include, exclude=exclude, ips=ips, urls=urls, domains=domains)):
             yield item
@@ -391,6 +420,7 @@ async def search_by_ioc(self, ip=None, domain=None, ttp=None, imphash=None):
         :param imphash: ImpHash to search by
         :return: Generator of ArtifactInstance resources
         """
+        ip, domain = self._refang(ip), self._refang(domain)
         logger.info('Searching by ioc %s', dict(ip=ip, domain=domain, ttp=ttp, imphash=imphash))
         async for item in self._paginate(resources.IOC.ioc_search(self, ip=ip, domain=domain, ttp=ttp, imphash=imphash)):
             yield item
@@ -403,6 +433,7 @@ async def check_known_hosts(self, ips=[], domains=[]):
         :param domains
         :return: Generator of IOC resources
         """
+        ips, domains = self._refang_all(ips), self._refang_all(domains)
         logger.info('Checking known hosts ips: %s, domains: %s', ips, domains)
         async for item in self._paginate(resources.IOC.check_known_hosts(self, ips, domains)):
             yield item
@@ -416,6 +447,7 @@ async def add_known_good_host(self, type, source, host):
         :param host
         :return: IOC resource
         """
+        host = self._refang(host)
         logger.info('Creating known good ioc %s %s %s', type, host, source)
         return await self._single(resources.IOC.create_known_good(self, type, host, source))
 
@@ -428,6 +460,7 @@ async def add_known_bad_host(self, type, source, host):
         :param host
         :return: IOC resource
         """
+        host = self._refang(host)
         logger.info('Creating known bad ioc %s %s %s', type, host, source)
         return await self._single(resources.IOC.create_known_bad(self, type, host, source))
 
@@ -440,6 +473,7 @@ async def update_known_good_host(self, id, type, source, host, good):
         :param host
         :return: IOC resource
         """
+        host = self._refang(host)
         logger.info('Updating known good ioc %s %s %s %s', id, type, host, source)
         return await self._single(resources.IOC.update_known_good(self, id, type, host, source, good))
 
@@ -1373,11 +1407,12 @@ async def submit(
                     self, artifact, artifact_type=artifact_type, artifact_name=artifact_name
                 )
             elif artifact_type == resources.ArtifactType.URL:
-                if preprocessing and preprocessing["type"] == "qrcode":
+                if preprocessing and preprocessing.get("type") == "qrcode":
                     artifact = resources.LocalArtifact.from_path(
                         self, artifact, artifact_type=artifact_type, artifact_name=artifact_name
                     )
                 else:
+                    artifact = self._refang(artifact)
                     artifact = resources.LocalArtifact.from_content(
                         self, artifact, artifact_name=artifact_name or artifact,
                         artifact_type=artifact_type,
@@ -1480,6 +1515,10 @@ async def sandbox_file(
                     self, artifact, artifact_type=artifact_type, artifact_name=artifact_name
                 )
             elif artifact_type == resources.ArtifactType.URL:
+                # A QR-code submission's argument is an image path, not a URL
+                # (same rule as ``submit``), so it is never refanged.
+                if not (preprocessing and preprocessing.get("type") == "qrcode"):
+                    artifact = self._refang(artifact)
                 artifact = resources.LocalArtifact.from_content(
                     self, artifact, artifact_name=artifact_name or artifact,
                     artifact_type=artifact_type,
@@ -1542,6 +1581,7 @@ async def sandbox_url(
             else:
                 local = artifact
         else:
+            url = self._refang(url)
             local = resources.LocalArtifact.from_content(
                 self, url, artifact_name=artifact_name or url,
                 artifact_type=resources.ArtifactType.URL,