hanzo/studio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(studio/docs): decomplect documents AI-fill onto the ONE canonical gateway_auth

Browse Source 
documents.py carried a parity copy of the gateway-auth allowlist (_jwt_alg,
_forwardable_bearer, _SECRET_PREFIXES) — and it DID drift: the same crafted-alg
hardening + exact-JWS tightening had to be applied to both copies. _ai_token now
delegates to gateway_auth.ai_bearer(request, server_env=_SERVER_TOKEN_ENV); the
~55 duplicated lines are deleted. One and only one fail-closed allowlist, so a
future fix can never land on one copy and miss the other. Tests repointed to the
_ai_token delegation surface (crafted-alg no-crash, HS256/secret refused, pk-/
asymmetric forwarded, server-key-first). 67 pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
z
2026-07-25 09:36:21 -07:00
co-authored by Claude Opus 4.8
parent 20d2f809af
commit d8399c312f
2 changed files with 31 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
middleware/documents.py
+7 -52
View File
@@ -756,64 +756,19 @@ def render_document(tmpl, fields, fmt):
# ── AI-fill: description -> structured fields via the Hanzo gateway ─────────────────
_SECRET_PREFIXES = ("sk-", "sk_", "hk-", "hk_", "rk-", "rk_")
_SERVER_TOKEN_ENV = ("STUDIO_DOC_TOKEN", "STUDIO_COPILOT_TOKEN", "STUDIO_COMMERCE_TOKEN")
def _jwt_alg(tok):
"""The ``alg`` from a JWT header, or "" if not a decodable JWT. Used to distinguish a
JWKS-verifiable asymmetric token (gateway-acceptable) from an HS256 session token."""
import base64
try:
h = tok.split(".", 1)[0]
h += "=" * (-len(h) % 4)
alg = json.loads(base64.urlsafe_b64decode(h.encode())).get("alg", "")
# A crafted header can make alg a non-string (null/list/number/dict); the
# contract is "the alg string or ''", so a non-string is not a usable alg.
return alg if isinstance(alg, str) else ""
except Exception:
return ""
def _forwardable_bearer(request):
"""The caller's own token to forward to the gateway, or "" — fail-closed.
Reuses the app's one token path (``gpu_dispatch._bearer``: Authorization header or
the browser session cookie), then applies a strict allowlist:
* a publishable ``pk-`` key — allowed;
* a JWT — allowed ONLY when asymmetric/JWKS-verifiable (RS/ES/PS/EdDSA); a symmetric
HS256 hanzo.id session token is refused (the gateway rejects it, so forwarding it
only produces log noise and a 502);
* a secret ``sk-``/``hk-``/``rk-`` key — refused outright.
So a secret can never reach the gateway from the browser edge and none is in page source."""
from middleware.gpu_dispatch import _bearer # lazy: keeps this module import light
raw = _bearer(request)
if not raw:
return ""
tok = raw[7:].strip() if raw[:7].lower() == "bearer " else raw.strip()
if not tok or tok.lower().startswith(_SECRET_PREFIXES):
return ""
if tok.startswith(("pk-", "pk_")):
return "Bearer " + tok
if tok.count(".") == 2:
# Exact JWS asymmetric set (a prefix match also admitted RSA-OAEP / "Edwina"…);
# `alg in <tuple>` is crash-safe for a non-string alg too.
if _jwt_alg(tok) in ("RS256", "RS384", "RS512", "ES256", "ES384", "ES512",
"PS256", "PS384", "PS512", "EdDSA"):
return "Bearer " + tok
return ""
def _ai_token(request):
"""The bearer AI-fill sends to the gateway, "" when none is usable (→ form fallback).
A server-side gateway key wins (the configured, working path; never in page source);
otherwise the caller's own gateway-verifiable token via the fail-closed allowlist."""
for env in _SERVER_TOKEN_ENV:
v = os.environ.get(env, "").strip()
if v:
return v if v[:7].lower() == "bearer " else "Bearer " + v
return _forwardable_bearer(request)
Delegates to the ONE canonical fail-closed allowlist (``middleware.gateway_auth``) —
NO private copy (they used to drift: the same crafted-alg hardening had to be applied
to both). A server-side gateway key wins (never in page source); else the caller's own
token ONLY when the gateway can verify it (a ``pk-`` key or an asymmetric/JWKS JWT) —
an HS256 session token and ``sk-``/``hk-``/``rk-`` secrets are refused."""
from middleware import gateway_auth # lazy: keeps this module import light
return gateway_auth.ai_bearer(request, server_env=_SERVER_TOKEN_ENV)
def _gateway_url():
tests-unit/documents_test/test_documents.py
+24 -20
View File
@@ -193,16 +193,16 @@ def test_forwardable_bearer_is_fail_closed(monkeypatch):
for env in D._SERVER_TOKEN_ENV:
monkeypatch.delenv(env, raising=False)
rs, hs = _jwt("RS256"), _jwt("HS256")
assert D._forwardable_bearer(_Req("Bearer " + rs)) == "Bearer " + rs # JWKS-verifiable JWT
assert D._forwardable_bearer(_Req("Bearer " + hs)) == "" # HS256 session token REFUSED
assert D._forwardable_bearer(_Req("Bearer pk-publishable")) == "Bearer pk-publishable"
assert D._forwardable_bearer(_Req("Bearer sk-secret")) == "" # secret keys refused
assert D._forwardable_bearer(_Req("Bearer hk-secret")) == ""
assert D._forwardable_bearer(_Req("Bearer rk-secret")) == ""
assert D._forwardable_bearer(_Req(None)) == "" # no token
assert D._forwardable_bearer(_Req("Bearer notoken")) == "" # opaque, not JWT/pk-
assert D._ai_token(_Req("Bearer " + rs)) == "Bearer " + rs # JWKS-verifiable JWT
assert D._ai_token(_Req("Bearer " + hs)) == "" # HS256 session token REFUSED
assert D._ai_token(_Req("Bearer pk-publishable")) == "Bearer pk-publishable"
assert D._ai_token(_Req("Bearer sk-secret")) == "" # secret keys refused
assert D._ai_token(_Req("Bearer hk-secret")) == ""
assert D._ai_token(_Req("Bearer rk-secret")) == ""
assert D._ai_token(_Req(None)) == "" # no token
assert D._ai_token(_Req("Bearer notoken")) == "" # opaque, not JWT/pk-
# session-cookie path (what the browser sends) is honored for a verifiable JWT
assert D._forwardable_bearer(_Req(None, {"access_token": rs})) == "Bearer " + rs
assert D._ai_token(_Req(None, {"access_token": rs})) == "Bearer " + rs
def test_ai_token_prefers_server_gateway_key(monkeypatch):
@@ -477,20 +477,24 @@ def test_control_chars_stripped_and_ooxml_stays_valid():
assert _valid("xlsx", D.render_document(b, bf, "xlsx")[0])
def test_crafted_jwt_alg_never_crashes_the_allowlist():
import base64
def test_ai_token_delegates_to_canonical_gateway_auth(monkeypatch):
"""After the decomplect, documents._ai_token routes through the ONE gateway_auth
allowlist (no private copy — they used to drift). Prove the fail-closed behavior via
the public _ai_token: a crafted non-string alg must not 500; HS256/secret refused;
pk-/asymmetric forwarded; a server key wins."""
for env in D._SERVER_TOKEN_ENV:
monkeypatch.delenv(env, raising=False)
def jwt_with_alg(alg_value): # a JWT header carrying an arbitrary (maybe non-string) alg
import base64
h = base64.urlsafe_b64encode(json.dumps({"alg": alg_value}).encode()).rstrip(b"=").decode()
return f"{h}.eyJzdWIiOiIxIn0.sig"
for bad in (None, [], 123, {"x": 1}, True):
assert D._jwt_alg(jwt_with_alg(bad)) == "", "a non-string alg must yield '' (no exception)"
assert D._jwt_alg(jwt_with_alg("RS256")) == "RS256" # real asymmetric alg still read
from aiohttp.test_utils import make_mocked_request
req = make_mocked_request("POST", "/v1/documents/generate",
headers={"Authorization": "Bearer " + jwt_with_alg(None)})
assert D._forwardable_bearer(req) == "" # crafted alg → not forwarded, no 500
ok = make_mocked_request("POST", "/v1/documents/generate",
headers={"Authorization": "Bearer " + jwt_with_alg("RS256")})
assert D._forwardable_bearer(ok) == "Bearer " + jwt_with_alg("RS256") # asymmetric still forwarded
assert D._ai_token(_Req("Bearer " + jwt_with_alg(bad))) == "" # crafted alg → no 500
assert D._ai_token(_Req("Bearer " + jwt_with_alg("HS256"))) == "" # HS256 session token refused
assert D._ai_token(_Req("Bearer sk-secret")) == "" # secret refused
assert D._ai_token(_Req("Bearer pk-x")) == "Bearer pk-x" # publishable forwarded
assert D._ai_token(_Req("Bearer " + jwt_with_alg("RS256"))) == "Bearer " + jwt_with_alg("RS256")
monkeypatch.setenv("STUDIO_DOC_TOKEN", "sk-server") # server key wins
assert D._ai_token(_Req("Bearer pk-x")) == "Bearer sk-server"