Simple FastAPI Example¶
The following is a single module FastAPI app written as a demo of the Saleor SDK Python package. This should not be treated as a production ready application.
For the sake of time let's import everything that will be needed upfront, initialize the app and put a JWKS storage in place:
from fastapi import FastAPI, Request, Depends, Header
from pydantic_settings import BaseSettings
import httpx
from jwt.api_jwk import PyJWKSet
from saleor_sdk.schemas.manifest import Manifest, Webhook
from saleor_sdk.schemas.enums import Permission, WebhookAsyncEvents
from saleor_sdk.crypto.utils import decode_webook_payload, decode_jwt
from saleor_sdk.crypto.exceptions import JWKSKeyMissing
class Settings(BaseSettings):
debug: bool = False
settings = Settings(debug=True)
app = FastAPI(debug=settings.debug)
saleor_jwks = None
Declare the App manifest¶
@app.get("/api/manifest")
async def manifest(request: Request):
return Manifest(
id="simple-sdk-test-app",
version="0.0.0",
name="Simple SDK Test APP",
permissions=[Permission.MANAGE_ORDERS, "MANAGE_CHECKOUTS"],
app_url=str(request.url_for("app_config")),
token_target_url=str(request.url_for("register")),
webhooks=[
Webhook(
name="Order Handler",
async_events=[WebhookAsyncEvents.ORDER_CREATED, "ORDER_UPDATED"],
query="subscription { event { issuedAt issuingPrincipal { ... on App { id } ... on User { id } } ... on OrderCreated { order { id }} ... on OrderUpdated { order { id }}}}",
target_url=str(request.url_for("order_handler")),
is_active=True,
)
],
)
There are a few things to notice here:
- enums - the SDK will try to keep the definitions up-to-date with Saleor but there is no promise this will always meet your needs (time-wise), this is why you can use both, the definitions from
saleor_sdk.schemas.enumsand plain strings. request.url_for()- Saleor needs a full URL to your endpoint, here we leverage FastAPI's resolver (be mindful about what your proxy is doing, i.e. it's easy to hide crucial information from FastAPI with Gunicorn)- Subscription queries - the payload that webhook is to carry.
More on Subscription queries¶
Since Saleor 3.2 (and 3.6 for synchronous events) one can define the payload that Saleor will send to an app - this allows you to define a payload that is meaningful to the app.
Saleor docs: Subscription Webhook Payloads
A formatted query from the example looks like this:
subscription {
event {
issuedAt
issuingPrincipal {
... on App {
id
}
... on User {
id
}
}
... on OrderCreated {
order {
id
}
}
... on OrderUpdated {
order {
id
}
}
}
}
and will result in the following payload being sent to the app:
{
"issuedAt": "2022-12-12T00:37:17.405467+00:00",
"issuingPrincipal": {
"id": "VXNlcjoyMDU5ODA1MTg0"
},
"order": {
"id": "T3JkZXI6MWFkNzZjOTctZDkxNy00NjRmLWIwNzUtOTljNzcwY2IzOWI4"
}
}
Define some commonly used dependencies¶
async def get_saleor_event(saleor_event: str = Header(..., alias="Saleor-Event")):
return saleor_event
async def get_saleor_domain(saleor_domain: str = Header(..., alias="Saleor-Domain")):
return saleor_domain
async def get_saleor_signature(
saleor_signature: str = Header(..., alias="Saleor-Signature")
):
return saleor_signature
JWKS and user JWT verification¶
There are currently two ways to verify the authenticity of a Saleor issued JWT: - online - by calling Saleor to check (deprecated by Saleor) - offline - by checking with Saleor's public keyset. This library only supports the offline method.
Saleor instances expose a /.well-known/jwks.json endpoint which exposes the public part of the RSA key used to sign the JWTs.
With that we can verify if the incoming token is indeed coming straight from Saleor and if we can trust it's claims.
To do that we need a storage within the app's memory to hold on to the JWKS (JSON Web Key Set), in this example we are simply using a global variable.
The get_saleor_user dependency ensures the JWKS was initialized (first request in the runtime), then validates the token. If the local JWKS is missing a key, a JWKSKeyMissing error will be raised and an attempt to get a fresh one from Saleor will be made - this might happen when Saleor rotates the keys and starts signing JWTs with the new key.
async def fetch_jwks(saleor_domain):
async with httpx.AsyncClient() as client:
response = await client.get(f"https://{saleor_domain}/.well-known/jwks.json")
return response.content
async def get_saleor_user(
saleor_domain: str = Depends(get_saleor_domain),
saleor_token: str = Header(..., alias="Saleor-Token"),
):
global saleor_jwks
if not saleor_jwks:
saleor_jwks = PyJWKSet.from_json(await fetch_jwks(saleor_domain))
max_attempts = 1
while max_attempts:
try:
return decode_jwt(
jwt=saleor_token,
jwks=saleor_jwks,
)
except JWKSKeyMissing as exc:
if max_attempts:
saleor_jwks = PyJWKSet.from_json(await fetch_jwks(saleor_domain))
max_attempts -= 1
else:
raise
Here's how it roughly works in sequence
sequenceDiagram
actor USR as User
participant SAL as Saleor
participant APP as App
participant STO as JWKS Storage
USR ->> SAL: Here are my credentials, give me a JWT
SAL ->> USR: Here's your JWT
USR ->> APP: Give me that secret resource
APP ->> STO: I've got a JWT with kid=1 do you have something on that?
alt Storage does not have the kid=1
STO ->> APP: No I don't
APP ->> SAL: Fetch JWKS
SAL ->> APP: JWKS
APP ->> STO: Save JWKS
else
STO ->> APP: Yes, here is the public key
end
APP ->> APP: Verify the JWT signature
APP ->> USR: Here's the dataWebhook signature verification¶
A very similar process applies to webhook authenticity verification. Leveraging the public key pair issued by Saleor the signature of a JWS (a JWT with detached payload) that is sent with a webhook can be verified in an offline manner.
async def verify_webhook_signature(
request: Request,
saleor_domain: str = Depends(get_saleor_domain),
jws: str = Depends(get_saleor_signature),
):
global saleor_jwks
if not saleor_jwks:
saleor_jwks = PyJWKSet.from_json(await fetch_jwks(saleor_domain))
max_attempts = 1
while max_attempts:
try:
return decode_webook_payload(
jws=jws,
jwks=saleor_jwks,
webhook_payload=await request.body(),
)
except JWKSKeyMissing as exc:
if max_attempts:
saleor_jwks = PyJWKSet.from_json(await fetch_jwks(saleor_domain))
max_attempts -= 1
else:
raise
Finally define the app endpoints¶
You need the two endpoints required by Saleor's framework:
@app.get("/", name="app_config")
async def app_config():
return "OK"
@app.post("/api/register", name="register")
async def register():
return "OK"
app_config- this is the endpoint that is expected to respond with HTML that will initialize a React App in the Saleor Dashboard (read more on @saleor/app-sdk)register- this endpoint receives the app_token which needs to be persisted securely, it's used by the app to authenticate with Saleor
Further we define the order webhook handler which leverages our verify_webhook_signature dependency.
@app.post("/api/webhook/order", name="order_handler")
async def order_handler(
request: Request,
_verify_webhook_signature=Depends(verify_webhook_signature),
event_type=Depends(get_saleor_event),
):
print(event_type) # Use in case you have one handler for many event types
print(await request.body())
return "OK"
And an additional endpoint that could be used by the Dashboard UI to for example change the app configuration
async def user_login(saleor_user=Depends(get_saleor_user)):
print(saleor_user)
return "OK"
Running the complete example¶
Install the examples dependencies with:
hatch -e examples shell
Then navigate to the example and run it with uvicorn:
cd docs/examples/simple
uvicorn app:app --reload