I'm developing a Shopify app using FastAPI and encountering issues with the OAuth flow. Specifically, I'm facing two problems:
- Route Not Found (404 Error): The /install endpoint is returning a 404 error during testing.
- 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)