grokbot_ty
2f9b8ebd02
Full-screen WooSetupGate on App mount when this browser lacks device WP user + app password. Save + Test Woo dismisses only after a successful test. Publish/test-woo auth failures reopen the gate. Server: client_for_request requires X-Woo-* on publish, taxonomy sync, and test-woo (401 woo_device_creds_required); health probe keeps Settings. Optional SLAB_WOO_ALLOW_SETTINGS_FALLBACK for legacy tests. See PROTOTYPE.md.
260 lines
9.9 KiB
Python
260 lines
9.9 KiB
Python
"""WooCommerce REST v3 client (stdlib urllib). Basic-auth, HTTPS-only, paginated.
|
|
|
|
Shop-default creds live in Settings (wp_username plain, app_password AES-GCM) and
|
|
are used by the health probe only. HTTP Woo routes (publish / taxonomy sync /
|
|
test-woo) require X-Woo-Username + X-Woo-App-Password on the request — no
|
|
Settings fallback. Missing headers raise WooDeviceCredsRequired. See PROTOTYPE.md.
|
|
|
|
Optional escape hatch for legacy tests: SLAB_WOO_ALLOW_SETTINGS_FALLBACK=1.
|
|
"""
|
|
|
|
import base64
|
|
import json
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any, Callable
|
|
|
|
from .crypto import DecryptError, decrypt
|
|
|
|
# WordPress security plugins (Wordfence, WAF, CDN) commonly block the default
|
|
# Python-urllib User-Agent with 403. Send a descriptive one.
|
|
_USER_AGENT = "SlabUploader/1.0 (+https://www.whidbeywoodstore.com)"
|
|
|
|
|
|
class WooError(Exception):
|
|
"""Base for all Woo client failures."""
|
|
|
|
|
|
class WooCredentialsMissing(WooError):
|
|
"""Woo base URL or credentials are not configured (fail-closed)."""
|
|
|
|
|
|
class WooDeviceCredsRequired(WooCredentialsMissing):
|
|
"""X-Woo-Username + X-Woo-App-Password required; Settings fallback disabled."""
|
|
|
|
|
|
class WooAuthError(WooError):
|
|
"""The store rejected the credentials (401)."""
|
|
|
|
|
|
class WooForbiddenError(WooError):
|
|
"""The store blocked the request (403) — usually a security plugin or WAF."""
|
|
|
|
|
|
class WooNotFoundError(WooError):
|
|
pass
|
|
|
|
|
|
class WooServerError(WooError):
|
|
pass
|
|
|
|
|
|
class WooConnectionError(WooError):
|
|
"""Store unreachable (DNS, TLS, timeout)."""
|
|
|
|
|
|
def _default_request(url: str, method: str, data: bytes | None, headers: dict[str, str]):
|
|
req = urllib.request.Request(url, method=method, data=data)
|
|
for k, v in headers.items():
|
|
req.add_header(k, v)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=15) as r:
|
|
return r.getcode(), dict(r.headers.items()), r.read()
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read()
|
|
try:
|
|
parsed = json.loads(body)
|
|
except (ValueError, json.JSONDecodeError):
|
|
parsed = {}
|
|
msg = parsed.get("message") if isinstance(parsed, dict) else None
|
|
if e.code == 401:
|
|
raise WooAuthError(msg or "authentication rejected (401)") from e
|
|
if e.code == 403:
|
|
raise WooForbiddenError(msg or "request forbidden (403)") from e
|
|
if e.code == 404:
|
|
raise WooNotFoundError(msg or "not found (404)") from e
|
|
raise WooServerError(f"server error {e.code}: {msg}") from e
|
|
except (urllib.error.URLError, OSError, TimeoutError) as e:
|
|
raise WooConnectionError(f"store unreachable: {e}") from e
|
|
|
|
|
|
class WooClient:
|
|
def __init__(self, username: str, password: str, base_url: str,
|
|
request: Callable[..., tuple[int, dict, bytes]] = _default_request) -> None:
|
|
if not base_url or not base_url.startswith("https://"):
|
|
raise WooCredentialsMissing("WOO_BASE_URL must be https:// (SETTINGS-10)")
|
|
self.base_url = base_url.rstrip("/")
|
|
self._request = request
|
|
self._auth = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii")
|
|
|
|
@classmethod
|
|
def from_settings(cls, base_url: str, username: str | None, encrypted_password: str | None,
|
|
request: Callable[..., tuple[int, dict, bytes]] | None = None) -> "WooClient":
|
|
"""Build from Settings values; raise WooCredentialsMissing if incomplete (fail-closed)."""
|
|
if not username:
|
|
raise WooCredentialsMissing("woo_wp_username not set (SETTINGS-15)")
|
|
if not encrypted_password:
|
|
raise WooCredentialsMissing("woo_app_password not configured (SETTINGS-15)")
|
|
try:
|
|
password = decrypt(encrypted_password)
|
|
except DecryptError as e:
|
|
raise WooCredentialsMissing("woo_app_password cannot be decrypted (re-enter it)") from e
|
|
if request is None:
|
|
request = _default_request
|
|
return cls(username, password, base_url, request)
|
|
|
|
def _get(self, path: str) -> tuple[int, dict, bytes]:
|
|
url = f"{self.base_url}/wp-json/wc/v3{path}"
|
|
return self._request(url, "GET", None, self._auth_headers())
|
|
|
|
def _fetch_all(self, path: str) -> list[dict[str, Any]]:
|
|
page = 1
|
|
results: list[dict[str, Any]] = []
|
|
while True:
|
|
code, headers, body = self._get(f"{path}?per_page=100&page={page}")
|
|
if code != 200:
|
|
raise WooServerError(f"expected 200, got {code}")
|
|
results.extend(json.loads(body))
|
|
total_pages = int(headers.get("X-WP-TotalPages") or headers.get("x-wp-totalpages") or 1)
|
|
if page >= total_pages:
|
|
return results
|
|
page += 1
|
|
|
|
def get_categories(self) -> list[dict[str, Any]]:
|
|
return self._fetch_all("/products/categories")
|
|
|
|
def get_tags(self) -> list[dict[str, Any]]:
|
|
return self._fetch_all("/products/tags")
|
|
|
|
def get_attributes(self) -> list[dict[str, Any]]:
|
|
return self._fetch_all("/products/attributes")
|
|
|
|
def get_attribute_terms(self, attr_id: int) -> list[dict[str, Any]]:
|
|
return self._fetch_all(f"/products/attributes/{attr_id}/terms")
|
|
|
|
def _auth_headers(self, extra: dict[str, str] | None = None) -> dict[str, str]:
|
|
"""Shared headers for every outbound Woo/WP request (User-Agent + auth)."""
|
|
h = {
|
|
"Authorization": f"Basic {self._auth}",
|
|
"Accept": "application/json",
|
|
"User-Agent": _USER_AGENT,
|
|
}
|
|
if extra:
|
|
h.update(extra)
|
|
return h
|
|
|
|
def get_products_by_sku(self, sku: str) -> list[dict[str, Any]]:
|
|
"""GET products filtered by SKU. Raises typed Woo* errors on non-200."""
|
|
code, _h, body = self._request(
|
|
f"{self.base_url}/wp-json/wc/v3/products?sku={sku}",
|
|
"GET",
|
|
None,
|
|
self._auth_headers(),
|
|
)
|
|
if code != 200:
|
|
raise WooServerError(f"sku lookup returned {code}")
|
|
return list(json.loads(body))
|
|
|
|
def create_product(self, payload: dict[str, Any]) -> int:
|
|
"""POST /products. Raises typed Woo* errors on failure; returns product id."""
|
|
body = json.dumps(payload).encode("utf-8")
|
|
code, _h, resp = self._request(
|
|
f"{self.base_url}/wp-json/wc/v3/products",
|
|
"POST",
|
|
body,
|
|
self._auth_headers({"Content-Type": "application/json"}),
|
|
)
|
|
if code not in (200, 201):
|
|
raise WooServerError(f"product create returned {code}")
|
|
return int(json.loads(resp)["id"])
|
|
|
|
def upload_media(self, png_bytes: bytes, filename: str) -> int:
|
|
"""POST a PNG to WordPress core media; return the attachment id."""
|
|
boundary = "----slabuploader-media-boundary"
|
|
body = (
|
|
f"--{boundary}\r\n"
|
|
f'Content-Disposition: form-data; name="file"; filename="{filename}"\r\n'
|
|
"Content-Type: image/png\r\n\r\n"
|
|
).encode("utf-8") + png_bytes + f"\r\n--{boundary}--\r\n".encode("utf-8")
|
|
url = f"{self.base_url}/wp-json/wp/v2/media"
|
|
code, _headers, resp = self._request(
|
|
url, "POST", body,
|
|
self._auth_headers({"Content-Type": f"multipart/form-data; boundary={boundary}"}),
|
|
)
|
|
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
|
|
# HTTP Woo routes REQUIRE both headers (no Settings fallback). Health probe
|
|
# still uses shop-default Settings via WooClient.from_settings / settings_woo_creds.
|
|
# Never write override headers into the Settings row.
|
|
|
|
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 _allow_settings_fallback() -> bool:
|
|
"""Legacy-test escape hatch. Production HTTP Woo routes leave this unset."""
|
|
import os
|
|
|
|
return os.environ.get("SLAB_WOO_ALLOW_SETTINGS_FALLBACK", "").strip().lower() in (
|
|
"1",
|
|
"true",
|
|
"yes",
|
|
)
|
|
|
|
|
|
def client_for_request(
|
|
conn,
|
|
headers=None,
|
|
request: Callable[..., tuple[int, dict, bytes]] | None = None,
|
|
*,
|
|
require_override: bool = True,
|
|
) -> WooClient:
|
|
"""Build a WooClient for one HTTP Woo request.
|
|
|
|
Requires X-Woo-* headers by default (no Settings fallback). Set
|
|
require_override=False or SLAB_WOO_ALLOW_SETTINGS_FALLBACK=1 only for
|
|
health-adjacent / legacy-test paths. Never mutates Settings.
|
|
"""
|
|
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)
|
|
if require_override and not _allow_settings_fallback():
|
|
raise WooDeviceCredsRequired(
|
|
"X-Woo-Username and X-Woo-App-Password are required on this device"
|
|
)
|
|
return WooClient.from_settings(base, settings_user, settings_enc, request=request)
|