docs+fix: audit live n8n state, require auth on the return webhook
Inspected the shared n8n instance (n8n.itworx.tech) live: both existing Fleet Ops workflows are genuinely active and structurally match the repo, but the shared X-Service-Token secret was stored as plaintext literal text in both HTTP Request nodes (exportable in the clear), and the production return webhook had n8n-level Authentication set to "None" (publicly callable by anyone who discovered the URL). Findings recorded in docs/live-ai-integration/n8n-current-state.md. Fixed on the n8n side (both workflows published): the shared token now lives in a single Header Auth credential instead of two literal copies; the return webhook now requires a second, distinct Header Auth credential. Fixed on the Fleet Ops side to match: the outbox dispatcher now sends the new X-Fleet-Ops-Trigger-Token header (new MOBILITYOPS_WEBHOOK_TRIGGER_TOKEN setting) when calling the webhook. Live-verified against the real webhook: a request with no header is now rejected (403); a request with the correct header passes n8n's auth and reaches Fleet Ops's own business logic. That same live test also surfaced a real robustness gap: an n8n execution that errors before its "Respond to Webhook" node runs can still answer with a 2xx status and an empty body, which made response.json() raise an uncaught exception, potentially leaving the outbox event stuck in "delivering". Now treated as an explicit, retryable failure (error_code=malformedResponse), with a regression test reproducing the exact case.
This commit is contained in:
@@ -23,6 +23,7 @@ class Settings(BaseSettings):
|
||||
ragcore_api_token: str = ""
|
||||
ragcore_http_timeout_seconds: float = 5.0
|
||||
n8n_webhook_url: str = "http://n8n:5678/webhook/mobilityops-return"
|
||||
n8n_webhook_trigger_token: str = "replace-me-n8n-webhook-trigger-token"
|
||||
n8n_callback_token: str = "replace-me-n8n-callback-token"
|
||||
n8n_dispatch_enabled: bool = True
|
||||
n8n_dispatch_interval_seconds: float = 3.0
|
||||
|
||||
@@ -121,13 +121,30 @@ def _deliver_one(event_id: uuid.UUID) -> None:
|
||||
response = httpx.post(
|
||||
settings.n8n_webhook_url,
|
||||
json=wire_event,
|
||||
headers={"X-Fleet-Ops-Trigger-Token": settings.n8n_webhook_trigger_token},
|
||||
timeout=settings.n8n_http_timeout_seconds,
|
||||
)
|
||||
response.raise_for_status()
|
||||
body = response.json()
|
||||
success = bool(body.get("ok", True))
|
||||
error = None if success else f"n8n reported failure: {body}"
|
||||
error_code = None if success else "remoteReportedFailure"
|
||||
try:
|
||||
body = response.json()
|
||||
except ValueError:
|
||||
body = None
|
||||
if isinstance(body, dict):
|
||||
success = bool(body.get("ok", True))
|
||||
error = None if success else f"n8n reported failure: {body}"
|
||||
error_code = None if success else "remoteReportedFailure"
|
||||
else:
|
||||
# A 2xx status with a non-object (or unparsable) body means the workflow
|
||||
# itself errored before its "Respond to Webhook" node ran -- n8n's default
|
||||
# error response still carries a 2xx-looking status here. Treat it as a
|
||||
# failure so the event is retried rather than lost or wrongly marked
|
||||
# succeeded.
|
||||
success = False
|
||||
error = (
|
||||
"Unexpected non-JSON-object response from n8n "
|
||||
f"(status {response.status_code})"
|
||||
)
|
||||
error_code = "malformedResponse"
|
||||
except httpx.HTTPError as exc:
|
||||
success = False
|
||||
error = f"{type(exc).__name__}: {exc}"
|
||||
|
||||
@@ -67,7 +67,7 @@ def test_deliver_one_success(monkeypatch):
|
||||
event_id = _make_pending_event("MO-002")
|
||||
dispatcher._claim_due_events()
|
||||
|
||||
def fake_post(url, json, timeout):
|
||||
def fake_post(url, json, headers, timeout):
|
||||
return SimpleNamespace(
|
||||
raise_for_status=lambda: None,
|
||||
json=lambda: {"ok": True, "event_id": str(event_id), "result": {}},
|
||||
@@ -88,7 +88,7 @@ def test_deliver_one_failure_schedules_retry(monkeypatch):
|
||||
event_id = _make_pending_event("MO-003")
|
||||
dispatcher._claim_due_events()
|
||||
|
||||
def fake_post(url, json, timeout):
|
||||
def fake_post(url, json, headers, timeout):
|
||||
raise dispatcher.httpx.ConnectError("simulated connection failure")
|
||||
|
||||
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
||||
@@ -102,11 +102,38 @@ def test_deliver_one_failure_schedules_retry(monkeypatch):
|
||||
assert event.last_error_code == "connectionError"
|
||||
|
||||
|
||||
def test_deliver_one_treats_empty_2xx_body_as_failure(monkeypatch):
|
||||
# Reproduces a real failure mode found while live-validating the n8n webhook auth
|
||||
# fix: a workflow that errors internally before its "Respond to Webhook" node runs
|
||||
# can still answer with a 2xx status and an empty body. response.json() on that body
|
||||
# raises json.JSONDecodeError -- this must be treated as a retryable failure, not an
|
||||
# unhandled exception that leaves the event stuck in "delivering" forever.
|
||||
event_id = _make_pending_event("MO-005")
|
||||
dispatcher._claim_due_events()
|
||||
|
||||
def fake_post(url, json, headers, timeout):
|
||||
def raise_json_error():
|
||||
raise ValueError("Expecting value: line 1 column 1 (char 0)")
|
||||
|
||||
return SimpleNamespace(
|
||||
raise_for_status=lambda: None, json=raise_json_error, status_code=200
|
||||
)
|
||||
|
||||
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
||||
dispatcher._deliver_one(event_id)
|
||||
|
||||
event = _get_event(event_id)
|
||||
assert event.delivery_status == "pending"
|
||||
assert event.attempts == 1
|
||||
assert event.next_attempt_at is not None
|
||||
assert event.last_error_code == "malformedResponse"
|
||||
|
||||
|
||||
def test_deliver_one_exhausts_attempts_to_failed(monkeypatch):
|
||||
event_id = _make_pending_event("MO-004")
|
||||
settings = get_settings()
|
||||
|
||||
def fake_post(url, json, timeout):
|
||||
def fake_post(url, json, headers, timeout):
|
||||
raise dispatcher.httpx.ConnectError("still down")
|
||||
|
||||
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
||||
@@ -149,7 +176,7 @@ def test_deliver_one_handles_malformed_payload_without_getting_stuck(monkeypatch
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
def fake_post(url, json, timeout):
|
||||
def fake_post(url, json, headers, timeout):
|
||||
raise AssertionError("must not attempt delivery with a malformed payload")
|
||||
|
||||
monkeypatch.setattr(dispatcher.httpx, "post", fake_post)
|
||||
@@ -229,7 +256,7 @@ def test_run_dispatch_cycle_recovers_a_stale_lease_before_claiming(monkeypatch):
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
def fake_post(url, json, timeout):
|
||||
def fake_post(url, json, headers, timeout):
|
||||
return SimpleNamespace(
|
||||
raise_for_status=lambda: None,
|
||||
json=lambda: {"ok": True, "event_id": str(event_id), "result": {}},
|
||||
@@ -245,7 +272,7 @@ def test_run_dispatch_cycle_recovers_a_stale_lease_before_claiming(monkeypatch):
|
||||
def test_run_dispatch_cycle_end_to_end(monkeypatch):
|
||||
event_id = _make_pending_event("MO-005")
|
||||
|
||||
def fake_post(url, json, timeout):
|
||||
def fake_post(url, json, headers, timeout):
|
||||
return SimpleNamespace(
|
||||
raise_for_status=lambda: None,
|
||||
json=lambda: {"ok": True, "event_id": str(event_id), "result": {}},
|
||||
|
||||
Reference in New Issue
Block a user