FastAPI API'sini koruma

Bu hızlı başlangıç, bir FastAPI API'sini KeenAuth erişim belirteçleriyle korur. Python SDK'sının RequireAuth bağımlılığını eklersiniz, bir rotada bir kapsam gerektirirsiniz ve API'yi SDK test kiti ile çevrimdışı test edersiniz.

API'nin üç uç noktası vardır:

Başlamadan önce

• Bir KeenAuth ortamı. Issuer adresi https://your-environment.eu.keenauth.app biçimindedir.
• Ortamda kayıtlı bir API (API oluşturun). Bu API'nin identifier değeri (örneğin https://api.example.com), API'nizin kabul ettiği belirteçlerin hedef kitlesidir.
• Bu API üzerinde, onu çağıran uygulamaya verilmiş messages:read kapsamı.
• Python 3.10 veya üzeri.

1. SDK'yı kurun

pip install "keenauth-python[fastapi]" uvicorn

SDK henüz yayımlanmadı. Yayımlanana kadar örnek, SDK kaynağından kurar. Örnek, her paketi requirements.txt içinde tam bir sürüme sabitler.

2. Yapılandırın

API yapılandırmasını yalnızca ortam değişkenlerinden okur:

3. Rotalarınızı koruyun

create_app korumayı ve rotaları oluşturur. Depends(RequireAuth(protector)) geçerli bir belirteç gerektirir ve doğrulanmış bildirimlerini döndürür; scopes=[...] ayrıca kapsam ister. install(app) hataları, her challenge için bir WWW-Authenticate başlığı içeren application/problem+json yanıtlarına dönüştürür.

def create_app(
    verifier: AccessTokenVerifier,
    *,
    public_base_url: str,
    revocation: Optional[RevocationChecker] = None,
    webhook_secrets: Sequence[str] = (),
) -> FastAPI:
    """Builds the API. The verifier is a parameter so tests can pass the offline test kit's verifier. With
    webhook_secrets (the idp_whsec_ secrets of your webhook endpoint) the token.revoked webhook is served and feeds
    `revocation`, which must then be a MemoryRevocationCache."""
    protector = Protector(
        verifier,
        public_base_url=public_base_url,
        revocation=revocation,
        # The error never contains the token or claim values; never log the Authorization or DPoP header.
        on_error=lambda err: log.warning("rejected: %s", err),
    )
    app = FastAPI()
    install(app)  # problem+json responses and one WWW-Authenticate header per challenge

    @app.get("/api/public")
    def public() -> dict[str, Any]:
        # No token needed.
        return {"message": "Hello from a public endpoint."}

    @app.get("/api/private")
    def private(claims: AccessClaims = Depends(RequireAuth(protector))) -> dict[str, Any]:
        # Any valid access token for this API.
        return {"message": "Hello from a private endpoint.", "sub": claims.subject}

    @app.get("/api/messages")
    def messages(claims: AccessClaims = Depends(RequireAuth(protector, scopes=["messages:read"]))) -> dict[str, Any]:
        # A valid access token that also carries the messages:read scope.
        return {"messages": ["Hello from a scoped endpoint."]}

    # token.revoked deliveries from KeenAuth.
    if webhook_secrets:
        if not isinstance(revocation, MemoryRevocationCache):
            raise ValueError("the token.revoked webhook needs a MemoryRevocationCache to feed")
        app.include_router(webhook_router(RevocationWebhook(webhook_secrets, revocation)))
    else:
        log.warning("KEENAUTH_WEBHOOK_SECRET is not set: token.revoked events are not received, so revoked tokens stay valid until they expire")

    return app

app_from_env yapılandırmadan gerçek doğrulayıcıyı oluşturur. Doğrulayıcı, bir belirteci ilk gördüğünde imzalama anahtarlarını ortamınızdan alır. Bağımlılık senkrondur, bu yüzden FastAPI onu iş parçacığı havuzunda çalıştırır.

def app_from_env() -> FastAPI:
    """The production app: KEENAUTH_ISSUER, KEENAUTH_AUDIENCE and PUBLIC_BASE_URL are required."""
    issuer = os.environ.get("KEENAUTH_ISSUER", "")
    audience = os.environ.get("KEENAUTH_AUDIENCE", "")
    # The URL clients call: DPoP proofs are checked against it, never against the Host header.
    public_base_url = os.environ.get("PUBLIC_BASE_URL", "")
    if not issuer or not audience or not public_base_url:
        raise SystemExit("KEENAUTH_ISSUER, KEENAUTH_AUDIENCE and PUBLIC_BASE_URL are required")
    # The verifier refuses an issuer under the reserved .invalid domain, so a test token can never pass here.
    verifier = Verifier(VerifierConfig(issuer=issuer, audience=audience))
    # Fed by the token.revoked webhook.
    revocation = MemoryRevocationCache()
    # The signing secret (idp_whsec_...) of the webhook endpoint that delivers token.revoked; comma separated while rotating.
    webhook_secrets = [x.strip() for x in os.environ.get("KEENAUTH_WEBHOOK_SECRET", "").split(",") if x.strip()]
    return create_app(verifier, public_base_url=public_base_url, revocation=revocation, webhook_secrets=webhook_secrets)

4. Çalıştırın ve API'yi çağırın

export KEENAUTH_ISSUER=https://your-environment.eu.keenauth.app
export KEENAUTH_AUDIENCE=https://api.example.com
export PUBLIC_BASE_URL=https://api.example.com
uvicorn main:app_from_env --factory --port 8080

İlk çağrı için, API'nizde messages:read kapsamına sahip bir makineden makineye uygulamanın istemci kimlik bilgileri akışıyla bir belirteç alın:

ACCESS_TOKEN=$(curl -s -u "$CLIENT_ID:$CLIENT_SECRET" \
  -d grant_type=client_credentials \
  -d resource=https://api.example.com \
  -d scope=messages:read \
  https://your-environment.eu.keenauth.app/oauth2/token | jq -r .access_token)

curl -i http://localhost:8080/api/messages -H "Authorization: Bearer $ACCESS_TOKEN"

5. Ağ olmadan test edin

keenauth.testing ortak test anahtarıyla imzalanmış belirteçler üretir ve testing.verifier yalnızca bu belirteçleri kabul eder. Testler bu doğrulayıcıyı create_app öğesine geçirir ve uygulamayı Starlette'in test istemcisiyle çağırır; bu yüzden ağa ya da bir KeenAuth hesabına ihtiyaç duymazlar. Bkz. Çevrimdışı test kiti.

@pytest.fixture()
def api() -> Api:
    cache = MemoryRevocationCache()
    app = create_app(
        testing.verifier(audience=AUDIENCE),
        public_base_url=BASE_URL,
        revocation=cache,
        webhook_secrets=[testing.webhook_secret()],
    )
    return TestClient(app, base_url=BASE_URL), cache

def test_scope(api: Api) -> None:
    client, _ = api
    no_scope = testing.mint_access_token(AUDIENCE)
    assert_problem(client.get("/api/messages", headers=bearer(no_scope)), 403, "insufficient_scope", 'scope="messages:read"')
    with_scope = testing.mint_access_token(AUDIENCE, claims={"scope": "profile messages:read"})
    assert client.get("/api/messages", headers=bearer(with_scope)).status_code == 200

pip install --no-deps -r requirements-dev.txt
python -m pytest -q

Örneğin testleri, geçerli bir belirtecin kabul edildiğini; eksik, süresi dolmuş veya yanlış hedef kitleli bir belirtecin, eksik bir kapsamın, URL sorgusundaki bir belirtecin ve iptal edilmiş bir belirtecin aşağıdaki durum kodlarıyla reddedildiğini denetler. Ayrıca PUBLIC_BASE_URL için bir kanıtla DPoP'a bağlı bir belirteç gönderir ve app_from_env öğesinin test issuer'ını reddettiğini denetler. Testler ayrıca imzalı token.revoked teslimatları da gönderir: geçerli bir teslimat belirteci iptal eder; sahte, değiştirilmiş veya eski bir teslimat 400 alır ve hiçbir şeyi iptal etmez; tekrarlanan bir teslimatın tek bir etkisi olur.

Yanıtlar

Gövde application/problem+json ve WWW-Authenticate başlığı standart challenge'ı taşır. Tam tablo Ara katmanla API'nizi koruma sayfasındadır.

Vekil sunucu arkasında

İstemciler DPoP kullanıyorsa ara katman, her kanıtın istemcinin çağırdığı URL için yapıldığını denetler. PUBLIC_BASE_URL öğesini, istemcilerin çağırdığı dış URL'ye ayarlayın, örneğin https://api.example.com; bu, bir vekil sunucu ya da yük dengeleyici arkasında da geçerlidir. Bu zorunludur ve https kullanmalıdır (yalnızca localhost, 127.0.0.1 veya [::1] için http); bu olmadan uygulama başlamaz. Ara katman X-Forwarded-* başlıklarını kendisi okumaz. Ancak Uvicorn, bağlantı --forwarded-allow-ips içindeki bir adresten geldiğinde istek URL'sine X-Forwarded-Proto uygular (varsayılan olarak 127.0.0.1,::1); PUBLIC_BASE_URL ayarlandığında denetlenen URL buna bağlı olmaz.

İptal edilmiş belirteçler

Bir erişim belirteci süresi dolana kadar geçerli kalır; bu yüzden KeenAuth, belirteçler iptal edildiğinde API'nize bildirir. Örnek, koruyucuya bellek içi bir iptal önbelleği geçirir ve bunu besleyen bir token.revoked webhook'unu POST /webhooks/keenauth (webhook.py) adresinde sunar.

1. Panelde, token.revoked olayına abone olan https://<your host>/webhooks/keenauth URL'li bir webhook uç noktası oluşturun ve imzalama sırrını KEENAUTH_WEBHOOK_SECRET olarak dışa aktarın.
2. İşleyici, ham gövde üzerinde imzayı SDK ile doğrular ve imza geçersizse ya da zaman damgası beş dakikadan fazla sapmışsa 400 yanıtını verir. Olayın seçicisini (access_token, session, user, client, org, org_user, client_user) bir iptal işaretçisine eşler ve 204 yanıtını verir; ara katman daha sonra not_before öncesinde verilmiş eşleşen her belirteci reddeder.
3. KeenAuth en az bir kez teslim eder, bu yüzden işleyici her webhook-id öğesini bir kez uygular. İşaretçi kaydedilemezse 503 yanıtını verir ve KeenAuth olayı yeniden teslim eder. Diğer olay türlerini ve hiçbir erişim belirtecinin eşleşemeyeceği seçicileri (refresh_family, api) yok sayar.

İşleyici olay kimliğini, seçici türünü ve nedeni günlüğe yazar; gövdeyi, imzayı veya sırrı asla yazmaz. Her çalışan süreç kendi önbelleğine sahiptir; birden çok çalışan ya da örnek varsa bunun yerine paylaşılan bir depo kullanın. Bkz. İptal önbelleği.

Sonraki adımlar

• Ara katmanla API'nizi koruma: Flask, adım artırma gereksinimleri, DPoP ve tüm seçenekler.
• Belirteçleri doğrulama: ara katmanın altındaki doğrulayıcı.
• Çevrimdışı test kiti.