KeycloakのService accounts rolesで発行されたTokenを使用したFastAPIでの認証方法

前提条件

初めての実装です

実現したいこと

以下のようにKeycloakからTokenを取得した場合に、取得したTokenをFastAPI側で検証して問題がなければFastAPIの処理を実行したいです。

python
1import requests, base64, json
2
3class TokenGetter:
4    _grant_type: str = "client_credentials"
5    _client_id: str = "demo-client"
6    _client_secret: str = "bk7CmZgCJ4KlmbVyLHNPb8XFJEf0DX8e"
7    _token_endpoint: str = "http://keycloak:8080/realms/demo/protocol/openid-connect/token"
8    """
9    TOKEN取得URLです。
10    Docker上のコンテナ名がIPアドレス替わりになるので注意
11    """
12    def __init__(self) -> None:
13        pass
14
15    def get_token(self) -> str | None:
16
17        authorization = base64.b64encode((self._client_id + ":" + self._client_secret).encode()).decode()
18        headers = {
19            "Authorization": f"Basic {authorization}",
20            "Content-Type": "application/x-www-form-urlencoded"
21        }        
22        data = f"grant_type={self._grant_type}"
23        
24        responce = requests.post(self._token_endpoint, headers=headers, data=data)
25        print(responce)
26
27        if responce is None:
28            return None
29
30        result = json.loads(responce.text)
31        if "access_token" not in result:
32            return None
33        # ここで取得したTokenを利用してFastAPI側でTokenの検証とアクセス制御を行いたい
34        return result["access_token"]

取得したTokenを使用してFastAPI側でTokenの検証とアクセス制御を行いたい

ここに実現したいことを箇条書きで書いてください。

▲▲機能を動作するようにする

環境

VSCode Dev Container

json
1// python用のデフォルトコンテナ
2{
3    "name": "webapi oauth 2.0 sample",
4    "dockerComposeFile": [
5        "docker-compose-keyclock.yml",
6        "docker-compose.yml"
7    ],
8    // docker-composeのserviceと一致する名前
9    "service": "default",
10    // dockerのイメージで使用するデフォルトワークスペース
11    "workspaceFolder": "/workspace",
12    "shutdownAction": "stopCompose",
13    // カスタマイズ設定
14    "customizations": {
15        // VSCode関連のカスタマイズ設定
16        "vscode": {
17            // デフォルトでインストールする拡張機能
18            "extensions": [
19                "MS-CEINTL.vscode-language-pack-ja",
20                "Gruntfuggly.todo-tree",
21                "mhutchie.git-graph",
22                "ms-python.python",
23                "esbenp.prettier-vscode",
24                "ms-python.black-formatter",
25                "humao.rest-client",
26                "njpwerner.autodocstring",
27                "yzhang.markdown-all-in-one",
28                "ms-azuretools.vscode-docker"
29            ],
30            // インストールした拡張機能を含めたデフォルトの設定
31            "settings": {
32                "black-formatter.args": [
33                    // 折り返し列数指定
34                    "--line-length=150"
35                ],
36                "editor.rulers": [120, 150]
37            }
38        }
39    }
40}

keycloak

docker-compose-keyclock.yml

yml
1version: "3.9"
2
3networks:
4  dev_network:
5    external: true
6
7volumes:
8  mysql_data:
9    driver: local
10
11services:
12  mysql:
13    image: mysql:8.0.33
14    container_name: mysql
15    volumes:
16      - mysql_data:/var/lib/mysql
17    environment:
18      MYSQL_ROOT_PASSWORD: password
19      MYSQL_DATABASE: keycloak
20      MYSQL_USER: keycloak
21      MYSQL_PASSWORD: password
22    ports:
23      - 3306:3306
24    networks:
25      - dev_network
26
27  keycloak:
28    image: quay.io/keycloak/keycloak:21.0.2
29    container_name: keycloak
30    environment:
31      KEYCLOAK_ADMIN: admin
32      KEYCLOAK_ADMIN_PASSWORD: admin
33      KC_DB : mysql
34      KC_DB_URL: jdbc:mysql://mysql:3306/keycloak
35      KC_DB_URL_DATABASE: keycloak
36      KC_DB_USERNAME: keycloak
37      KC_DB_PASSWORD: password
38    ports:
39      - 8080:8080
40    command:
41      - start-dev
42    networks:
43      - dev_network

fastapi

docker-compose.yml

yml
1version: "3.9"
2
3networks:
4  dev_network:
5    external: true
6
7services:
8  default:                                    # devcontainer.jsonのserviceと一致する名前
9    container_name: 'webapi_oauth_2_sample'   # Dockerのコンテナ名
10    hostname: 'webapi_oauth_2'                # ホスト名。ゲストOSのシェルに表示される名前
11
12    # 
13    build: .                  # Dockerfileがあるディレクトリ
14    restart: always           # 常に再起動
15    working_dir: '/workspace' # デフォルトの作業ディレクトリ
16    tty: true                 # コンテナを終了させずに待たせる
17    ports:
18      - 7000:7000
19    volumes:                  # ホストとゲストのディレクトリを紐づける
20      - type: bind            # ホストのディレクトリをゲストにマウントさせる
21        source: ..            # ホストのdockerファイルの直上ディレクトリを指定
22        target: /workspace    # ゲストの`/workspace`にマウントさせる
23    networks:
24      - dev_network

Dockerfile

Dockerfile
1FROM python:3.11.6
2
3RUN apt-get update && apt-get install -y \
4    curl \
5    git \
6 && apt-get clean \
7 && rm -rf /var/lib/apt/lists/*
8
9# 環境変数設定(日本語とか)
10ENV TZ='Asia/Tokyo'
11
12COPY requirements.txt .
13RUN pip install --upgrade pip
14RUN pip install -r requirements.txt

確認したい事

前提

JWTトークンの実装でいいと思っています。
https://fastapi.tiangolo.com/tutorial/security/oauth2-jwt/

わからない部分

Keycloakとの連携はFastAPI側はどのように記載すればよいのか

JWTトークンの認証自体はFastAPIのサンプルとしてあるが、Keycloakへトークンの認証を依頼する形になると思っているが、ここの連携方法がわからない。(これが違ったらすいません)

行動規範の内容に同意します

回答1件

自己解決

発行したトークンの有効無効をKeycloak側の以下のURLで確認できたので、解決。
【トークンの発行】
http://keycloak:8080/realms/demo/protocol/openid-connect/token

【トークンの確認】
http://keycloak:8080/realms/demo/protocol/openid-connect/token/introspect

最終的には以下のようなコードに

python
1GRANT_TYPE: str = "client_credentials"
2CLIENT_ID: str = "demo-client"
3CLIENT_SECRET: str = "bk7CmZgCJ4KlmbVyLHNPb8XFJEf0DX8e"
4TOKEN_INTROSPECT_ENDPOINT: str = "http://keycloak:8080/realms/demo/protocol/openid-connect/token/introspect"
5
6
7oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
8
9def get_token_info(token: str = Security(oauth2_scheme)) -> bool:
10    response = requests.post(
11        TOKEN_INTROSPECT_ENDPOINT,
12        data={"token": token, "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET},
13        headers={"Content-Type": "application/x-www-form-urlencoded"}
14    )
15
16    if response.status_code != 200:
17        return False
18    
19    active_info = json.loads(response.text)
20    return active_info["active"]
21
22@app.get("/resources/hoge")
23async def get_resources_hoge(request: Request, is_active = Depends(get_token_info)):
24    
25    TOKEN_TAG = "authorization"
26
27    headers = request.headers
28    if TOKEN_TAG not in headers:
29        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail={"msg": "token情報の取得に失敗しました"})
30    
31    token = headers['authorization']
32
33    if is_active:
34        return {"msg": "アクティブなトークンです", "token": token}
35
36    return {"msg": "無効なトークンです", "token": token}