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/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/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: diff --git a/src/polyswarm/client/search.py b/src/polyswarm/client/search.py index 5f9cd65..1276607 100644 --- a/src/polyswarm/client/search.py +++ b/src/polyswarm/client/search.py @@ -71,15 +71,19 @@ 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 ' + '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..ada11cf 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,97 @@ 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_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: + 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()