FastAPI集成Keycloak出现Invalid token错误的解决方法

问题描述 #

在FastAPI应用中集成Keycloak后，通过/login接口获取令牌，携带令牌调用/project接口时收到"Invalid token"错误。

相关代码 #

keycloak.py #

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from keycloak import KeycloakOpenID
import requests
import logging
import os

from .config import settings


# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Keycloak configuration
KEYCLOAK_SERVER_URL = settings.KEYCLOAK_SERVER_URL
KEYCLOAK_REALM = settings.KEYCLOAK_REALM
KEYCLOAK_CLIENT_ID = settings.KEYCLOAK_CLIENT_ID
KEYCLOAK_CLIENT_SECRET = settings.KEYCLOAK_CLIENT_SECRET
ALGORITHM = "RS256"
TOKEN_URL = f"{KEYCLOAK_SERVER_URL}/realms/fastapi-realm/protocol/openid-connect/token"


# Initialize KeycloakOpenID
keycloak_openid = KeycloakOpenID(
    server_url=f"{KEYCLOAK_SERVER_URL}",
    client_id=KEYCLOAK_CLIENT_ID,
    realm_name=KEYCLOAK_REALM,
    client_secret_key=KEYCLOAK_CLIENT_SECRET,
    verify=False
)
config_well_known = keycloak_openid.well_known()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_token(token: str = Depends(oauth2_scheme)):
    try:
        decoded_token = keycloak_openid.decode_token(
            token,validate=True,
        )
        username = decoded_token['preferred_username']
        logger.info(f"Decoded token: {decoded_token}")
        # Verify the issuer claim
        issuer = decoded_token["iss"]

        expected_issuer = f"{KEYCLOAK_SERVER_URL}/realms/{KEYCLOAK_REALM}"
        logger.info(f"XXX_ issuer={issuer}")
        logger.info(f"XXX_ expected_issuer={expected_issuer}")
        if issuer != expected_issuer:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid issuer")
        
        logger.info(f"username: {username}")
        return decoded_token
    except Exception as e:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")

main.py #

from fastapi import Depends, FastAPI, HTTPException, status, Security


from .keycloak import verify_token, oauth2_scheme, KEYCLOAK_CLIENT_ID, KEYCLOAK_CLIENT_SECRET, TOKEN_URL

app = FastAPI()

@app.post("/project", response_model=schemas.Project, tags=["project",])
def create_project(project: schemas.CreateProject, db: Session = Depends(get_db), payload: dict = Security(verify_token)):
    ...


@app.post("/login")
def get_token(body: schemas.Login):
    data = {
        "grant_type": "password", # TODO: clarify grant_type client_credentials (requires only client id and secret or password - requires password and login)
        "client_id": KEYCLOAK_CLIENT_ID,
        "client_secret": KEYCLOAK_CLIENT_SECRET,
        "password": body.password,
        "username": body.login
    }
    response = requests.post(TOKEN_URL, data=data,  verify=False)
    return JSONResponse(status_code=response.status_code, content=response.json())

核心修复步骤 #

1. 修正TOKEN_URL的Realm名称 #

当前TOKEN_URL硬编码了fastapi-realm，但配置中使用的是settings.KEYCLOAK_REALM，会导致获取的令牌来自错误的Realm，验证必然失败。替换为：

TOKEN_URL = f"{KEYCLOAK_SERVER_URL}/realms/{KEYCLOAK_REALM}/protocol/openid-connect/token"

2. 调整OAuth2认证的tokenUrl路径 #

OAuth2PasswordBearer的tokenUrl需要匹配实际的登录接口路径/login，否则FastAPI的OAuth2流程会出现不兼容问题：

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login")

3. 细化异常捕获与日志输出 #

原代码用笼统的Exception捕获所有错误，无法定位具体问题（比如令牌过期、签名无效、受众不匹配等）。替换为具体异常捕获并增加日志：

def verify_token(token: str = Depends(oauth2_scheme)):
    try:
        # 可选：指定audience匹配令牌中的aud字段，示例中令牌aud为"account"，可改为KEYCLOAK_CLIENT_ID根据实际配置调整
        decoded_token = keycloak_openid.decode_token(
            token,
            validate=True,
            audience="account"
        )
        username = decoded_token['preferred_username']
        logger.info(f"Decoded token: {decoded_token}")
        
        issuer = decoded_token["iss"]
        expected_issuer = f"{KEYCLOAK_SERVER_URL}/realms/{KEYCLOAK_REALM}"
        logger.info(f"XXX_ issuer={issuer}")
        logger.info(f"XXX_ expected_issuer={expected_issuer}")
        
        if issuer != expected_issuer:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid issuer")
        
        logger.info(f"username: {username}")
        return decoded_token
    except JWTError as e:
        logger.error(f"JWT验证错误: {str(e)}")
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=f"JWT错误: {str(e)}")
    except Exception as e:
        logger.error(f"令牌验证失败: {str(e)}")
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=f"验证错误: {str(e)}")

4. 检查配置一致性 #

确认settings中的KEYCLOAK_SERVER_URL、KEYCLOAK_REALM等参数与Keycloak后台完全一致，注意URL末尾是否带斜杠（如https://xxx vs https://xxx/），避免拼接expected_issuer时出现格式错误。

额外排查建议 #

• 调用/login获取令牌后，解码令牌检查iss、exp、aud字段是否与预期匹配；
• 确保Keycloak客户端的Access Type设置为confidential（因为使用了client_secret），且Web Origins配置正确；
• 生产环境建议移除verify=False，配置正确的证书信任链，避免安全风险。

内容的提问来源于stack exchange，提问作者mascai