fix(inject_expectation): bulk_update always raises OpenAEVParsingError on OpenAEV's empty 200 response #370

Description

InjectExpectation.bulk_update() always raises OpenAEVParsingError: Failed to parse the server message, even though OpenAEV applies the update.

• PUT /api/injects/expectations/bulk is a void endpoint in OpenAEV (ExpectationApi.updateInjectExpectation(... InjectExpectationBulkUpdateInput inputs)). It answers 200 with an empty body and no Content-Type, and its OpenAPI spec declares no response content.
• OpenAEV.http_put() (pyoaev/client.py) always calls result.json() on the response. An empty body therefore raises JSONDecodeError, which is re-raised as OpenAEVParsingError.

Example

A collector matches one alert to an inject's two expectations (DETECTION and PREVENTION) and pushes both results in a single call:

api.inject_expectation.bulk_update({
    "1233ca24-87e4-4d58-b96c-c50a6e344ed1": {"collector_id": "8d2c4b7a-...", "result": "Detected", "is_success": True},
    "e22d6dbc-babf-4f91-b2ef-bed8d90710e1": {"collector_id": "8d2c4b7a-...", "result": "Prevented", "is_success": True},
})

What happens:

1. OpenAEV stores both results and recomputes the scores, then answers 200 with an empty body.
2. pyoaev raises OpenAEVParsingError: Failed to parse the server message.
3. The collector concludes the bulk upload failed and re-sends both results, one PUT /api/injects/expectations/{id} each:

WARNING [ResilientUploader] Bulk upload failed, falling back to individual updates: Failed to parse the server message
DEBUG   [ResilientUploader] Uploading individual expectation index 1233ca24-87e4-4d58-b96c-c50a6e344ed1
DEBUG   [ResilientUploader] Successfully uploaded expectation index 1233ca24-87e4-4d58-b96c-c50a6e344ed1
DEBUG   [ResilientUploader] Uploading individual expectation index e22d6dbc-babf-4f91-b2ef-bed8d90710e1
DEBUG   [ResilientUploader] Successfully uploaded expectation index e22d6dbc-babf-4f91-b2ef-bed8d90710e1
INFO    [ResilientUploader] Individual uploads completed: 2 successful, 0 failed

This happens on every cycle that has results. A batch of N results costs 1 + N requests and two score propagations per expectation instead of one request, and every run logs a misleading WARNING. OpenAEV updates the existing result for the same source in place, so no duplicate results are created.

Concerned collectors

Every collector calling inject_expectation.bulk_update. All 11 pin pyoaev = "3.261022.0" and all fall back to individual updates on any exception:

Not concerned: bulk trace upload (POST /api/inject-expectations-traces/bulk, which returns JSON), and collectors that only use the single update.

Environment

1. OS: macOS (Apple Silicon), OpenAEV run locally with the official OpenAEV-Platform/docker compose file (platform, pgsql, Elasticsearch, MinIO, RabbitMQ)
2. Version: OpenAEV platform 3.261022.0 (openaev/platform:latest), pyoaev 3.261022.0 (latest on PyPI; pyoaev/client.py on main has the same http_put)
3. Other environment details: collector microsoft-defender from OpenAEV-Platform/collectors main for the collector-side logs

Reproducible steps

Steps to create the smallest reproducible scenario:

1. Without OpenAEV (about 10 seconds). A stub answers the bulk endpoint exactly like OpenAEV does. Run uv run --no-project --with pyoaev==3.261022.0 python repro_minimal.py:

   repro_minimal.py

   # Minimal reproduction, no OpenAEV needed: a stub answering the bulk endpoint
   # exactly like OpenAEV does (200, empty body, no Content-Type).
   #   uv run --no-project --with pyoaev==3.261022.0 python repro_minimal.py
   import threading
   from http.server import BaseHTTPRequestHandler, HTTPServer
   
   from pyoaev.client import OpenAEV
   
   
   class FakeOpenAEV(BaseHTTPRequestHandler):
       def do_PUT(self):
           self.rfile.read(int(self.headers["Content-Length"]))
           self.send_response(200)
           self.send_header("Content-Length", "0")
           self.end_headers()
   
       def log_message(self, *args):
           pass
   
   
   server = HTTPServer(("127.0.0.1", 0), FakeOpenAEV)
   threading.Thread(target=server.serve_forever, daemon=True).start()
   
   api = OpenAEV(url=f"http://127.0.0.1:{server.server_port}", token="any-token")
   try:
       api.inject_expectation.bulk_update(
           {
               "584f3457-904e-42ef-9680-239f32a720ee": {
                   "collector_id": "6b1f6a2e-0c1d-4c7e-9a55-b11c0c7ec001",
                   "result": "Detected",
                   "is_success": True,
               }
           }
       )
       print("OK: bulk_update returned normally")
   except Exception as e:
       print(f"BUG: {type(e).__name__}: {e} (cause: {e.__cause__!r})")

   Output:

   BUG: OpenAEVParsingError: Failed to parse the server message (cause: JSONDecodeError('Expecting value: line 1 column 1 (char 0)'))
   

2. Against a real OpenAEV. Start OpenAEV (e.g. docker compose up -d openaev with the OpenAEV-Platform/docker compose file), then run:
   OPENAEV_URL=http://localhost:8080 OPENAEV_TOKEN=<admin token> uv run --no-project --with pyoaev==3.261022.0 python repro_openaev.py

   The script registers a collector (as collector daemons do) and registers an endpoint with an agent. It then creates a command payload and launches an atomic testing on that endpoint, which creates real pending DETECTION/PREVENTION expectations. Finally it bulk-updates them like the collectors do.

   repro_openaev.py

   # Full reproduction against a real OpenAEV (e.g. the OpenAEV-Platform/docker compose).
   #   OPENAEV_URL=http://localhost:8080 OPENAEV_TOKEN=<admin token> \
   #     uv run --no-project --with pyoaev==3.261022.0 python repro_openaev.py
   import base64
   import os
   import time
   import uuid
   
   from pyoaev.client import OpenAEV
   
   api = OpenAEV(url=os.environ["OPENAEV_URL"], token=os.environ["OPENAEV_TOKEN"])
   run = uuid.uuid4().hex[:8]
   
   # 1. Register a collector, as every collector daemon does at startup
   collector_id = str(uuid.uuid4())
   icon = base64.b64decode(
       "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
   )
   doc = api.document.upsert(document={}, file=(f"{collector_id}.png", icon, "image/png"))
   platform = api.security_platform.upsert(
       {
           "asset_name": f"Bulk repro EDR {run}",
           "asset_external_reference": collector_id,
           "security_platform_type": "EDR",
           "security_platform_logo_light": doc["document_id"],
           "security_platform_logo_dark": doc["document_id"],
       }
   )
   api.collector.create(
       {
           "collector_id": collector_id,
           "collector_name": f"Bulk repro collector {run}",
           "collector_type": "openaev_bulk_repro",
           "collector_period": 60,
           "collector_security_platform": platform["asset_id"],
       },
       (f"{collector_id}.png", icon, "image/png"),
   )
   
   # 2. Get pending DETECTION/PREVENTION expectations: an endpoint with an agent,
   #    a command payload, and an atomic testing launched on that endpoint
   endpoint = api.http_post(
       "/endpoints/register",
       post_data={
           "asset_external_reference": f"bulk-repro-{run}",
           "asset_name": f"bulk-repro-{run}",
           "asset_hostname": f"bulk-repro-{run}",
           "asset_ips": ["192.0.2.50"],
           "endpoint_platform": "Windows",
           "endpoint_arch": "x86_64",
           "endpoint_agent_version": "1.0.0",
           "agent_is_service": True,
           "agent_is_elevated": True,
           "agent_executed_by_user": "SYSTEM",
           "agent_installation_mode": "service",
           "agent_installation_directory": "C:\\openaev",
           "agent_service_name": "openaev-agent",
       },
   )
   payload = api.http_post(
       "/payloads",
       post_data={
           "payload_type": "Command",
           "payload_name": f"bulk repro {run}",
           "payload_source": "MANUAL",
           "payload_status": "VERIFIED",
           "payload_platforms": ["Windows"],
           "payload_execution_arch": "ALL_ARCHITECTURES",
           "payload_expectations": ["DETECTION", "PREVENTION"],
           "payload_domains": [],
           "command_executor": "psh",
           "command_content": "echo bulk repro",
       },
   )
   contracts = api.http_post("/injector_contracts/search", post_data={"page": 0, "size": 500})
   # the payload's contract is labelled with the payload name
   contract = next(
       c
       for c in contracts["content"]
       if (c.get("injector_contract_labels") or {}).get("en") == payload["payload_name"]
   )
   inject = api.http_post(
       "/atomic-testings",
       post_data={
           "inject_title": f"bulk repro {run}",
           "inject_injector_contract": contract["injector_contract_id"],
           "inject_assets": [endpoint["asset_id"]],
           "inject_content": {"expectations": []},
       },
   )
   api.http_post(f"/atomic-testings/{inject['inject_id']}/launch")
   
   
   def pending():
       return [
           e
           for e in api.http_get(f"/injects/expectations/assets/{collector_id}")
           if e["inject_expectation_inject"] == inject["inject_id"]
       ]
   
   
   for _ in range(20):
       expectations = pending()
       if expectations:
           break
       time.sleep(3)
   print("pending before:", sorted(e["inject_expectation_type"] for e in expectations))
   
   # 3. Bulk update them, exactly like the collectors do
   try:
       api.inject_expectation.bulk_update(
           {
               e["inject_expectation_id"]: {
                   "collector_id": collector_id,
                   "result": "Detected",
                   "is_success": True,
               }
               for e in expectations
           }
       )
       print("OK: bulk_update returned normally")
   except Exception as e:
       print(f"BUG: {type(e).__name__}: {e} (cause: {e.__cause__!r})")
   
   # 4. ...yet OpenAEV applied the update
   print("pending after:", sorted(e["inject_expectation_type"] for e in pending()))

   Output:

   pending before: ['DETECTION', 'PREVENTION']
   BUG: OpenAEVParsingError: Failed to parse the server message (cause: JSONDecodeError('Expecting value: line 1 column 1 (char 0)'))
   pending after: []
   

   pending after: [] shows that OpenAEV applied the update the client reported as failed.

3. With a collector. Run any concerned collector against OpenAEV with at least one expectation to answer. The Bulk upload failed, falling back to individual updates: Failed to parse the server message warning appears on every cycle that pushes results.

Expected output

bulk_update() returns normally when OpenAEV answers 2xx, so collectors send one bulk request per cycle and log Successfully bulk upload N expectation.

Actual output

bulk_update() raises OpenAEVParsingError: Failed to parse the server message (cause JSONDecodeError: Expecting value: line 1 column 1 (char 0)) although the update was applied. The stored result in injects_expectations.inject_expectation_results:

DETECTION|100|[{"sourceId":"<collector_id>","sourceType":"collector","sourcePlatform":"EDR","score":100.0,"result":"Detected",...}]

The collector then re-sends every result individually, as shown in the example above.

Additional information

Proposed fix: make http_put treat an empty 2xx body as success instead of parsing it:

result = self.http_request("put", path, ...)
if not result.content:
    return {}
return result.json()

I tested this patch against the same OpenAEV 3.261022.0 instance:

• bulk_update() returned normally.
• Both results were applied.
• inject_expectation.update() still returned its parsed JSON body.

http_post and the other helpers may deserve the same guard for other void endpoints. After the fix is released, the 11 collectors above need a pyoaev bump to stop the fallback.