mirror of
https://github.com/BigBodyCobain/Shadowbroker.git
synced 2026-07-12 15:06:37 +02:00
Fix #302: split OpenClaw HMAC reveal into dedicated POST with no-store
Reported by @tg12. Pre-fix, two problems lived on the GET endpoint:
1. `GET /api/ai/connect-info?reveal=true` returned the full HMAC
secret in the response body on every Connect modal open. Even
gated to require_local_operator, that put the secret into
browser history, dev-tools network panels, browser disk caches,
HAR exports, and screen captures.
2. The same GET endpoint auto-bootstrapped (generated + persisted)
the secret on a mere read. Side effects on a GET are a footgun:
browser prefetchers, mirror tools, and casual curl-from-history
would all silently mint+persist a fresh secret.
Backend (backend/routers/ai_intel.py)
-------------------------------------
GET /api/ai/connect-info — always returns the MASKED
fingerprint (first6 + bullets
+ last4). No `?reveal` param.
NO auto-bootstrap. When the
secret is missing, returns
`hmac_secret_set: false` and
tells the caller to POST to
/bootstrap.
POST /api/ai/connect-info/bootstrap — NEW. Mints+persists the secret
if missing. Idempotent. Never
returns the full secret in the
response body.
POST /api/ai/connect-info/reveal — NEW. Returns the full secret
with Cache-Control: no-store,
no-cache, must-revalidate +
Pragma: no-cache + Expires: 0.
POST so the body never lands
in URL history. 404 (with a
pointer to /bootstrap) when
the secret isn't set.
POST /api/ai/connect-info/regenerate — keeps existing one-time-reveal
behavior (regen IS a deliberate
destructive action triggered
by the operator). Same
no-store/no-cache headers added
so even the regen response
doesn't get cached.
Frontend (AIIntelPanel.tsx, OnboardingModal.tsx)
------------------------------------------------
* On mount: GET (masked only). If hmac_secret_set: false, fire a
transparent POST /bootstrap and refresh the masked fingerprint.
Operator sees no behavior change from pre-#302.
* Reveal (eye icon): lazy POST /reveal — secret only travels when
the operator explicitly clicks the button.
* Copy: lazy POST /reveal too — copying without a prior reveal
works exactly like before, just routed through the new endpoint.
* Regenerate: POST returns the new secret (same as before, but the
response now has no-store headers).
* The displayed snippet uses the masked fingerprint until the
operator clicks Reveal or Copy.
Tests (backend/tests/test_openclaw_connect_info_reveal.py — 13 tests)
---------------------------------------------------------------------
* GET returns masked + the full secret never appears in r.text
* GET does NOT auto-bootstrap when missing
* GET silently ignores any ?reveal=true query (back-compat noise)
* POST /bootstrap mints when missing, idempotent when set
* POST /bootstrap never returns the full secret
* POST /reveal returns the full secret with Cache-Control: no-store,
no-cache + Pragma: no-cache + Expires: 0
* POST /reveal 404s with a pointer to /bootstrap when no secret
* POST /regenerate returns the new secret with the same headers
* Anonymous remote callers get 403 on ALL FOUR endpoints (parametric
regression against the same allowlist used elsewhere).
Adjacent suites still green: test_openclaw_route_security,
test_no_new_duplicate_routes, test_control_surface_auth. 67/67 pass
locally.
Credit: @tg12 for the audit report.
This commit is contained in:
+193
-39
@@ -2521,45 +2521,85 @@ async def api_capabilities(request: Request):
|
||||
# OpenClaw Connection Management (local-operator only — NOT via HMAC)
|
||||
# These endpoints manage the HMAC secret itself, so they MUST require
|
||||
# local operator access to prevent privilege escalation.
|
||||
#
|
||||
# Issue #302 (tg12): pre-fix, GET /api/ai/connect-info had two problems:
|
||||
#
|
||||
# 1. ``?reveal=true`` made the full secret travel through every operator
|
||||
# page-load that opened the Connect modal. Even gated to
|
||||
# ``require_local_operator``, that put the secret into browser
|
||||
# history, dev-tools network panels, browser disk caches, HAR
|
||||
# exports, and screen captures. Every time the modal opened.
|
||||
#
|
||||
# 2. The same GET endpoint auto-bootstrapped (generated + persisted)
|
||||
# the secret on first read. Side effects on a GET are a footgun:
|
||||
# browser prefetchers, mirror tools, and casual curl-from-history
|
||||
# would all silently mint+persist a fresh secret. (Gated, but
|
||||
# still surprising — and noisy in the audit log.)
|
||||
#
|
||||
# Resolution:
|
||||
#
|
||||
# GET /api/ai/connect-info — always returns the MASKED
|
||||
# secret. No ?reveal param.
|
||||
# No auto-bootstrap; if the
|
||||
# secret is missing,
|
||||
# ``hmac_secret_set: false``
|
||||
# tells the frontend to call
|
||||
# /bootstrap.
|
||||
#
|
||||
# POST /api/ai/connect-info/bootstrap — NEW. Generates + persists the
|
||||
# secret if missing. Idempotent.
|
||||
# Returns metadata only, never
|
||||
# the full secret.
|
||||
#
|
||||
# POST /api/ai/connect-info/reveal — NEW. Returns the full secret in
|
||||
# the body with strict
|
||||
# ``Cache-Control: no-store,
|
||||
# no-cache, must-revalidate``
|
||||
# + ``Pragma: no-cache`` so
|
||||
# it does not land in browser
|
||||
# caches. POST means it does
|
||||
# not land in URL history.
|
||||
#
|
||||
# POST /api/ai/connect-info/regenerate — keeps existing one-time-reveal
|
||||
# behavior (regenerate IS a
|
||||
# deliberate destructive action
|
||||
# the operator triggered, so
|
||||
# displaying the new secret
|
||||
# once is the only path that
|
||||
# makes the operation useful).
|
||||
# Same no-store headers added.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@router.get("/api/ai/connect-info", dependencies=[Depends(require_local_operator)])
|
||||
@limiter.limit("30/minute")
|
||||
async def get_connect_info(request: Request, reveal: bool = False):
|
||||
"""Return connection details for the OpenClaw Connect modal.
|
||||
# Cache-Control headers that should accompany every response carrying the
|
||||
# full HMAC secret. Reused across the reveal + regenerate endpoints so a
|
||||
# future refactor that splits or renames them can't forget the headers.
|
||||
_NO_STORE_HEADERS = {
|
||||
"Cache-Control": "no-store, no-cache, must-revalidate, private",
|
||||
"Pragma": "no-cache",
|
||||
"Expires": "0",
|
||||
}
|
||||
|
||||
The HMAC secret is masked by default. Pass ?reveal=true to see the full key.
|
||||
Private keys are NEVER returned.
|
||||
|
||||
def _mask_hmac_secret(secret: str) -> str:
|
||||
"""Return a fingerprint-style mask (first6 + bullets + last4) suitable
|
||||
for display in the UI before the operator clicks Reveal."""
|
||||
if not secret:
|
||||
return ""
|
||||
if len(secret) > 10:
|
||||
return secret[:6] + "••••••••" + secret[-4:]
|
||||
return "••••••••"
|
||||
|
||||
|
||||
def _connect_info_metadata(settings) -> dict:
|
||||
"""Return everything the Connect modal needs EXCEPT the secret itself.
|
||||
|
||||
Shared between GET /api/ai/connect-info (where the full secret is
|
||||
masked) and POST /api/ai/connect-info/bootstrap (where the operator
|
||||
just generated a secret but we don't return it inline — they have to
|
||||
call /reveal to see it).
|
||||
"""
|
||||
import os
|
||||
import secrets
|
||||
from services.config import get_settings
|
||||
|
||||
settings = get_settings()
|
||||
hmac_secret = str(settings.OPENCLAW_HMAC_SECRET or "").strip()
|
||||
access_tier = str(settings.OPENCLAW_ACCESS_TIER or "restricted").strip().lower()
|
||||
|
||||
# Auto-generate if not set
|
||||
if not hmac_secret:
|
||||
hmac_secret = secrets.token_hex(24) # 48 chars
|
||||
_write_env_value("OPENCLAW_HMAC_SECRET", hmac_secret)
|
||||
# Clear settings cache so next read picks up the new value
|
||||
get_settings.cache_clear()
|
||||
|
||||
masked = hmac_secret[:6] + "••••••••" + hmac_secret[-4:] if len(hmac_secret) > 10 else "••••••••"
|
||||
|
||||
return {
|
||||
"ok": True,
|
||||
"hmac_secret": hmac_secret if reveal else masked,
|
||||
"hmac_secret_set": bool(hmac_secret),
|
||||
"bootstrap_behavior": {
|
||||
"auto_generates_when_missing": True,
|
||||
"auto_generated_this_call": not bool(settings.OPENCLAW_HMAC_SECRET or ""),
|
||||
"notes": [
|
||||
"If no HMAC secret exists yet, this endpoint bootstraps one and persists it to .env.",
|
||||
"Regenerating the HMAC secret revokes all existing direct-mode OpenClaw callers at once.",
|
||||
],
|
||||
},
|
||||
"access_tier": access_tier,
|
||||
"trust_model": {
|
||||
"remote_http_principal": "holder_of_openclaw_hmac_secret",
|
||||
@@ -2613,24 +2653,138 @@ async def get_connect_info(request: Request, reveal: bool = False):
|
||||
}
|
||||
|
||||
|
||||
@router.post("/api/ai/connect-info/regenerate", dependencies=[Depends(require_local_operator)])
|
||||
@limiter.limit("5/minute")
|
||||
async def regenerate_hmac_secret(request: Request):
|
||||
"""Generate a new HMAC secret. Old secret immediately stops working."""
|
||||
@router.get("/api/ai/connect-info", dependencies=[Depends(require_local_operator)])
|
||||
@limiter.limit("30/minute")
|
||||
async def get_connect_info(request: Request):
|
||||
"""Return connection details for the OpenClaw Connect modal.
|
||||
|
||||
The HMAC secret is always returned as a fingerprint mask
|
||||
(``first6 + bullets + last4``); the full value is only ever served by
|
||||
``POST /api/ai/connect-info/reveal`` (see #302). When the secret has
|
||||
not been bootstrapped yet, ``hmac_secret_set`` is false and the
|
||||
frontend should call ``POST /api/ai/connect-info/bootstrap``.
|
||||
|
||||
Private keys are NEVER returned.
|
||||
"""
|
||||
from services.config import get_settings
|
||||
|
||||
settings = get_settings()
|
||||
hmac_secret = str(settings.OPENCLAW_HMAC_SECRET or "").strip()
|
||||
|
||||
return {
|
||||
"ok": True,
|
||||
"masked_hmac_secret": _mask_hmac_secret(hmac_secret),
|
||||
"hmac_secret_set": bool(hmac_secret),
|
||||
"bootstrap_behavior": {
|
||||
"auto_generates_when_missing": False,
|
||||
"notes": [
|
||||
"Call POST /api/ai/connect-info/bootstrap to mint a secret on first use.",
|
||||
"Call POST /api/ai/connect-info/reveal to see the full secret (no-store).",
|
||||
"Regenerating the HMAC secret revokes all existing direct-mode OpenClaw callers at once.",
|
||||
],
|
||||
},
|
||||
**_connect_info_metadata(settings),
|
||||
}
|
||||
|
||||
|
||||
@router.post("/api/ai/connect-info/bootstrap", dependencies=[Depends(require_local_operator)])
|
||||
@limiter.limit("10/minute")
|
||||
async def bootstrap_hmac_secret(request: Request):
|
||||
"""Mint and persist the OpenClaw HMAC secret if it isn't already set.
|
||||
|
||||
Idempotent: if a secret already exists, returns ``generated: false``
|
||||
and leaves the existing secret untouched. Never returns the secret
|
||||
value in the response body — the operator calls
|
||||
``POST /api/ai/connect-info/reveal`` to see it.
|
||||
"""
|
||||
import secrets
|
||||
from services.config import get_settings
|
||||
|
||||
settings = get_settings()
|
||||
existing = str(settings.OPENCLAW_HMAC_SECRET or "").strip()
|
||||
if existing:
|
||||
return {
|
||||
"ok": True,
|
||||
"generated": False,
|
||||
"hmac_secret_set": True,
|
||||
"masked_hmac_secret": _mask_hmac_secret(existing),
|
||||
"detail": "HMAC secret already configured. Use /reveal to see it.",
|
||||
}
|
||||
|
||||
new_secret = secrets.token_hex(24) # 48 chars
|
||||
_write_env_value("OPENCLAW_HMAC_SECRET", new_secret)
|
||||
get_settings.cache_clear()
|
||||
|
||||
return {
|
||||
"ok": True,
|
||||
"hmac_secret": new_secret,
|
||||
"detail": "HMAC secret regenerated. Update your OpenClaw agent configuration.",
|
||||
"generated": True,
|
||||
"hmac_secret_set": True,
|
||||
"masked_hmac_secret": _mask_hmac_secret(new_secret),
|
||||
"detail": "HMAC secret generated. Call /reveal to copy it into your OpenClaw config.",
|
||||
}
|
||||
|
||||
|
||||
@router.post("/api/ai/connect-info/reveal", dependencies=[Depends(require_local_operator)])
|
||||
@limiter.limit("10/minute")
|
||||
async def reveal_hmac_secret(request: Request):
|
||||
"""Return the full HMAC secret in the response body.
|
||||
|
||||
POST (not GET) so the secret never lands in URL history, access logs,
|
||||
or browser visit history. Strict ``Cache-Control: no-store`` headers
|
||||
prevent intermediaries from persisting the response. Returns 404 if
|
||||
no secret has been bootstrapped — the frontend should call
|
||||
``POST /api/ai/connect-info/bootstrap`` first.
|
||||
"""
|
||||
from services.config import get_settings
|
||||
|
||||
settings = get_settings()
|
||||
hmac_secret = str(settings.OPENCLAW_HMAC_SECRET or "").strip()
|
||||
if not hmac_secret:
|
||||
raise HTTPException(
|
||||
404,
|
||||
"No HMAC secret configured. Call POST /api/ai/connect-info/bootstrap first.",
|
||||
)
|
||||
return JSONResponse(
|
||||
content={
|
||||
"ok": True,
|
||||
"hmac_secret": hmac_secret,
|
||||
"masked_hmac_secret": _mask_hmac_secret(hmac_secret),
|
||||
},
|
||||
headers=_NO_STORE_HEADERS,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/api/ai/connect-info/regenerate", dependencies=[Depends(require_local_operator)])
|
||||
@limiter.limit("5/minute")
|
||||
async def regenerate_hmac_secret(request: Request):
|
||||
"""Generate a new HMAC secret. Old secret immediately stops working.
|
||||
|
||||
Returns the new secret in the response body — this is the only
|
||||
operation where the full secret travels back through the response,
|
||||
because regenerating IS a deliberate destructive action the operator
|
||||
triggered and they need to see the new value once to update their
|
||||
OpenClaw configuration. Strict ``Cache-Control: no-store`` headers
|
||||
keep it from being persisted by browser caches, proxies, or HAR
|
||||
capture tooling.
|
||||
"""
|
||||
import secrets
|
||||
from services.config import get_settings
|
||||
|
||||
new_secret = secrets.token_hex(24) # 48 chars
|
||||
_write_env_value("OPENCLAW_HMAC_SECRET", new_secret)
|
||||
get_settings.cache_clear()
|
||||
|
||||
return JSONResponse(
|
||||
content={
|
||||
"ok": True,
|
||||
"hmac_secret": new_secret,
|
||||
"masked_hmac_secret": _mask_hmac_secret(new_secret),
|
||||
"detail": "HMAC secret regenerated. Update your OpenClaw agent configuration.",
|
||||
},
|
||||
headers=_NO_STORE_HEADERS,
|
||||
)
|
||||
|
||||
|
||||
@router.put("/api/ai/connect-info/access-tier", dependencies=[Depends(require_local_operator)])
|
||||
@limiter.limit("10/minute")
|
||||
async def set_access_tier(request: Request, body: dict):
|
||||
|
||||
@@ -0,0 +1,334 @@
|
||||
"""Issue #302 (tg12): OpenClaw connect-info HMAC secret disclosure.
|
||||
|
||||
Before this change, ``GET /api/ai/connect-info?reveal=true`` returned the
|
||||
full HMAC secret in the response body on every modal open AND the same
|
||||
GET endpoint auto-bootstrapped (generated + persisted) the secret on a
|
||||
mere read. Even gated to ``require_local_operator``, that put the full
|
||||
secret into:
|
||||
|
||||
* browser visit history
|
||||
* dev-tools network panel
|
||||
* browser disk cache
|
||||
* HAR exports
|
||||
* screen captures / shoulder-surfing
|
||||
|
||||
Every single time the OpenClaw Connect modal opened.
|
||||
|
||||
After this change:
|
||||
|
||||
GET /api/ai/connect-info — always returns the MASKED
|
||||
fingerprint. No ?reveal param.
|
||||
No side effects (auto-bootstrap
|
||||
gone).
|
||||
POST /api/ai/connect-info/bootstrap — mints+persists the secret if
|
||||
missing. Idempotent. Never
|
||||
returns the full secret.
|
||||
POST /api/ai/connect-info/reveal — returns the full secret with
|
||||
strict Cache-Control: no-store
|
||||
headers. POST so the body
|
||||
doesn't land in URL history.
|
||||
POST /api/ai/connect-info/regenerate — keeps the one-time-disclosure
|
||||
for the new secret (regen IS a
|
||||
deliberate destructive action).
|
||||
Same no-store headers added.
|
||||
|
||||
These tests pin every property.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Loopback test client. ``require_local_operator`` resolves true for
|
||||
# request.client.host == "127.0.0.1"; FastAPI's TestClient sets it to
|
||||
# "testclient" which isn't on the allowlist. Use raw ASGITransport.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def loopback():
|
||||
from main import app
|
||||
|
||||
class _Client:
|
||||
def __init__(self, peer_ip: str = "127.0.0.1"):
|
||||
self._loop = asyncio.new_event_loop()
|
||||
self._transport = ASGITransport(app=app, client=(peer_ip, 12345))
|
||||
self._base = f"http://{peer_ip}:8000"
|
||||
|
||||
def _do(self, method: str, url: str, **kw):
|
||||
async def go():
|
||||
async with AsyncClient(transport=self._transport, base_url=self._base) as ac:
|
||||
return await ac.request(method, url, **kw)
|
||||
return self._loop.run_until_complete(go())
|
||||
|
||||
def get(self, url, **kw): return self._do("GET", url, **kw)
|
||||
def post(self, url, **kw): return self._do("POST", url, **kw)
|
||||
def close(self): self._loop.close()
|
||||
|
||||
c = _Client()
|
||||
yield c
|
||||
c.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def remote():
|
||||
from main import app
|
||||
|
||||
class _Client:
|
||||
def __init__(self):
|
||||
self._loop = asyncio.new_event_loop()
|
||||
self._transport = ASGITransport(app=app, client=("1.2.3.4", 12345))
|
||||
self._base = "http://1.2.3.4:8000"
|
||||
|
||||
def _do(self, method: str, url: str, **kw):
|
||||
async def go():
|
||||
async with AsyncClient(transport=self._transport, base_url=self._base) as ac:
|
||||
return await ac.request(method, url, **kw)
|
||||
return self._loop.run_until_complete(go())
|
||||
|
||||
def get(self, url, **kw): return self._do("GET", url, **kw)
|
||||
def post(self, url, **kw): return self._do("POST", url, **kw)
|
||||
def close(self): self._loop.close()
|
||||
|
||||
c = _Client()
|
||||
yield c
|
||||
c.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def stub_env(monkeypatch):
|
||||
"""Isolate connect-info tests from the dev's real backend .env.
|
||||
|
||||
Pydantic ``Settings()`` reads from ``.env`` file directly on
|
||||
instantiation, so monkey-patching ``os.environ`` isn't sufficient
|
||||
— the real ``OPENCLAW_HMAC_SECRET`` would leak through. Instead we
|
||||
override ``get_settings()`` in the route module to return a fresh
|
||||
``Settings`` instance whose env values are driven entirely by an
|
||||
in-test dict, AND we replace ``_write_env_value`` so writes update
|
||||
that same dict instead of touching the developer's filesystem.
|
||||
|
||||
Yields the dict so individual tests can pre-seed values or assert
|
||||
that writes happened.
|
||||
"""
|
||||
import routers.ai_intel as ai_intel
|
||||
import services.config as config
|
||||
|
||||
state: dict[str, str] = {}
|
||||
|
||||
class _FakeSettings:
|
||||
@property
|
||||
def OPENCLAW_HMAC_SECRET(self) -> str:
|
||||
return state.get("OPENCLAW_HMAC_SECRET", "")
|
||||
|
||||
@property
|
||||
def OPENCLAW_ACCESS_TIER(self) -> str:
|
||||
return state.get("OPENCLAW_ACCESS_TIER", "restricted")
|
||||
|
||||
fake = _FakeSettings()
|
||||
|
||||
def _fake_get_settings():
|
||||
return fake
|
||||
|
||||
# Route code calls ``get_settings.cache_clear()`` after writing the
|
||||
# env. The production version is wrapped with ``@lru_cache``, so
|
||||
# cache_clear exists. Attach a no-op shim here.
|
||||
_fake_get_settings.cache_clear = lambda: None # type: ignore[attr-defined]
|
||||
|
||||
monkeypatch.setattr(config, "get_settings", _fake_get_settings)
|
||||
|
||||
def _fake_write_env_value(key: str, value: str) -> None:
|
||||
state[key] = value
|
||||
|
||||
monkeypatch.setattr(ai_intel, "_write_env_value", _fake_write_env_value)
|
||||
|
||||
yield state
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /api/ai/connect-info — always masked, no auto-bootstrap
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetConnectInfoMasking:
|
||||
def test_returns_masked_when_secret_set(self, loopback, stub_env):
|
||||
secret = "abcdef" + "0" * 38 + "wxyz"
|
||||
stub_env["OPENCLAW_HMAC_SECRET"] = secret
|
||||
|
||||
r = loopback.get("/api/ai/connect-info")
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
# Body must NOT carry the full secret value anywhere.
|
||||
assert secret not in r.text, (
|
||||
"GET /api/ai/connect-info MUST NOT include the full HMAC "
|
||||
"secret. Response body contained the secret value."
|
||||
)
|
||||
assert body["hmac_secret_set"] is True
|
||||
assert body["masked_hmac_secret"].startswith("abcdef")
|
||||
assert body["masked_hmac_secret"].endswith("wxyz")
|
||||
assert "•" in body["masked_hmac_secret"]
|
||||
# Pre-fix field is gone.
|
||||
assert "hmac_secret" not in body
|
||||
|
||||
def test_no_auto_bootstrap_when_secret_missing(self, loopback, stub_env):
|
||||
"""Side-effect-on-GET was the second half of issue #302. A GET
|
||||
with no secret configured must NOT mint one — that should
|
||||
require an explicit POST /bootstrap."""
|
||||
r = loopback.get("/api/ai/connect-info")
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["hmac_secret_set"] is False
|
||||
assert body["masked_hmac_secret"] == ""
|
||||
# The bootstrap_behavior block should advertise the new flow.
|
||||
assert body["bootstrap_behavior"]["auto_generates_when_missing"] is False
|
||||
# And no _write_env_value call happened.
|
||||
assert "OPENCLAW_HMAC_SECRET" not in stub_env
|
||||
|
||||
def test_no_reveal_query_param(self, loopback, stub_env):
|
||||
"""Pre-fix, ?reveal=true would return the full secret. Post-fix
|
||||
the param is silently ignored — the response is the same as
|
||||
without it (still masked, no leak)."""
|
||||
secret = "abcdef" + "0" * 38 + "wxyz"
|
||||
stub_env["OPENCLAW_HMAC_SECRET"] = secret
|
||||
|
||||
r = loopback.get("/api/ai/connect-info?reveal=true")
|
||||
assert r.status_code == 200
|
||||
assert secret not in r.text, (
|
||||
"?reveal=true must be a no-op on GET — the full secret "
|
||||
"MUST NOT come back in the response body."
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /api/ai/connect-info/bootstrap
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBootstrap:
|
||||
def test_mints_when_missing(self, loopback, stub_env):
|
||||
r = loopback.post("/api/ai/connect-info/bootstrap")
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["ok"] is True
|
||||
assert body["generated"] is True
|
||||
assert body["hmac_secret_set"] is True
|
||||
# Bootstrap must NOT return the full secret in-line.
|
||||
assert "hmac_secret" not in body or not body.get("hmac_secret")
|
||||
assert "•" in body["masked_hmac_secret"]
|
||||
# _write_env_value was actually called.
|
||||
assert stub_env.get("OPENCLAW_HMAC_SECRET")
|
||||
# The full value isn't echoed back in the response text either.
|
||||
assert stub_env["OPENCLAW_HMAC_SECRET"] not in r.text
|
||||
|
||||
def test_idempotent_when_already_set(self, loopback, stub_env):
|
||||
existing = "abcdef" + "0" * 38 + "wxyz"
|
||||
stub_env["OPENCLAW_HMAC_SECRET"] = existing
|
||||
|
||||
r = loopback.post("/api/ai/connect-info/bootstrap")
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["ok"] is True
|
||||
assert body["generated"] is False
|
||||
assert body["hmac_secret_set"] is True
|
||||
# Existing secret untouched — value is still the seeded one.
|
||||
assert stub_env["OPENCLAW_HMAC_SECRET"] == existing
|
||||
# No full secret in the response.
|
||||
assert existing not in r.text
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /api/ai/connect-info/reveal
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestReveal:
|
||||
def test_returns_full_secret_when_set(self, loopback, stub_env):
|
||||
secret = "abcdef" + "0" * 38 + "wxyz"
|
||||
stub_env["OPENCLAW_HMAC_SECRET"] = secret
|
||||
|
||||
r = loopback.post("/api/ai/connect-info/reveal")
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["ok"] is True
|
||||
assert body["hmac_secret"] == secret
|
||||
|
||||
def test_strict_cache_control_headers(self, loopback, stub_env):
|
||||
"""The whole point of POST /reveal vs GET ?reveal=true is that
|
||||
the response carries headers that prevent every cache layer
|
||||
from persisting the secret."""
|
||||
secret = "abcdef" + "0" * 38 + "wxyz"
|
||||
stub_env["OPENCLAW_HMAC_SECRET"] = secret
|
||||
|
||||
r = loopback.post("/api/ai/connect-info/reveal")
|
||||
cc = r.headers.get("cache-control", "")
|
||||
assert "no-store" in cc, (
|
||||
f"reveal MUST set Cache-Control: no-store — got {cc!r}"
|
||||
)
|
||||
assert "no-cache" in cc
|
||||
# Pragma + Expires as well for HTTP/1.0 caches.
|
||||
assert r.headers.get("pragma", "").lower() == "no-cache"
|
||||
assert r.headers.get("expires") == "0"
|
||||
|
||||
def test_404_when_no_secret_configured(self, loopback, stub_env):
|
||||
r = loopback.post("/api/ai/connect-info/reveal")
|
||||
assert r.status_code == 404
|
||||
# Hint should point at the bootstrap endpoint, not just say "404".
|
||||
detail = r.json().get("detail", "")
|
||||
assert "/bootstrap" in detail or "bootstrap" in detail.lower()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /api/ai/connect-info/regenerate — still returns the new secret
|
||||
# inline (deliberate destructive action), but with no-store headers.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRegenerate:
|
||||
def test_returns_new_secret_with_no_store_headers(self, loopback, stub_env):
|
||||
# Seed an existing secret so we can prove it changes.
|
||||
old = "oldold" + "0" * 38 + "1234"
|
||||
stub_env["OPENCLAW_HMAC_SECRET"] = old
|
||||
|
||||
r = loopback.post("/api/ai/connect-info/regenerate")
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["ok"] is True
|
||||
assert body["hmac_secret"]
|
||||
assert body["hmac_secret"] != old
|
||||
# no-store headers MUST be present so the new secret doesn't
|
||||
# land in browser disk cache after the regenerate click.
|
||||
cc = r.headers.get("cache-control", "")
|
||||
assert "no-store" in cc and "no-cache" in cc
|
||||
assert r.headers.get("pragma", "").lower() == "no-cache"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Auth-gate regression — every endpoint still rejects anonymous remote
|
||||
# callers. This is the property we already enforce for the rest of the
|
||||
# operator-only surface; adding the three new endpoints to the audit
|
||||
# coverage prevents a future refactor from dropping the dependency.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestAnonymousRejection:
|
||||
@pytest.mark.parametrize(
|
||||
"method,path,body",
|
||||
[
|
||||
("get", "/api/ai/connect-info", None),
|
||||
("post", "/api/ai/connect-info/bootstrap", None),
|
||||
("post", "/api/ai/connect-info/reveal", None),
|
||||
("post", "/api/ai/connect-info/regenerate", None),
|
||||
],
|
||||
)
|
||||
def test_remote_rejected(self, remote, method, path, body):
|
||||
fn = getattr(remote, method)
|
||||
r = fn(path, json=body) if body is not None else fn(path)
|
||||
assert r.status_code == 403, (
|
||||
f"{method.upper()} {path} must reject anonymous remote callers; "
|
||||
f"got {r.status_code}"
|
||||
)
|
||||
Reference in New Issue
Block a user