From d617aceb31586f3f05e7f0097cfae285bbd7640f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?V=C3=ADctor=20Mart=C3=ADnez?= Date: Tue, 29 Sep 2026 16:00:00 -0300 Subject: [PATCH 1/3] chore: raise the SDK floor to polyswarm_api>=4.7.0 for search_by_ioc(with_artifacts=) CI installs the SDK from the same-named branch archive, so the floor names the version the paired SDK change declares: 4.7.0, the next version above the 4.6.0 the SDK's develop already declares for refanging. It is mergeable once that change is on the SDK's develop, and releasable once 4.7.0 is on PyPI. --- pyproject.toml | 2 +- specs/05-sdk-contract.md | 9 ++++++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 11692aa..1a500e8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,7 +22,7 @@ classifiers = [ ] dependencies = [ - "polyswarm_api>=4.6.0,<5.0.0", + "polyswarm_api>=4.7.0,<5.0.0", "click>=7.1", "colorama>=0.4.6", "click-log>=0.4.0", diff --git a/specs/05-sdk-contract.md b/specs/05-sdk-contract.md index 563764e..8bd2170 100644 --- a/specs/05-sdk-contract.md +++ b/specs/05-sdk-contract.md @@ -108,7 +108,7 @@ When a CLI feature needs an SDK surface that doesn't exist yet: **Read the declared version off the archive's own tree, and mind pre-release suffixes.** PEP 440 orders `4.2.0.dev1 < 4.2.0`, so a `develop` head carrying a dev suffix (the SDK's `pyproject.toml` has a `[tool.bumpversion.parts.dev]`) would *not* satisfy a `>=4.2.0` floor even though it looks like 4.2.0 — and the archive build would be silently replaced from PyPI. Check the version string in the SDK branch's `pyproject.toml` / `__init__.py`, not the last release tag. When the floor was last verified this way both were read from `origin/develop` as `4.2.0`, no suffix; the pin has since moved on (§Current floor is the one authoritative statement of its value), and every bump should be re-checked the same way. -### Current floor — `polyswarm_api>=4.6.0` +### Current floor — `polyswarm_api>=4.7.0` The floor is whatever `pyproject.toml` pins; this header follows it. It lives in ONE authoritative place for a reason — a copy here drifted behind the pin once already. The 4.2.0 rationale below still holds transitively; on 4.1.0 both behaviours fail *silently*, which is why the floor is a hard requirement rather than a preference: @@ -135,8 +135,11 @@ signature check: against a 4.4.0 SDK it fails at the mock, not at the server). I refanging — the `polyswarm_api.refang` module, the `refang_iocs=` constructor keyword and the refang inside the SDK's endpoint methods, which `--refang/--no-refang` relies on — is 4.6.0, and that is what moved the floor to 4.6.0 (on 4.5.0 the constructor rejects -`refang_iocs=` and every command fails). This is a paired change: the SDK branch of the -same name declares 4.6.0. Code and tests use them directly. +`refang_iocs=` and every command fails). `search ioc … --with-artifacts` forwards +`search_by_ioc(with_artifacts=True)`, a keyword 4.7.0 adds, and that is what moved the +floor to 4.7.0 (the autospec assertions in `tests/search_test.py` are its signature +check). Both are paired changes: the SDK branch of the same name declares the floor. +Code and tests use them directly. **Raising the floor is the whole procedure** when this repo needs something new from the SDK: From 0080b0a95e8283fce6013d4820b81d392106a3de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?V=C3=ADctor=20Mart=C3=ADnez?= Date: Wed, 23 Sep 2026 08:58:57 -0300 Subject: [PATCH 2/3] feat(search): ioc --with-artifacts returns the matching artifacts' metadata `search ioc ` prints bare sha256s. With `--with-artifacts` it asks the server for each artifact's metadata-search row and renders it with the `metadata` formatter, the block `search metadata` prints, so the text, json and hash output formats all apply. `output.ioc` cannot render these rows (it indexes IOC keys a metadata row lacks). Without the flag the SDK call is unchanged. On the sha256/sha1/md5 forward lookup the flag is refused as a usage error rather than silently ignored. Tests mock at the SDK boundary with autospec, so each call is a signature check against the SDK the floor installs. --- specs/02-commands.md | 2 +- src/polyswarm/client/search.py | 17 ++++-- tests/search_test.py | 94 ++++++++++++++++++++++++++++++++++ 3 files changed, 109 insertions(+), 4 deletions(-) diff --git a/specs/02-commands.md b/specs/02-commands.md index a41b8c7..f085ecd 100644 --- a/specs/02-commands.md +++ b/specs/02-commands.md @@ -16,7 +16,7 @@ The top-level command groups, what each is for, and the primary `polyswarm-api` | Group / commands (module) | Purpose | Primary SDK methods wrapped | |---|---|---| -| `search` (`search.py`) | Search by hash / URL / metadata / IOC / scans; metadata mapping; `field-property` subgroup | `search_hashes`, `search_urls`, `search_by_metadata`, `search_by_ioc`, `iocs_by_hash`, `search_scans`, `metadata_mapping`, `metadata_field_properties_{write,get,delete,list}` | +| `search` (`search.py`) | Search by hash / URL / metadata / IOC / scans; metadata mapping; `field-property` subgroup. `search ioc ` with an ip, domain, ttp or imphash type is the reverse IOC search and prints one sha256 per matching artifact; `--with-artifacts` asks the server for each artifact's metadata-search row instead (SDK 4.7.0, guaranteed by the pin — see [05-sdk-contract.md](./05-sdk-contract.md) §Current floor) and renders it with the `metadata` formatter, the same block `search metadata` prints, so the `sha256` / `sha1` / `md5` output formats work on it. The flag is refused (exit 2, a click `UsageError`) on the sha256 / sha1 / md5 forward lookup, which has no artifact rows to return. It needs a server that supports the parameter: an older one ignores it and answers sha256 strings, which the SDK cannot parse as metadata rows | `search_hashes`, `search_urls`, `search_by_metadata`, `search_by_ioc`, `iocs_by_hash`, `search_scans`, `metadata_mapping`, `metadata_field_properties_{write,get,delete,list}` | | `known` + `search known` (`search.py`) | Manage / check known-good/-bad hosts | `add_known_good_host`, `add_known_bad_host`, `update_known_good_host`, `delete_known_good_host`, `check_known_hosts` | | `kgb` (`kgb.py`) | Known Good **Binaries** — internal-only CRUD on a sha256-keyed known-good record (`create` / `get` / `delete`, no update). Distinct from `known`, which manages known-good/-bad *hosts*. | `known_good_create`, `known_good_get`, `known_good_delete` | | `scan`, `lookup`, `wait`, `rescan`, `rescan-id` (`scan.py`) | Submit files/dirs and await results; look up / rescan existing scans | `scan_file`, `scan_lookup`, `wait_for`, `rescan`, `rescan_id` (wrapper methods over `submit`/`lookup`/`rescan`/`rescan_id`) | diff --git a/src/polyswarm/client/search.py b/src/polyswarm/client/search.py index 5f9cd65..62ac192 100644 --- a/src/polyswarm/client/search.py +++ b/src/polyswarm/client/search.py @@ -73,13 +73,17 @@ def metadata(ctx, query_string, include, exclude, ip, url, domain): @search.command('ioc', short_help='Retrieve IOCs by artifact hash.') @click.option('-h', '--hide-known-good', type=click.BOOL, is_flag=True) +@click.option('--with-artifacts', is_flag=True, + help='For ip|domain|ttp|imphash: return the matching artifacts\' metadata ' + 'instead of bare sha256s.') @click.argument('type', required=True, type=click.Choice(['ip', 'domain', 'ttp', 'imphash', 'sha256', 'sha1', 'md5'], case_sensitive=False)) @click.argument('value', required=True) @click.pass_context -def iocs_by_hash(ctx, type, value, hide_known_good): +def iocs_by_hash(ctx, type, value, hide_known_good, with_artifacts): """ - Provide an artifact hash to get the associated IOCs. + Provide an artifact hash to get the associated IOCs, or an ip, domain, + ttp or imphash to get the artifacts that reported it. """ api = ctx.obj['api'] output = ctx.obj['output'] @@ -93,7 +97,14 @@ def iocs_by_hash(ctx, type, value, hide_known_good): elif type == 'imphash': params['imphash'] = value - if params: + if with_artifacts and not params: + raise click.UsageError('--with-artifacts applies only to ip, domain, ttp and imphash searches.') + + if with_artifacts: + # Rows are metadata-search documents, so they render like `search metadata`. + for result in api.search_by_ioc(with_artifacts=True, **params): + output.metadata(result) + elif params: for result in api.search_by_ioc(**params): output.ioc(result) else: diff --git a/tests/search_test.py b/tests/search_test.py index 931d371..986ba13 100644 --- a/tests/search_test.py +++ b/tests/search_test.py @@ -3,11 +3,13 @@ These tests mock the SDK methods directly to keep the test runtime self-contained (no live artifact-index, no VCR cassettes). """ +import json from unittest import TestCase, mock from unittest.mock import MagicMock from click.testing import CliRunner from polyswarm_api import exceptions as api_exceptions +from polyswarm_api import resources from polyswarm.client import polyswarm as client @@ -66,3 +68,95 @@ def fake_search(hash_, hash_type=None): assert '"id": 789' in result.output assert 'One or more items did not return any results' in result.output assert result.exit_code == 1, result.output + + +def _ioc_artifact_row(): + """A reverse-IOC artifact row as the server returns it under + ``with_artifacts``: a metadata-search document cut to artifact.* and the + scan summary.""" + return { + 'artifact': { + 'created': '2026-06-05T19:01:38.104756+00:00', + 'id': '31542742786663251', + 'md5': '5d3bc3c626be6b6f59195afc4ab80fc8', + 'sha1': 'f1d73dcb1286cf42e707b5d7fcf918dd45290411', + 'sha256': _FOUND_HASH, + }, + 'scan': { + 'detections': {'benign': 0, 'malicious': 1, 'total': 1}, + 'filename': ['artifact'], + 'latest_scan': {'created': '2026-06-05T19:01:38.104756+00:00', + 'polyscore': 0.97}, + 'mimetype': {'extended': 'EICAR virus test files', 'mime': 'text/plain'}, + }, + 'polyunite': {'malware_family': 'EICAR'}, + } + + +class SearchIocWithArtifactsCliTest(TestCase): + """``search ioc --with-artifacts``. + + autospec makes every call a signature check against the installed SDK, so + against an SDK below the floor (no ``with_artifacts`` keyword) these fail at + the mock. The rows are real ``Metadata`` resources: ``output.ioc`` would + KeyError on them, so the formatter choice is what is under test.""" + + def setUp(self): + self.cli = CliRunner() + + def _run(self, *cmd, fmt='text'): + return self.cli.invoke( + client.polyswarm_cli, + ['-a', _API_KEY, '-u', _API_URL, '-c', _COMMUNITY, + '--output-format', fmt] + list(cmd), + catch_exceptions=False, + ) + + def _patched(self, rows): + return mock.patch('polyswarm_api.api.PolyswarmAPI.search_by_ioc', + autospec=True, return_value=iter(rows)) + + def test_text_renders_the_metadata_block(self): + row = resources.Metadata(_ioc_artifact_row()) + with self._patched([row]) as search_by_ioc: + result = self._run('search', 'ioc', 'ip', '9.9.9.9', '--with-artifacts') + assert result.exit_code == 0, result.output + search_by_ioc.assert_called_once_with(mock.ANY, ip='9.9.9.9', with_artifacts=True) + assert 'Metadata' in result.output + assert 'Artifact id: 31542742786663251' in result.output + assert f'SHA256: {_FOUND_HASH}' in result.output + assert 'Malicious: 1' in result.output + + def test_json_renders_the_row(self): + row = resources.Metadata(_ioc_artifact_row()) + with self._patched([row]): + result = self._run('search', 'ioc', 'imphash', 'a' * 32, '--with-artifacts', + fmt='json') + assert result.exit_code == 0, result.output + assert json.loads(result.output) == _ioc_artifact_row() + + def test_sha256_output_format_prints_the_hash(self): + row = resources.Metadata(_ioc_artifact_row()) + with self._patched([row]): + result = self._run('search', 'ioc', 'domain', 'evil.test', '--with-artifacts', + fmt='sha256') + assert result.exit_code == 0, result.output + assert result.output.strip() == _FOUND_HASH + + def test_without_the_flag_the_call_is_unchanged(self): + with self._patched([]) as search_by_ioc: + result = self._run('search', 'ioc', 'ttp', 'T1081') + assert result.exit_code == 0, result.output + search_by_ioc.assert_called_once_with(mock.ANY, ttp='T1081') + + def test_the_flag_is_refused_on_a_hash_lookup(self): + with mock.patch('polyswarm_api.api.PolyswarmAPI.iocs_by_hash', + autospec=True) as iocs_by_hash, self._patched([]) as search_by_ioc: + result = self.cli.invoke( + client.polyswarm_cli, + ['-a', _API_KEY, '-u', _API_URL, '-c', _COMMUNITY, + 'search', 'ioc', 'sha256', _FOUND_HASH, '--with-artifacts']) + assert result.exit_code == 2, result.output + assert '--with-artifacts applies only to ip, domain, ttp and imphash' in result.output + iocs_by_hash.assert_not_called() + search_by_ioc.assert_not_called() From 4a99039411ad0ec5adfd3307ed7b700acf57cd9d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?V=C3=ADctor=20Mart=C3=ADnez?= Date: Wed, 23 Sep 2026 11:54:53 -0300 Subject: [PATCH 3/3] test(search): cover every hash output format on ioc --with-artifacts rows Drive sha256, sha1 and md5 through the metadata formatter on an artifact row, and name the reverse search in the command's short help. --- src/polyswarm/client/search.py | 2 +- tests/search_test.py | 16 +++++++++------- 2 files changed, 10 insertions(+), 8 deletions(-) diff --git a/src/polyswarm/client/search.py b/src/polyswarm/client/search.py index 62ac192..1276607 100644 --- a/src/polyswarm/client/search.py +++ b/src/polyswarm/client/search.py @@ -71,7 +71,7 @@ def metadata(ctx, query_string, include, exclude, ip, url, domain): output.metadata(metadata_result) -@search.command('ioc', short_help='Retrieve IOCs by artifact hash.') +@search.command('ioc', short_help='Retrieve IOCs by artifact hash, or artifacts by IOC.') @click.option('-h', '--hide-known-good', type=click.BOOL, is_flag=True) @click.option('--with-artifacts', is_flag=True, help='For ip|domain|ttp|imphash: return the matching artifacts\' metadata ' diff --git a/tests/search_test.py b/tests/search_test.py index 986ba13..ada11cf 100644 --- a/tests/search_test.py +++ b/tests/search_test.py @@ -135,13 +135,15 @@ def test_json_renders_the_row(self): assert result.exit_code == 0, result.output assert json.loads(result.output) == _ioc_artifact_row() - def test_sha256_output_format_prints_the_hash(self): - row = resources.Metadata(_ioc_artifact_row()) - with self._patched([row]): - result = self._run('search', 'ioc', 'domain', 'evil.test', '--with-artifacts', - fmt='sha256') - assert result.exit_code == 0, result.output - assert result.output.strip() == _FOUND_HASH + def test_hash_output_formats_print_the_matching_hash(self): + row_json = _ioc_artifact_row() + for fmt in ('sha256', 'sha1', 'md5'): + with self.subTest(fmt=fmt): + with self._patched([resources.Metadata(row_json)]): + result = self._run('search', 'ioc', 'domain', 'evil.test', + '--with-artifacts', fmt=fmt) + assert result.exit_code == 0, result.output + assert result.output.strip() == row_json['artifact'][fmt] def test_without_the_flag_the_call_is_unchanged(self): with self._patched([]) as search_by_ioc: