feat: add search filter to payload list API (#3288)

* feat: add name filter query parameter to payload list API (#3055)

Add optional `?name=` query parameter to `GET /api/v2/payloads` that
performs a case-insensitive substring match on payload filenames,
enabling callers to search/filter the payload list without retrieving
and filtering the full set client-side.

* fix: address Copilot review feedback on payload name filter

- Change name_filter annotation to Optional[str]
- Filter only on PurePosixPath.name to avoid matching directory segments
  when add_path=True (e.g. 'plugins/stockpile/payloads/file.txt' should
  only match on 'file.txt')
- Strengthen filter tests to assert non-matching payloads are excluded

* Fix flake8 E127: correct continuation line indentation in test_payloads_api.py

* Address Copilot review: use PurePath for cross-platform compatibility

Replace PurePosixPath with PurePath in both the payload handler and
tests so basename extraction works correctly on Windows where paths
may use backslash separators. Move pathlib import to module scope.

* Address Copilot review: schema fix, stronger tests, combination test

- Add allow_none=True to PayloadQuerySchema name field
- Parametrize filter tests (matches + case-insensitive)
- Strengthen assertions to verify no false positives
- Add combination test with sort=true and add_path=true

* Fix flake8 E127: continuation line indentation

---------

Co-authored-by: deacon-mp <mperry@mitre.org>

Author: deacon-mp

app/api/v2/handlers/payload_api.py

@@ -4,6 +4,7 @@
 import pathlib
 import re
 from io import IOBase
+from typing import Optional
 
 import aiohttp_apispec
 from aiohttp import web
@@ -27,14 +28,16 @@ def add_routes(self, app: web.Application):
 
     @aiohttp_apispec.docs(tags=['payloads'],
                           summary='Retrieve payloads',
-                          description='Retrieves all stored payloads.')
+                          description='Retrieves all stored payloads. Supports optional filtering by name '
+                                      '(case-insensitive substring match via the `name` query parameter).')
     @aiohttp_apispec.querystring_schema(PayloadQuerySchema)
     @aiohttp_apispec.response_schema(PayloadSchema(),
                                      description='Returns a list of all payloads in PayloadSchema format.')
     async def get_payloads(self, request: web.Request):
         sort: bool = request['querystring'].get('sort')
         exclude_plugins: bool = request['querystring'].get('exclude_plugins')
         add_path: bool = request['querystring'].get('add_path')
+        name_filter: Optional[str] = request['querystring'].get('name')
 
         cwd = pathlib.Path.cwd()
         payload_dirs = [cwd / 'data' / 'payloads']
@@ -52,6 +55,11 @@ async def get_payloads(self, request: web.Request):
         }
 
         payloads = list(payloads)
+
+        if name_filter:
+            name_filter_lower = name_filter.lower()
+            payloads = [p for p in payloads if name_filter_lower in pathlib.PurePath(p).name.lower()]
+
         if sort:
             payloads.sort()
 

app/api/v2/schemas/payload_schemas.py

@@ -5,6 +5,7 @@ class PayloadQuerySchema(schema.Schema):
     sort = fields.Boolean(required=False, load_default=False)
     exclude_plugins = fields.Boolean(required=False, load_default=False)
     add_path = fields.Boolean(required=False, load_default=False)
+    name = fields.String(required=False, load_default=None, allow_none=True)
 
 
 class PayloadSchema(schema.Schema):

tests/api/v2/handlers/test_payloads_api.py

@@ -1,4 +1,5 @@
 import os
+import pathlib
 import tempfile
 from http import HTTPStatus
 
@@ -49,6 +50,35 @@ async def test_get_payloads(self, api_v2_client, api_cookies, expected_payload_f
 
         assert filtered_payload_file_names == expected_payload_file_names
 
+    @pytest.mark.parametrize('query_name', ['payload_', 'PAYLOAD_'])
+    async def test_get_payloads_name_filter(self, api_v2_client, api_cookies, expected_payload_file_names, query_name):
+        resp = await api_v2_client.get(f'/api/v2/payloads?name={query_name}', cookies=api_cookies)
+        assert resp.status == HTTPStatus.OK
+        payload_file_names = await resp.json()
+
+        # All expected payloads should be present
+        assert expected_payload_file_names <= set(payload_file_names)
+        # Every returned payload must match the filter (no false positives)
+        assert all('payload_' in pathlib.PurePath(p).name.lower() for p in payload_file_names)
+
+    async def test_get_payloads_name_filter_no_match(self, api_v2_client, api_cookies):
+        resp = await api_v2_client.get('/api/v2/payloads?name=__no_match_xyzzy__', cookies=api_cookies)
+        assert resp.status == HTTPStatus.OK
+        assert await resp.json() == []
+
+    async def test_get_payloads_name_filter_with_sort_and_add_path(
+            self, api_v2_client, api_cookies, expected_payload_file_names):
+        resp = await api_v2_client.get('/api/v2/payloads?name=payload_&sort=true&add_path=true', cookies=api_cookies)
+        assert resp.status == HTTPStatus.OK
+        payload_paths = await resp.json()
+
+        # Results should be sorted
+        assert payload_paths == sorted(payload_paths)
+        # Every returned path's filename must match the filter
+        assert all('payload_' in pathlib.PurePath(p).name.lower() for p in payload_paths)
+        # Results should contain paths (not bare filenames)
+        assert all(os.sep in p or '/' in p for p in payload_paths)
+
     async def test_unauthorized_get_payloads(self, api_v2_client):
         resp = await api_v2_client.get('/api/v2/payloads')
         assert resp.status == HTTPStatus.UNAUTHORIZED