Keycloak: implement client_credentials grant flow

providers/keycloak/docs/auth-manager/token.rst

@@ -25,6 +25,14 @@ In order to use the :doc:`Airflow public API <apache-airflow:stable-rest-api-ref
 You can then include this token in your Airflow public API requests.
 To generate a JWT token, use the ``Create Token`` API in :doc:`/api-ref/token-api-ref`.
 
+Several endpoints exist to create tokens depending on the authentication method you want to use.
+
+If a user or service needs to interact with the Airflow public API, they can create a token using their credentials.
+
+- ``/auth/token``: Create token using username and password or client credentials with a ``[config][api_auth]jwt_expiration_time`` expiration time.
+- ``/auth/token/cli``: Create token for Airflow CLI using username and password with a ``[config][api_auth]jwt_cli_expiration_time`` expiration time.
+
+
 Example
 '''''''
 
@@ -40,3 +48,20 @@ Example
         }'
 
 This process will return a token that you can use in the Airflow public API requests.
+The body can also contain a ``grant_type`` field with value ``password`` but it is optional since it is the default value.
+
+.. code-block:: bash
+
+    ENDPOINT_URL="http://localhost:8080 "
+    curl -X 'POST' \
+        "${ENDPOINT_URL}/auth/token" \
+        -H 'Content-Type: application/json' \
+        -d '{
+        "grant_type": "client_credentials",
+        "client_id": "<client_id>",
+        "client_secret": "<client_secret>"
+        }'
+
+If other services need to interact with the Airflow public API, they can create a token using the client credentials grant flow.
+The client must live in the same realm the Auth Manager is configured to use. Its service account must have the appropriate roles / permissions to access the Airflow public API.
+This process will return a token obtained using client credentials grant flow.

providers/keycloak/src/airflow/providers/keycloak/auth_manager/datamodels/token.py

@@ -17,9 +17,15 @@
 # under the License.
 from __future__ import annotations
 
-from pydantic import Field
+from typing import Annotated, Literal
+
+from pydantic import Field, RootModel, model_validator
 
 from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
+from airflow.providers.keycloak.auth_manager.services.token import (
+    create_client_credentials_token,
+    create_token_for,
+)
 
 
 class TokenResponse(BaseModel):
@@ -28,8 +34,47 @@ class TokenResponse(BaseModel):
     access_token: str
 
 
-class TokenBody(StrictBaseModel):
-    """Token serializer for post bodies."""
+class TokenPasswordBody(StrictBaseModel):
+    """Password grant token serializer for post bodies."""
 
+    grant_type: Literal["password"] = "password"
     username: str = Field()
     password: str = Field()
+
+    def create_token(self, expiration_time_in_seconds: int) -> str:
+        """Create token using password grant."""
+        return create_token_for(
+            self.username, self.password, expiration_time_in_seconds=expiration_time_in_seconds
+        )
+
+
+class TokenClientCredentialsBody(StrictBaseModel):
+    """Client credentials grant token serializer for post bodies."""
+
+    grant_type: Literal["client_credentials"]
+    client_id: str = Field()
+    client_secret: str = Field()
+
+    def create_token(self, expiration_time_in_seconds: int) -> str:
+        """Create token using client credentials grant."""
+        return create_client_credentials_token(
+            self.client_id, self.client_secret, expiration_time_in_seconds=expiration_time_in_seconds
+        )
+
+
+TokenUnion = Annotated[
+    TokenPasswordBody | TokenClientCredentialsBody,
+    Field(discriminator="grant_type"),
+]
+
+
+class TokenBody(RootModel[TokenUnion]):
+    """Token request body."""
+
+    @model_validator(mode="before")
+    @classmethod
+    def default_grant_type(cls, data):
+        """Add default grant_type for discrimination."""
+        if "grant_type" not in data:
+            data["grant_type"] = "password"
+        return data

providers/keycloak/src/airflow/providers/keycloak/auth_manager/keycloak_auth_manager.py

@@ -147,6 +147,13 @@ def get_url_logout(self) -> str | None:
         return urljoin(base_url, f"{AUTH_MANAGER_FASTAPI_APP_PREFIX}/logout")
 
     def refresh_user(self, *, user: KeycloakAuthManagerUser) -> KeycloakAuthManagerUser | None:
+        # According to RFC6749 section 4.4.3, a refresh token should not be included when using
+        # the Service accounts/client_credentials flow.
+        # We check whether the user has a refresh token; if not, we assume it's a service account
+        # and return None.
+        if not user.refresh_token:
+            return None
+
         if self._token_expired(user.access_token):
             tokens = self.refresh_tokens(user=user)
 
@@ -158,6 +165,10 @@ def refresh_user(self, *, user: KeycloakAuthManagerUser) -> KeycloakAuthManagerU
         return None
 
     def refresh_tokens(self, *, user: KeycloakAuthManagerUser) -> dict[str, str]:
+        if not user.refresh_token:
+            # It is a service account. It used the client credentials flow and no refresh token is issued.
+            return {}
+
         try:
             log.debug("Refreshing the token")
             client = self.get_keycloak_client()
@@ -316,9 +327,22 @@ def get_cli_commands() -> list[CLICommand]:
         ]
 
     @staticmethod
-    def get_keycloak_client() -> KeycloakOpenID:
-        client_id = conf.get(CONF_SECTION_NAME, CONF_CLIENT_ID_KEY)
-        client_secret = conf.get(CONF_SECTION_NAME, CONF_CLIENT_SECRET_KEY)
+    def get_keycloak_client(client_id: str | None = None, client_secret: str | None = None) -> KeycloakOpenID:
+        """
+        Get a KeycloakOpenID client instance.
+
+        :param client_id: Optional client ID to override config. If provided, client_secret must also be provided.
+        :param client_secret: Optional client secret to override config. If provided, client_id must also be provided.
+        """
+        if (client_id is None) != (client_secret is None):
+            raise ValueError(
+                "Both `client_id` and `client_secret` must be provided together, or both must be None"
+            )
+
+        if client_id is None:
+            client_id = conf.get(CONF_SECTION_NAME, CONF_CLIENT_ID_KEY)
+            client_secret = conf.get(CONF_SECTION_NAME, CONF_CLIENT_SECRET_KEY)
+
         realm = conf.get(CONF_SECTION_NAME, CONF_REALM_KEY)
         server_url = conf.get(CONF_SECTION_NAME, CONF_SERVER_URL_KEY)
 

providers/keycloak/src/airflow/providers/keycloak/auth_manager/openapi/v2-keycloak-auth-manager-generated.yaml

@@ -128,7 +128,7 @@ paths:
         content:
           application/json:
             schema:
-              $ref: '#/components/schemas/TokenBody'
+              $ref: '#/components/schemas/TokenPasswordBody'
         required: true
       responses:
         '201':
@@ -180,7 +180,43 @@ components:
       type: object
       title: HTTPValidationError
     TokenBody:
+      oneOf:
+      - $ref: '#/components/schemas/TokenPasswordBody'
+      - $ref: '#/components/schemas/TokenClientCredentialsBody'
+      title: TokenBody
+      description: Token request body.
+      discriminator:
+        propertyName: grant_type
+        mapping:
+          client_credentials: '#/components/schemas/TokenClientCredentialsBody'
+          password: '#/components/schemas/TokenPasswordBody'
+    TokenClientCredentialsBody:
       properties:
+        grant_type:
+          type: string
+          const: client_credentials
+          title: Grant Type
+        client_id:
+          type: string
+          title: Client Id
+        client_secret:
+          type: string
+          title: Client Secret
+      additionalProperties: false
+      type: object
+      required:
+      - grant_type
+      - client_id
+      - client_secret
+      title: TokenClientCredentialsBody
+      description: Client credentials grant token serializer for post bodies.
+    TokenPasswordBody:
+      properties:
+        grant_type:
+          type: string
+          const: password
+          title: Grant Type
+          default: password
         username:
           type: string
           title: Username
@@ -192,8 +228,8 @@ components:
       required:
       - username
       - password
-      title: TokenBody
-      description: Token serializer for post bodies.
+      title: TokenPasswordBody
+      description: Password grant token serializer for post bodies.
     TokenResponse:
       properties:
         access_token:

providers/keycloak/src/airflow/providers/keycloak/auth_manager/routes/token.py

@@ -24,8 +24,11 @@
 from airflow.api_fastapi.common.router import AirflowRouter
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
 from airflow.providers.common.compat.sdk import conf
-from airflow.providers.keycloak.auth_manager.datamodels.token import TokenBody, TokenResponse
-from airflow.providers.keycloak.auth_manager.services.token import create_token_for
+from airflow.providers.keycloak.auth_manager.datamodels.token import (
+    TokenBody,
+    TokenPasswordBody,
+    TokenResponse,
+)
 
 log = logging.getLogger(__name__)
 token_router = AirflowRouter(tags=["KeycloakAuthManagerToken"])
@@ -37,7 +40,9 @@
     responses=create_openapi_http_exception_doc([status.HTTP_400_BAD_REQUEST, status.HTTP_401_UNAUTHORIZED]),
 )
 def create_token(body: TokenBody) -> TokenResponse:
-    token = create_token_for(body.username, body.password)
+    token = body.root.create_token(
+        expiration_time_in_seconds=int(conf.getint("api_auth", "jwt_expiration_time"))
+    )
     return TokenResponse(access_token=token)
 
 
@@ -46,10 +51,8 @@ def create_token(body: TokenBody) -> TokenResponse:
     status_code=status.HTTP_201_CREATED,
     responses=create_openapi_http_exception_doc([status.HTTP_400_BAD_REQUEST, status.HTTP_401_UNAUTHORIZED]),
 )
-def create_token_cli(body: TokenBody) -> TokenResponse:
-    token = create_token_for(
-        body.username,
-        body.password,
-        expiration_time_in_seconds=int(conf.getint("api_auth", "jwt_cli_expiration_time")),
+def create_token_cli(body: TokenPasswordBody) -> TokenResponse:
+    token = body.create_token(
+        expiration_time_in_seconds=int(conf.getint("api_auth", "jwt_cli_expiration_time"))
     )
     return TokenResponse(access_token=token)

providers/keycloak/src/airflow/providers/keycloak/auth_manager/services/token.py

@@ -51,3 +51,47 @@ def create_token_for(
     )
 
     return get_auth_manager().generate_jwt(user, expiration_time_in_seconds=expiration_time_in_seconds)
+
+
+def create_client_credentials_token(
+    client_id: str,
+    client_secret: str,
+    expiration_time_in_seconds: int = conf.getint("api_auth", "jwt_expiration_time"),
+) -> str:
+    """
+    Create token using OAuth2 client_credentials grant type.
+
+    This authentication flow uses the provided client_id and client_secret
+    to obtain a token for a service account. The Keycloak client must have:
+    - Service accounts roles: ON
+    - Client Authentication: ON (confidential client)
+
+    The service account must be configured with the appropriate roles/permissions.
+    """
+    # Get Keycloak client with service account credentials
+    client = KeycloakAuthManager.get_keycloak_client(
+        client_id=client_id,
+        client_secret=client_secret,
+    )
+
+    try:
+        tokens = client.token(grant_type="client_credentials")
+    except KeycloakAuthenticationError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Client credentials authentication failed",
+        )
+
+    # For client_credentials, get the service account user info
+    # The token represents the service account associated with the client
+    userinfo = client.userinfo(tokens["access_token"])
+    user = KeycloakAuthManagerUser(
+        user_id=userinfo["sub"],
+        name=userinfo.get("preferred_username", userinfo.get("clientId", "service-account")),
+        access_token=tokens["access_token"],
+        refresh_token=tokens.get(
+            "refresh_token"
+        ),  # client_credentials may not return refresh_token (RFC6749 section 4.4.3)
+    )
+
+    return get_auth_manager().generate_jwt(user, expiration_time_in_seconds=expiration_time_in_seconds)

providers/keycloak/src/airflow/providers/keycloak/auth_manager/user.py

@@ -22,7 +22,7 @@
 class KeycloakAuthManagerUser(BaseUser):
     """User model for users managed by Keycloak auth manager."""
 
-    def __init__(self, *, user_id: str, name: str, access_token: str, refresh_token: str) -> None:
+    def __init__(self, *, user_id: str, name: str, access_token: str, refresh_token: str | None) -> None:
         self.user_id = user_id
         self.name = name
         self.access_token = access_token