0

I'm developing a Shopify app using FastAPI and encountering issues with the OAuth flow. Specifically, I'm facing two problems:

  1. Route Not Found (404 Error): The /install endpoint is returning a 404 error during testing.
  2. HMAC Verification Failing (400 Error): The HMAC verification fails during the callback, resulting in a 400 error.

Environment: Python 3.11.10 FastAPI pytest 8.3.4

What I've Tried: Verified router inclusion in main.py Checked environment variable mocking in tests Ensured HMAC generation logic matches Shopify's documentation

My questions basically are: Why is the /install route returning a 404 error? How can I resolve the HMAC verification failure? Any insights or suggestions would be greatly appreciated!

I've been also trying to debug this with Cursor AI (different LLMs) but several hours later I have to give up.

main.py

from app.routers import shopify

app = FastAPI()

app.include_router(shopify.router, prefix="/shopify", tags=["shopify"])

shopify.py

from fastapi import APIRouter, HTTPException
from fastapi.responses import RedirectResponse
import os
import secrets
import hmac
import hashlib

router = APIRouter()

SHOPIFY_API_KEY = os.getenv("SHOPIFY_API_KEY")
SHOPIFY_API_SECRET = os.getenv("SHOPIFY_API_SECRET")
REDIRECT_URI = os.getenv("SHOPIFY_REDIRECT_URI")
SCOPES = "read_products,read_orders,read_analytics"

def verify_oauth_request(params: dict, secret: str) -> bool:
    params_without_hmac = params.copy()
    params_without_hmac.pop('hmac', None)
    sorted_params = sorted((k, str(v)) for k, v in params_without_hmac.items())
    query_string = '&'.join(f"{k}={v}" for k, v in sorted_params)
    calculated_hmac = hmac.new(
        secret.encode('utf-8'),
        query_string.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(calculated_hmac, params['hmac'])

@router.get("/install")
async def install(shop: str):
    if not shop:
        raise HTTPException(status_code=400, detail="Shop parameter required")
    nonce = secrets.token_urlsafe(16)
    install_url = (
        f"https://{shop}/admin/oauth/authorize?"
        f"client_id={SHOPIFY_API_KEY}&"
        f"scope={SCOPES}&"
        f"redirect_uri={REDIRECT_URI}&"
        f"state={nonce}&"
        f"access_mode=offline"
    )
    return RedirectResponse(url=install_url, status_code=307)

@router.get("/callback")
async def callback(code: str, shop: str, state: str, hmac: str):
    params = {"code": code, "shop": shop, "state": state, "hmac": hmac}
    if not verify_oauth_request(params, SHOPIFY_API_SECRET):
        raise HTTPException(status_code=400, detail="Invalid HMAC")
    # Token exchange logic...

test_oauth.py

from fastapi.testclient import TestClient
from app.main import app
import unittest
from unittest.mock import patch

client = TestClient(app)

class TestOAuthFlow(unittest.TestCase):
    @patch('app.routers.shopify.os.getenv')
    def test_install_endpoint(self, mock_getenv):
        mock_getenv.side_effect = lambda key: {
            'SHOPIFY_API_KEY': 'test_key',
            'SHOPIFY_API_SECRET': 'test_secret',
            'SHOPIFY_REDIRECT_URI': 'http://127.0.0.1:8000/shopify/callback',
            'SCOPES': 'read_products,read_orders,read_inventory'
        }.get(key, None)
        shop = "test_shop.myshopify.com"
        response = client.get(f"/shopify/install?shop={shop}")
        self.assertEqual(response.status_code, 307)

    @patch('app.routers.shopify.requests.post')
    @patch('app.routers.shopify.os.getenv')
    def test_callback_exchange_token(self, mock_getenv, mock_post):
        mock_getenv.side_effect = lambda key: {
            'SHOPIFY_API_KEY': 'test_key',
            'SHOPIFY_API_SECRET': 'test_secret',
        }.get(key, None)
        mock_post.return_value.status_code = 200
        mock_post.return_value.json.return_value = {"access_token": "test_access_token"}
        params = {"code": "test_code", "shop": "test_shop.myshopify.com", "state": "test_state", "hmac": "test_hmac"}
        query_string = '&'.join(f"{k}={v}" for k, v in params.items())
        response = client.get(f"/shopify/callback?{query_string}")
        self.assertEqual(response.status_code, 200)

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.