FastAPI API 보호

이 빠른 시작에서는 KeenAuth 액세스 토큰으로 FastAPI API를 보호합니다. Python SDK의 RequireAuth 의존성을 추가하고, 한 경로에 스코프를 요구하며, SDK 테스트 키트로 API를 오프라인에서 테스트합니다.

API에는 엔드포인트가 세 개 있습니다:

시작하기 전에

• KeenAuth 환경. 발급자는 https://your-environment.eu.keenauth.app 형식입니다.
• 환경에 등록된 API(API 만들기). 이 API의 identifier(예: https://api.example.com)이 API가 수락하는 토큰의 대상(audience)입니다.
• 이 API에 대한 messages:read 스코프가 있고, 이를 호출하는 애플리케이션에 부여되어 있어야 합니다.
• Python 3.10 이상.

1. SDK 설치

pip install "keenauth-python[fastapi]" uvicorn

SDK는 아직 공개되지 않았습니다. 공개 전까지 예제는 SDK 소스에서 설치합니다. 예제는 requirements.txt에서 모든 패키지를 정확한 버전으로 고정합니다.

2. 구성

API는 환경 변수에서만 구성을 읽습니다:

3. 경로 보호

create_app는 protector와 경로를 만듭니다. Depends(RequireAuth(protector))는 유효한 토큰을 요구하고 검증된 클레임을 반환합니다. scopes=[...]는 스코프도 요구합니다. install(app)는 오류를 application/problem+json 응답으로 만들고, 챌린지마다 WWW-Authenticate 헤더를 하나씩 붙입니다.

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가 구성에서 실제 verifier를 만듭니다. verifier는 처음 토큰을 볼 때 환경에서 서명 키를 가져옵니다. 이 의존성은 동기식이므로 FastAPI가 스레드 풀에서 실행합니다.

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. 실행하고 API 호출

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

첫 호출을 위해 messages:read 스코프가 있는 머신 투 머신 애플리케이션의 클라이언트 자격 증명 grant로 토큰을 받으세요:

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. 네트워크 없이 테스트

keenauth.testing는 공개 테스트 키로 서명한 토큰을 발급하고, testing.verifier는 그 토큰만 수락합니다. 테스트는 이 verifier를 create_app에 전달하고 Starlette 테스트 클라이언트로 앱을 호출하므로, 네트워크도 KeenAuth 계정도 필요하지 않습니다. 오프라인 테스트 키트를 참조하세요.

@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

예제 테스트는 유효한 토큰이 수락되는지 확인하고, 토큰이 없거나 만료되었거나 대상이 잘못된 경우, 스코프가 없는 경우, URL 쿼리의 토큰, 폐기된 토큰이 아래 상태 코드로 거부되는지 확인합니다. 또한 PUBLIC_BASE_URL에 대한 proof가 포함된 DPoP 바인딩 토큰을 보내고, app_from_env가 테스트 발급자를 거부하는지 확인합니다. 테스트는 서명된 token.revoked 전달도 보냅니다. 유효한 전달은 토큰을 폐기하고, 위조되었거나 변조되었거나 오래된 전달은 400을 받고 아무것도 폐기하지 않으며, 반복된 전달은 한 번만 효과가 있습니다.

응답

본문은 application/problem+json이고 WWW-Authenticate 헤더에는 표준 챌린지가 담깁니다. 전체 표는 미들웨어로 API 보호하기에 있습니다.

프록시 뒤에서

클라이언트가 DPoP를 사용하면 미들웨어는 각 proof가 클라이언트가 호출한 URL을 대상으로 만들어졌는지 확인합니다. PUBLIC_BASE_URL를 클라이언트가 호출하는 외부 URL로 설정하세요(예: https://api.example.com). 프록시나 로드 밸런서 뒤에 있는 경우에도 마찬가지입니다. 이 값은 필수이며 https를 사용해야 합니다(http는 localhost, 127.0.0.1, [::1]에만 허용됩니다). 이 값이 없으면 앱이 시작되지 않습니다. 미들웨어는 X-Forwarded-* 헤더를 직접 읽지 않습니다. 다만 Uvicorn은 연결이 --forwarded-allow-ips(기본값 127.0.0.1,::1)에 포함된 주소에서 올 때 X-Forwarded-Proto를 요청 URL에 적용합니다. PUBLIC_BASE_URL가 설정되어 있으면 검사되는 URL은 그에 의존하지 않습니다.

폐기된 토큰

액세스 토큰은 만료될 때까지 유효하므로, KeenAuth는 토큰이 폐기되면 API에 알려줍니다. 예제는 protector에 메모리 내 폐기 캐시를 전달하고, 이를 채우는 token.revoked 웹훅을 POST /webhooks/keenauth(webhook.py)에서 제공합니다.

1. 대시보드에서 URL https://<your host>/webhooks/keenauth로 웹훅 엔드포인트를 만들고 token.revoked을 구독하도록 설정한 후, 서명 시크릿을 KEENAUTH_WEBHOOK_SECRET로 내보내세요.
2. 핸들러는 원시 본문에 대해 SDK로 서명을 검증하고, 서명이 잘못되었거나 타임스탬프가 5분 넘게 차이 나면 400를 응답합니다. 이벤트의 셀렉터(access_token, session, user, client, org, org_user, client_user)를 폐기 마커로 변환하고 204를 응답합니다. 그러면 미들웨어가 not_before 이전에 발급된 일치하는 모든 토큰을 거부합니다.
3. KeenAuth는 최소 한 번 전달하므로, 핸들러는 각 webhook-id을 한 번만 적용합니다. 마커를 저장할 수 없으면 503를 응답하고 KeenAuth는 이벤트를 다시 전달합니다. 다른 이벤트 유형과 액세스 토큰과 일치할 수 없는 셀렉터(refresh_family, api)는 무시합니다.

핸들러는 이벤트 id, 셀렉터 유형, 사유를 기록하며, 본문, 서명, 시크릿은 기록하지 않습니다. 각 워커 프로세스는 자체 캐시를 가집니다. 워커나 인스턴스가 여러 개라면 대신 공유 저장소를 사용하세요. 폐기 캐시를 참조하세요.

다음 단계

• 미들웨어로 API 보호하기: Flask, 단계별 요구 사항, DPoP 및 모든 옵션.
• 토큰 검증: 미들웨어 아래의 verifier.
• 오프라인 테스트 키트.