proto(woo): per-browser Woo creds via X-Woo-* request headers

Prototype routing so 2–3 owners can share one store URL with different WP
application passwords. Device creds live in localStorage; server uses
request override → Settings fallback and never writes overrides into the
Settings row. Inference stays on shared Settings.

See PROTOTYPE.md for UAT with two browsers and known LAN forge risk.
This commit is contained in:
grokbot_ty
2026-09-26 12:26:21 -07:00
parent fc40c66c44
commit 2f6335ec26
8 changed files with 419 additions and 31 deletions
+12 -11
View File
@@ -213,22 +213,23 @@ from .woo import (
WooCredentialsMissing,
WooError,
WooForbiddenError,
client_for_request,
settings_woo_creds,
)
def _woo_credentials() -> tuple[str, str | None, str | None]:
conn = get_connection()
base = os.environ.get("WOO_BASE_URL", "")
user_row = conn.execute("SELECT value FROM settings WHERE key='woo_wp_username'").fetchone()
pass_row = conn.execute("SELECT value_enc FROM settings WHERE key='woo_app_password'").fetchone()
return base, (user_row[0] if user_row else None), (pass_row[0] if pass_row else None)
"""Shop-default Woo creds (Settings). Used by health probe only."""
return settings_woo_creds(get_connection())
@app.post("/api/v1/settings/test-woo")
async def test_woo() -> dict[str, Any]:
base, username, enc_password = _woo_credentials()
async def test_woo(request: Request) -> dict[str, Any]:
"""Probe Woo + sync taxonomy. Optional X-Woo-Username / X-Woo-App-Password
override shop-default Settings for this request only (never persisted).
"""
try:
client = WooClient.from_settings(base, username, enc_password)
client = client_for_request(get_connection(), request.headers)
except WooCredentialsMissing as e:
return {
"ok": False,
@@ -260,10 +261,10 @@ async def get_taxonomy() -> dict[str, Any]:
@app.post("/api/v1/admin/taxonomy/sync")
async def post_taxonomy_sync() -> dict[str, Any]:
base, username, enc_password = _woo_credentials()
async def post_taxonomy_sync(request: Request) -> dict[str, Any]:
"""Refresh taxonomy cache. Honors X-Woo-* override headers for this request."""
try:
client = WooClient.from_settings(base, username, enc_password)
client = client_for_request(get_connection(), request.headers)
except WooCredentialsMissing as e:
return JSONResponse(status_code=502, content=_error("woo_unavailable", str(e)))
try:
+8 -10
View File
@@ -12,7 +12,7 @@ import re
from datetime import datetime, timedelta, timezone
from typing import Any
from fastapi import APIRouter
from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse
from .db import get_connection
@@ -23,6 +23,8 @@ from .woo import (
WooCredentialsMissing,
WooError,
WooForbiddenError,
client_for_request,
settings_woo_creds,
)
from .review import compute_ready, _slab_dict
@@ -139,7 +141,7 @@ def _classify(e: Exception) -> tuple[str, str, bool]:
@router.post("/{slab_id}/publish")
async def publish(slab_id: str) -> dict:
async def publish(slab_id: str, request: Request) -> dict:
conn = get_connection()
row = conn.execute("SELECT * FROM slabs WHERE id=? AND deleted_at IS NULL", (slab_id,)).fetchone()
if row is None:
@@ -150,9 +152,9 @@ async def publish(slab_id: str) -> dict:
if not ready:
return JSONResponse(status_code=422, content=_error("unpublishable", "Slab is not ready", {"missing": missing}))
# 2. SKU lookup (PUBLISH-15) — stop if it exists in the store.
base, user, enc = _woo_creds(conn)
# Per-request Woo creds: X-Woo-* headers → Settings fallback (never persisted).
try:
client = WooClient.from_settings(base, user, enc)
client = client_for_request(conn, request.headers)
except WooCredentialsMissing as e:
return JSONResponse(status_code=422, content=_error("woo_unconfigured", str(e)))
try:
@@ -221,12 +223,8 @@ async def publish(slab_id: str) -> dict:
def _woo_creds(conn) -> tuple[str, str | None, str | None]:
import os
base = os.environ.get("WOO_BASE_URL", "")
u = conn.execute("SELECT value FROM settings WHERE key='woo_wp_username'").fetchone()
p = conn.execute("SELECT value_enc FROM settings WHERE key='woo_app_password'").fetchone()
return base, (u[0] if u else None), (p[0] if p else None)
"""Shop-default Woo creds (Settings). Prefer client_for_request for HTTP paths."""
return settings_woo_creds(conn)
def _create_status(conn) -> str:
+113
View File
@@ -0,0 +1,113 @@
"""Prototype: per-request X-Woo-* headers build distinct WooClients without clobbering Settings."""
import base64
import os
import shutil
import tempfile
import pytest
from starlette.datastructures import Headers
import backend.db as db_mod
from backend.crypto import encrypt, decrypt
from backend.woo import (
WooClient,
client_for_request,
request_woo_override,
settings_woo_creds,
HEADER_WOO_USERNAME,
HEADER_WOO_APP_PASSWORD,
)
@pytest.fixture
def db_dir():
tmp = tempfile.mkdtemp(prefix="per_user_woo_")
os.environ["SLAB_DB_PATH"] = os.path.join(tmp, "t.db")
os.environ["SLAB_IMAGES_DIR"] = tmp
os.environ["WOO_BASE_URL"] = "https://store.example.com"
os.environ["SLAB_AES_KEY"] = base64.b64encode(b"0123456789abcdef0123456789abcdef").decode()
db_mod._reset_for_tests()
conn = db_mod.init_db()
conn.execute(
"INSERT OR REPLACE INTO settings(key, value, value_enc) VALUES ('woo_wp_username','shop-default',NULL)"
)
conn.execute(
"INSERT OR REPLACE INTO settings(key, value, value_enc) VALUES ('woo_app_password',NULL,?)",
(encrypt("shop-secret"),),
)
conn.commit()
yield conn
db_mod._reset_for_tests()
shutil.rmtree(tmp, ignore_errors=True)
def _auth_pair(client: WooClient) -> tuple[str, str]:
raw = base64.b64decode(client._auth).decode("utf-8")
user, password = raw.split(":", 1)
return user, password
def test_override_headers_produce_distinct_auth_without_clobbering_settings(db_dir):
conn = db_dir
before = settings_woo_creds(conn)
assert before[1] == "shop-default"
assert decrypt(before[2]) == "shop-secret"
h_a = Headers({HEADER_WOO_USERNAME: "alice", HEADER_WOO_APP_PASSWORD: "alice-app-pw"})
h_b = Headers({HEADER_WOO_USERNAME: "bob", HEADER_WOO_APP_PASSWORD: "bob-app-pw"})
client_a = client_for_request(conn, h_a)
client_b = client_for_request(conn, h_b)
client_fallback = client_for_request(conn, Headers({}))
assert _auth_pair(client_a) == ("alice", "alice-app-pw")
assert _auth_pair(client_b) == ("bob", "bob-app-pw")
assert _auth_pair(client_fallback) == ("shop-default", "shop-secret")
assert client_a._auth != client_b._auth
after = settings_woo_creds(conn)
assert after == before
assert decrypt(after[2]) == "shop-secret"
# Explicit: Settings row values unchanged
row_u = conn.execute("SELECT value FROM settings WHERE key='woo_wp_username'").fetchone()[0]
row_p = conn.execute("SELECT value_enc FROM settings WHERE key='woo_app_password'").fetchone()[0]
assert row_u == "shop-default"
assert decrypt(row_p) == "shop-secret"
def test_partial_headers_fall_back_to_settings(db_dir):
conn = db_dir
# Username only — incomplete override
assert request_woo_override(Headers({HEADER_WOO_USERNAME: "alice"})) is None
client = client_for_request(conn, Headers({HEADER_WOO_USERNAME: "alice"}))
assert _auth_pair(client) == ("shop-default", "shop-secret")
def test_test_woo_endpoint_uses_request_override(db_dir, monkeypatch):
"""API: POST /settings/test-woo with X-Woo-* builds client for that user."""
from fastapi.testclient import TestClient
from backend import app as app_mod
seen: list[str] = []
def fake_sync(client, conn):
seen.append(_auth_pair(client)[0])
return {"taxonomy_anchor": "x", "taxonomy_last_sync_at": "2026-01-01T00:00:00"}
monkeypatch.setattr(app_mod.taxonomy_mod, "sync", fake_sync)
tc = TestClient(app_mod.app)
r1 = tc.post(
"/api/v1/settings/test-woo",
headers={HEADER_WOO_USERNAME: "alice", HEADER_WOO_APP_PASSWORD: "pw-a"},
)
r2 = tc.post(
"/api/v1/settings/test-woo",
headers={HEADER_WOO_USERNAME: "bob", HEADER_WOO_APP_PASSWORD: "pw-b"},
)
r3 = tc.post("/api/v1/settings/test-woo")
assert r1.status_code == 200 and r1.json()["ok"] is True
assert r2.status_code == 200 and r2.json()["ok"] is True
assert r3.status_code == 200 and r3.json()["ok"] is True
assert seen == ["alice", "bob", "shop-default"]
assert settings_woo_creds(db_dir)[1] == "shop-default"
+57 -3
View File
@@ -1,8 +1,12 @@
"""WooCommerce REST v3 client (stdlib urllib). Basic-auth, HTTPS-only, paginated.
Creds come from Settings (wp_username plain, app_password AES-GCM). Missing creds
raise WooCredentialsMissing (a fail-closed condition the caller maps to ok:false),
distinct from WooAuthError (a 401 from the store).
Default creds come from Settings (wp_username plain, app_password AES-GCM).
Missing creds raise WooCredentialsMissing (fail-closed), distinct from
WooAuthError (401 from the store).
Prototype per-user override: callers may pass X-Woo-Username + X-Woo-App-Password
request headers into client_for_request(); those build a per-request WooClient
and are never written into Settings. See PROTOTYPE.md.
"""
import base64
@@ -177,3 +181,53 @@ class WooClient:
if code not in (200, 201):
raise WooServerError(f"media upload returned {code}")
return int(json.loads(resp)["id"])
# --- Per-request Woo credential override (prototype) ---
# Scheme: browser sends plaintext WP Application Password credentials on
# Woo-touching API calls via:
# X-Woo-Username
# X-Woo-App-Password
# Server uses those for THIS request's WooClient only; never writes them into
# the Settings row. If either header is missing/blank, fall back to the
# shop-default singleton (woo_wp_username + encrypted woo_app_password).
HEADER_WOO_USERNAME = "X-Woo-Username"
HEADER_WOO_APP_PASSWORD = "X-Woo-App-Password"
def settings_woo_creds(conn) -> tuple[str, str | None, str | None]:
"""Shop-default Woo creds from env + Settings (encrypted password)."""
import os
base = os.environ.get("WOO_BASE_URL", "")
u = conn.execute("SELECT value FROM settings WHERE key='woo_wp_username'").fetchone()
p = conn.execute("SELECT value_enc FROM settings WHERE key='woo_app_password'").fetchone()
return base, (u[0] if u else None), (p[0] if p else None)
def request_woo_override(headers) -> tuple[str, str] | None:
"""Return (username, app_password) if both override headers are present and non-blank."""
if headers is None:
return None
# Starlette Headers are case-insensitive
user = (headers.get(HEADER_WOO_USERNAME) or "").strip()
password = (headers.get(HEADER_WOO_APP_PASSWORD) or "").strip()
if user and password:
return user, password
return None
def client_for_request(conn, headers=None, request: Callable[..., tuple[int, dict, bytes]] | None = None) -> WooClient:
"""Build a WooClient for one HTTP request: override headers → Settings fallback.
Never mutates Settings. Inference stays on shared Settings (caller responsibility).
"""
base, settings_user, settings_enc = settings_woo_creds(conn)
override = request_woo_override(headers)
if override is not None:
user, password = override
if request is None:
request = _default_request
return WooClient(user, password, base, request)
return WooClient.from_settings(base, settings_user, settings_enc, request=request)