feat: add metadata_raw_search for the raw metadata search endpoint

specs/02-resources.md

@@ -31,6 +31,7 @@ BaseResource                          # holds .api, ._content, .parse_result cla
     ├── Metadata                      # /search/metadata/query
     ├── MetadataFieldProperties       # /search/metadata/properties
     ├── MetadataMapping               # /search/metadata/mappings
+    ├── MetadataRawSearch             # /search/metadata/raw
     ├── IOC                           # /ioc (known good / known bad / search)
     ├── LiveYaraRuleset               # /hunt/rule/live
     ├── HistoricalHunt                # /hunt/historical

specs/03-endpoints.md

@@ -35,6 +35,7 @@ The full catalogue of methods on the public client surface and which transport h
 | Method | Resource builder |
 |---|---|
 | `metadata_mapping()` | inline (GET) — parsed into `MetadataMapping` |
+| `metadata_raw_search(body, community=None)` | `MetadataRawSearch.search` (POST). `body` is sent verbatim: it is an OpenSearch `_search` body, and the server owns the grammar it accepts. `community` rides the **query string**, not the body, because the server overlays the community filter itself. |
 | `metadata_field_properties_write(field_path, description, …)` | inline (POST) |
 | `metadata_field_properties_get(field_path)` | inline (GET) |
 | `metadata_field_properties_delete(field_path)` | inline (DELETE) |

specs/04-testing.md

@@ -26,6 +26,7 @@ How the test suite is organised. Three layers: pure unit tests (no HTTP at all 
 - `test/async_client_test.py` — async, VCR-backed integration tests (not yet on the parametrised harness — follow-up work).
 - `test/hunt_matched_strings_test.py` — pure-unit tests for the three-state `matched_strings` contract and its `matched_strings_dropped` sibling, across all four hunt-result classes. The endpoint behaviour they cannot see (that the detail route actually emits the keys and the list route sends `null`) is pinned live in `client_scan_test.py::test_live` / `async_client_test.py::test_async_live` — the e2e-first + pure-unit pairing invariant 1 asks for.
 - `test/known_good_test.py` — pure-unit tests for the known-good resource fields.
+- `test/metadata_raw_search_test.py` — pure-unit request-shape and parse tests for `metadata_raw_search`: that the query body is sent verbatim and `community` rides the query string (invariant 7 — the builder is pure data, so the routing reads straight off the `PolyswarmRequest`'s `request_parameters`, with no httpx), plus the `partial` reason codes the server returns. The lifecycle is live, in `client_scan_test.py` / `async_client_test.py`.
 - `test/jmespath_test.py` — unit tests for `BaseJsonResource.jmespath`.
 - `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).

src/polyswarm_api/aio/api.py

@@ -288,6 +288,51 @@ async def metadata_mapping(self):
             result_parser=resources.MetadataMapping,
         )
 
+    async def metadata_raw_search(self, body, community=None):
+        """Run a raw OpenSearch query body against the metadata corpus.
+
+        The server overlays the community filter itself and caps the work, so
+        ``body`` is a plain ``_search`` body and the community travels in the
+        query string rather than in it. The accepted grammar is closed: a
+        construct the server does not recognise is refused rather than passed
+        through, and so is a top-level ``community``/``communities`` key.
+
+        Entitlement-gated and metered per query, including a query the server
+        refuses after running it.
+
+        :param body: An OpenSearch ``_search`` body (``query``, ``aggs``,
+            ``size``, ``sort``, ``_source``, ...). Omitting ``size`` returns
+            **no documents**: the server defaults it to 0 so an aggregation
+            caller does not pay for hits they did not ask for. Ask for hits
+            by setting it.
+        :param community: Community to search; defaults to the client's.
+        :return: A ``MetadataRawSearch`` whose ``json`` carries the engine's
+            ``hits`` and ``aggregations``, plus ``partial`` when the server
+            could not establish that the answer is complete.
+
+            ``partial`` is a **list to inspect, not a flag**. Its entries are
+            ``{'code': ..., 'message': ...}`` and they make different claims,
+            so branch on ``code`` - ``message`` is for a human and may be
+            reworded. The codes are a closed set:
+
+            * ``timed_out`` / ``terminated_early`` - the search stopped early.
+            * ``shards_failed`` - a shard was lost; also carries ``failed``.
+            * ``buckets_clamped`` - a ``terms`` list dropped terms.
+            * ``composite_paged`` - a ``composite`` may have more pages; page
+              it with its ``after_key``.
+            * ``bucket_count_approximate`` - a ``terms`` count is approximate
+              rather than truncated.
+            * ``hit_total_capped`` - the hit counter stopped at
+              ``track_total_hits``. This is what *every* large query that asks
+              for hits returns, so on its own it says nothing about the
+              aggregations; the server omits it when the body asked for no
+              documents. ``hits.total.relation`` carries the same fact as the
+              engine's own field, ``eq`` exact and ``gte`` capped.
+        """
+        logger.info('Running a raw metadata search')
+        return await self._single(
+            resources.MetadataRawSearch.search(self, body, community))
+
     async def metadata_field_properties_write(self, field_path, description,
                                         example=None, category=None, aliases=None):
         """Upsert a metadata field properties entry (the single write path).

src/polyswarm_api/api.py

@@ -303,6 +303,50 @@ def metadata_mapping(self):
             result_parser=resources.MetadataMapping,
         )
 
+    def metadata_raw_search(self, body, community=None):
+        """Run a raw OpenSearch query body against the metadata corpus.
+
+        The server overlays the community filter itself and caps the work, so
+        ``body`` is a plain ``_search`` body and the community travels in the
+        query string rather than in it. The accepted grammar is closed: a
+        construct the server does not recognise is refused rather than passed
+        through, and so is a top-level ``community``/``communities`` key.
+
+        Entitlement-gated and metered per query, including a query the server
+        refuses after running it.
+
+        :param body: An OpenSearch ``_search`` body (``query``, ``aggs``,
+            ``size``, ``sort``, ``_source``, ...). Omitting ``size`` returns
+            **no documents**: the server defaults it to 0 so an aggregation
+            caller does not pay for hits they did not ask for. Ask for hits
+            by setting it.
+        :param community: Community to search; defaults to the client's.
+        :return: A ``MetadataRawSearch`` whose ``json`` carries the engine's
+            ``hits`` and ``aggregations``, plus ``partial`` when the server
+            could not establish that the answer is complete.
+
+            ``partial`` is a **list to inspect, not a flag**. Its entries are
+            ``{'code': ..., 'message': ...}`` and they make different claims,
+            so branch on ``code`` - ``message`` is for a human and may be
+            reworded. The codes are a closed set:
+
+            * ``timed_out`` / ``terminated_early`` - the search stopped early.
+            * ``shards_failed`` - a shard was lost; also carries ``failed``.
+            * ``buckets_clamped`` - a ``terms`` list dropped terms.
+            * ``composite_paged`` - a ``composite`` may have more pages; page
+              it with its ``after_key``.
+            * ``bucket_count_approximate`` - a ``terms`` count is approximate
+              rather than truncated.
+            * ``hit_total_capped`` - the hit counter stopped at
+              ``track_total_hits``. This is what *every* large query that asks
+              for hits returns, so on its own it says nothing about the
+              aggregations; the server omits it when the body asked for no
+              documents. ``hits.total.relation`` carries the same fact as the
+              engine's own field, ``eq`` exact and ``gte`` capped.
+        """
+        logger.info("Running a raw metadata search")
+        return self._single(resources.MetadataRawSearch.search(self, body, community))
+
     def metadata_field_properties_write(
         self, field_path, description, example=None, category=None, aliases=None
     ):

src/polyswarm_api/resources.py

@@ -77,6 +77,25 @@ class MetadataMapping(core.BaseJsonResource):
     RESOURCE_ENDPOINT = '/search/metadata/mappings'
 
 
+class MetadataRawSearch(core.BaseJsonResource):
+    RESOURCE_ENDPOINT = '/search/metadata/raw'
+
+    @classmethod
+    def search(cls, api, body, community=None):
+        """``body`` is the caller's whole query and travels untouched. The
+        community is scope, not query, and the server reads it from the query
+        string - folding it into the body would pick scope with a key the
+        server's grammar then rejects."""
+        return core.PolyswarmRequest(
+            api=api,
+            method='POST',
+            url=f'{api.uri}{cls.RESOURCE_ENDPOINT}',
+            params={'community': community or api.community},
+            json=body,
+            result_parser=cls,
+        )
+
+
 class MetadataFieldProperties(core.BaseJsonResource):
     RESOURCE_ENDPOINT = '/search/metadata/properties'
     RESOURCE_ID_KEYS = ['field_path']

test/async_client_test.py

@@ -286,6 +286,34 @@ async def test_async_metadata_search(self, uid):
                 lambda: api.search_by_metadata(f'artifact.sha256:{sha}'), tries=90)
         assert result and result[0].sha256 == sha
 
+    @vcr.use_cassette()
+    async def test_async_metadata_raw_search(self, uid):
+        # Async twin of the sync test_metadata_raw_search; own cassette, same
+        # exchange. See that test for what the two assertions are covering.
+        async with self._api() as api:
+            _, sha = await submit_and_scan(api, uid)
+            body = {
+                'query': {'term': {'artifact.sha256': sha}},
+                'aggs': {'extensions': {'terms': {'field': 'extension', 'size': 5}}},
+                'size': 1,
+            }
+
+            async def _hit_count():
+                # A 404 is the route missing, not the index lagging - see
+                # _raw_hit_count in client_scan_test.py.
+                try:
+                    result = await api.metadata_raw_search(body)
+                except exceptions.NotFoundException:
+                    pytest.fail('POST /search/metadata/raw answered 404: the '
+                                'route is not on this stack. The server change '
+                                'merges first.')
+                return result.json['hits']['total']['value']
+
+            assert await poll_equals_async(_hit_count, 1, tries=90) == 1
+            result = await api.metadata_raw_search(body)
+        assert result.json['aggregations']['extensions']['buckets']
+        assert 'partial' not in result.json
+
     # ── IOCs ──────────────────────────────────────────────────────────────────
 
     @vcr.use_cassette()

test/client_scan_test.py

@@ -69,6 +69,29 @@ def rescan_and_scan(api, sha):
     vcr.use_cassette = lambda *_a, **_k: (lambda _fn: _fn)
 
 
+def _raw_hit_count(api, body):
+    """A reader for ``poll_equals`` that tells a missing route from a missing
+    document.
+
+    poll_equals treats NotFoundException as "the index has not caught up", which
+    is right for a document and wrong for an endpoint: a search that matches
+    nothing is a 200 with total 0, so a 404 here means the route is absent.
+    Absorbed as lag it costs 90 polls and then fails as ``assert None == 1``,
+    which names nothing.
+
+    It fails rather than skips, and stays after the route ships: the e2e job
+    runs ``TESTS_VCR=off`` against a live stack, so a skip here would let a
+    renamed or withdrawn route report green and promote ``:latest``.
+    """
+    def read():
+        try:
+            return api.metadata_raw_search(body).json['hits']['total']['value']
+        except exceptions.NotFoundException:
+            pytest.fail('POST /search/metadata/raw answered 404: the route is '
+                        'not on this stack. The server change merges first.')
+    return read
+
+
 @contextmanager
 def temp_dir(files_dict):
     with tempfile.TemporaryDirectory() as tmp_dir:
@@ -364,6 +387,26 @@ def test_metadata_search(self):
             lambda: api.search_by_metadata(f'artifact.sha256:{sha}'), tries=90)
         assert result and result[0].sha256 == sha
 
+    @vcr.use_cassette()
+    def test_metadata_raw_search(self):
+        api = PolyswarmAPI(self.test_api_key, uri=f'http://ai:9696/{self.api_version}', community='gamma')
+        # The raw endpoint runs the caller's own query body, so this covers the
+        # two halves the SDK owns: the body survives the trip intact enough to
+        # match a document we just created, and an aggregation comes back beside
+        # the hits. The metadata write lags the scan, so poll for the hit first.
+        _, sha = submit_and_scan(api, self._testMethodName)
+        body = {
+            'query': {'term': {'artifact.sha256': sha}},
+            'aggs': {'extensions': {'terms': {'field': 'extension', 'size': 5}}},
+            'size': 1,
+        }
+        found = poll_equals(_raw_hit_count(api, body), 1, tries=90)
+        assert found == 1
+        result = api.metadata_raw_search(body)
+        assert result.json['aggregations']['extensions']['buckets']
+        # A complete answer carries no marker; a truncated one would say so.
+        assert 'partial' not in result.json
+
     def test_resolve_engine_name(self):
         with respx.mock(assert_all_called=False) as router:
             ok_payload = {'results': [

test/metadata_raw_search_test.py

@@ -0,0 +1,107 @@
+"""Pure-unit tests for the ``metadata_raw_search`` builder and its parse.
+
+Request-shape assertions belong in this tier rather than respx
+(``specs/04-testing.md`` invariant 7): ``MetadataRawSearch.search`` is pure
+data, so the body-vs-query routing reads straight off the ``PolyswarmRequest``
+with no httpx. The live lifecycle test is in ``client_scan_test.py`` /
+``async_client_test.py``.
+
+The routing is load-bearing rather than stylistic: the server reads
+``community``/``communities`` out of a POST body when it resolves scope, so a
+client that folded the community into the body would pick its scope with a key
+the server's own grammar then rejects.
+
+Both transports share this builder, so one assertion covers each of them.
+
+Assertions read ``request_parameters``, the projection of what is *sent*.
+``PolyswarmRequest.json`` is reused for the parsed *response* and is None
+until the call runs - see ``core_test`` for that trap.
+"""
+from polyswarm_api import resources
+from polyswarm_api.core import parse_response
+
+
+_EMPTY_HITS = {'total': {'value': 0, 'relation': 'eq'}, 'hits': []}
+
+
+class _FakeApi:
+    uri = 'https://api.example.test'
+    community = 'gamma'
+
+
+class _FakeResponse:
+    """Minimal stand-in for httpx.Response that ``parse_response`` reads."""
+
+    def __init__(self, status_code=200, body=None, headers=None):
+        self.status_code = status_code
+        self._body = body
+        self.headers = headers or {}
+
+    def json(self):
+        return self._body
+
+
+def _build(body, community=None):
+    return resources.MetadataRawSearch.search(_FakeApi(), body, community)
+
+
+def _parsed(result):
+    req = _build({'query': {'match_all': {}}})
+    parse_response(_FakeResponse(body={'status': 'OK', 'result': result}), req)
+    return req._result
+
+
+# ── Request shape ────────────────────────────────────────────────────
+
+
+class TestMetadataRawSearchRequest:
+    def test_the_body_reaches_the_server_verbatim(self):
+        body = {
+            'query': {'bool': {'must': [{'term': {'scan.mimetype': 'application/zip'}}]}},
+            'aggs': {'families': {'terms': {'field': 'polyunite.malware_family', 'size': 20}}},
+            'size': 0,
+            'sort': [{'artifact.created': {'order': 'desc'}}],
+        }
+        assert _build(body).request_parameters['json'] == body
+
+    def test_the_community_rides_the_query_string_not_the_body(self):
+        sent = _build({'query': {'match_all': {}}}).request_parameters
+        assert sent['params'] == {'community': 'gamma'}
+        assert 'community' not in sent['json']
+        assert 'communities' not in sent['json']
+
+    def test_an_explicit_community_overrides_the_clients(self):
+        sent = _build({'query': {'match_all': {}}}, community='delta').request_parameters
+        assert sent['params']['community'] == 'delta'
+
+    def test_it_posts_to_the_raw_search_endpoint(self):
+        sent = _build({'query': {'match_all': {}}}).request_parameters
+        assert sent['method'] == 'POST'
+        assert sent['url'] == 'https://api.example.test/search/metadata/raw'
+
+
+# ── Parse ────────────────────────────────────────────────────────────
+
+
+class TestMetadataRawSearchParse:
+    def test_hits_and_aggregations_are_both_reachable(self):
+        parsed = _parsed({
+            'hits': {'total': {'value': 2, 'relation': 'eq'},
+                     'hits': [{'_source': {'artifact': {'sha256': 'a' * 64}}}]},
+            'aggregations': {'families': {'buckets': [{'key': 'lockbit', 'doc_count': 2}]}},
+        })
+        assert parsed.json['hits']['total']['value'] == 2
+        assert parsed.json['aggregations']['families']['buckets'][0]['key'] == 'lockbit'
+
+    def test_a_partial_answer_keeps_its_reasons(self):
+        # A truncated answer is still a 200, so this list is the only thing
+        # separating it from a complete one. Dropping it in the parse layer
+        # would turn "we could not establish this is complete" into a number.
+        reasons = [{'code': 'terminated_early',
+                    'message': 'the search stopped at terminate_after'}]
+        parsed = _parsed({'hits': _EMPTY_HITS, 'partial': reasons})
+        assert parsed.json['partial'] == reasons
+        assert parsed.json['partial'][0]['code'] == 'terminated_early'
+
+    def test_a_complete_answer_carries_no_marker(self):
+        assert 'partial' not in _parsed({'hits': _EMPTY_HITS}).json