"""Shopify OAuth endpoints. Mounted at /api/shopify to match the redirect URL registered in the Shopify Dev Dashboard (app version galaxiq-5). Changing this prefix breaks the callback. """ import json import logging import secrets from fastapi import APIRouter, HTTPException, Query from fastapi.responses import RedirectResponse from app.core.config import settings from app.services.integrations.platform import ( TenantHasNoUserError, track_connect, upsert_integration, ) from app.services.catalog.sources import ( KIND_SHOPIFY, encrypt_credentials, get_sources, upsert_source, ) from app.services.infra.redis import get_redis_client from app.services.integrations.shopify_oauth import ( SHOPIFY_API_VERSION, ShopQueryError, TokenExchangeError, build_authorize_url, exchange_code, fetch_shop_info, normalise_shop_domain, verify_hmac, ) logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/shopify", tags=["shopify"]) STATE_PREFIX = "shopify:oauth:" STATE_TTL_SECONDS = 600 @router.get("/install") async def install( shop: str = Query(..., description=".myshopify.com"), tenant_id: str = Query(..., description="GalaxiQ tenant to bind the shop to"), ): """Start the authorization code grant. TODO(auth): tenant_id arrives from the query string because this codebase has no session. That lets anyone bind a shop to any tenant. It must come from an authenticated session before a single external merchant connects. """ domain = normalise_shop_domain(shop) if not domain: # Rejecting here is what prevents this route becoming an open redirect. logger.warning("Rejected install for invalid shop domain: %r", shop) raise HTTPException(status_code=400, detail="Invalid shop domain") state = secrets.token_hex(24) payload = json.dumps({"shop": domain, "tenant_id": tenant_id}) client = get_redis_client() async with client: await client.setex(f"{STATE_PREFIX}{state}", STATE_TTL_SECONDS, payload) logger.info("Starting Shopify install for %s (tenant %s)", domain, tenant_id) _safe_track(tenant_id, "oauth_authorize_url_ready", "success") return RedirectResponse(build_authorize_url(domain, state)) def _frontend_url(**params: str) -> str: """FRONTEND_URL with the given query params appended. Shared by the error and success paths so the "does FRONTEND_URL already have a query string" check lives in exactly one place. """ sep = "&" if "?" in settings.FRONTEND_URL else "?" query = "&".join(f"{k}={v}" for k, v in params.items()) return f"{settings.FRONTEND_URL}{sep}{query}" def _error_redirect(reason: str, status_code: int = 303) -> RedirectResponse: """Send the merchant back to the UI with a readable reason. Rendering an error page instead would leave the flow untestable from the browser, which is where it actually runs. """ # 303, not the failure's own status (400/403/502): browsers only follow # a Location header automatically on a 3xx response. The one exception is # no_owning_user (409): the platform genuinely could not record the # connection, and a merchant-visible redirect must not read like success. return RedirectResponse( _frontend_url(shopify_error=reason), status_code=status_code) def _safe_track(tenant_id: str, delivery_status: str, outcome: str, error_message: str = None) -> None: """The audit trail must never be the reason a connection fails or a merchant-facing error redirect doesn't happen.""" try: track_connect(tenant_id, "shopify", delivery_status, outcome, error_message) except Exception: logger.warning("Connect tracking failed for %s", tenant_id, exc_info=True) @router.get("/callback") async def callback( shop: str = Query(None), code: str = Query(None), state: str = Query(None), hmac: str = Query(None), timestamp: str = Query(None), host: str = Query(None), ): # 1. Shop domain. Never trust it twice. domain = normalise_shop_domain(shop) if not domain: logger.warning("Callback with invalid shop domain: %r", shop) return _error_redirect("invalid_shop") # 2. State nonce: read and delete before anything else, so a failed # attempt still burns the nonce and a forged signature cannot be # retried against a live state value. client = get_redis_client() async with client: raw = await client.get(f"{STATE_PREFIX}{state}") if state else None if raw: await client.delete(f"{STATE_PREFIX}{state}") if not raw: logger.warning("Callback with unknown or expired state for %s", domain) return _error_redirect("invalid_state") stored = json.loads(raw) if stored.get("shop") != domain: logger.warning("Callback shop mismatch: state had %s, callback had %s", stored.get("shop"), domain) _safe_track(stored.get("tenant_id"), "not_connected", "failed", "shop_mismatch") return _error_redirect("shop_mismatch") # 3. HMAC over every query parameter Shopify sent. params = {k: v for k, v in { "shop": shop, "code": code, "state": state, "hmac": hmac, "timestamp": timestamp, "host": host, }.items() if v is not None} if not verify_hmac(params): logger.warning("Callback HMAC verification failed for %s", domain) _safe_track(stored["tenant_id"], "not_connected", "failed", "hmac_failed") return _error_redirect("hmac_failed") # 4. Exchange the code for an offline token. try: token_data = await exchange_code(domain, code) except TokenExchangeError: _safe_track(stored["tenant_id"], "not_connected", "failed", "token_exchange_failed") return _error_redirect("token_exchange_failed") # 5. Prove the token works before storing it, and collect what the catalog # sync needs. An unverified token fails later, during a sync nobody is # watching, instead of now. try: info = await fetch_shop_info(domain, token_data["access_token"]) except ShopQueryError: _safe_track(stored["tenant_id"], "not_connected", "failed", "token_verification_failed") return _error_redirect("token_verification_failed") upsert_source( tenant_id=stored["tenant_id"], kind=KIND_SHOPIFY, external_ref=domain, config={ "shop_domain": domain, "shop_name": info["name"], "scopes": token_data["scope"], "api_version": SHOPIFY_API_VERSION, "currency_code": info["currency_code"], "primary_domain": info["primary_domain"], }, credentials={"access_token": token_data["access_token"]}, ) # Record the connection on the platform's shared integration tables so # Shopify shows up alongside the other providers. A bookkeeping failure # here must not undo a connection that already succeeded (upsert_source # above already committed), except when the platform truly cannot record # it at all -- see TenantHasNoUserError below. try: upsert_integration( tenant_id=stored["tenant_id"], provider="shopify", auth_type="oauth2", access_token_ciphertext=encrypt_credentials( {"access_token": token_data["access_token"]}), metadata={ "shop_domain": domain, "shop_name": info["name"], "scopes": token_data["scope"], "api_version": SHOPIFY_API_VERSION, "currency_code": info["currency_code"], "primary_domain": info["primary_domain"], }, ) except TenantHasNoUserError: # integrations.user is NOT NULL and this tenant has no owning user, so # the platform genuinely cannot record the connection. A clear refusal # beats a half-connected state the merchant cannot see. logger.error("Tenant %s has no owning user; cannot record integration", stored["tenant_id"]) _safe_track(stored["tenant_id"], "not_connected", "failed", "tenant has no owning user") return _error_redirect("no_owning_user", 409) logger.info("Connected Shopify store %s to tenant %s", domain, stored["tenant_id"]) _safe_track(stored["tenant_id"], "connected", "success") return RedirectResponse( _frontend_url(connected="shopify", shop=domain), status_code=307) @router.get("/status") async def status(tenant_id: str = Query(...)): """Connected sources for a tenant. Never returns credentials. Returns a list because a tenant may hold several sources once the HTTP API connector lands. """ return {"sources": get_sources(tenant_id)}