Third-Party System Accessing OpenShift API via Keycloak SSO — Solution Document

{
  "realm": "demo",
  "enabled": true,
  "registrationAllowed": false,
  "loginWithEmailAllowed": true,
  "sslRequired": "none",
  "accessTokenLifespan": 1800
}

{
  "clientId": "ocp-oauth",
  "name": "OpenShift OAuth",
  "enabled": true,
  "protocol": "openid-connect",
  "publicClient": false,
  "clientAuthenticatorType": "client-secret",
  "secret": "<ocp-oauth-secret>",
  "standardFlowEnabled": true,
  "directAccessGrantsEnabled": false,
  "serviceAccountsEnabled": true,
  "redirectUris": [
    "https://oauth-openshift.apps.<cluster-domain>/oauth2callback/keycloak"
  ],
  "webOrigins": ["+"],
  "attributes": {
    "token.exchange.standard.enabled": "true"
  }
}

{
  "clientId": "third-party-app",
  "name": "Third-Party Demo App",
  "enabled": true,
  "protocol": "openid-connect",
  "publicClient": false,
  "clientAuthenticatorType": "client-secret",
  "secret": "<third-party-app-secret>",
  "standardFlowEnabled": true,
  "directAccessGrantsEnabled": true,
  "redirectUris": [
    "http://<bastion-ip>:9080/callback/keycloak",
    "http://localhost:8080/callback/keycloak"
  ],
  "webOrigins": ["+"],
  "attributes": {
    "token.exchange.standard.enabled": "true",
    "post.logout.redirect.uris": "http://<bastion-ip>:9080/*##http://localhost:8080/*"
  }
}

{
  "access_token": "<new-token-with-azp=ocp-oauth>",
  "issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
  "token_type": "Bearer",
  "expires_in": 1800
}

apiVersion: config.openshift.io/v1
kind: OAuth
metadata:
  name: cluster
spec:
  identityProviders:
    - name: keycloak
      mappingMethod: claim
      type: OpenID
      openID:
        clientID: ocp-oauth
        clientSecret:
          name: keycloak-client-secret   # Secret in openshift-config namespace
        ca:
          name: keycloak-ca              # ConfigMap with Keycloak CA cert
        issuer: https://<bastion-ip>:18443/realms/demo
        claims:
          preferredUsername:
            - preferred_username
          name:
            - name
          email:
            - email

# Create client secret
oc create secret generic keycloak-client-secret \
  --from-literal=clientSecret=<ocp-oauth-secret> \
  -n openshift-config

# Create CA ConfigMap (self-signed certificate scenario)
oc create configmap keycloak-ca \
  --from-file=ca.crt=certs/tls.crt \
  -n openshift-config

apiVersion: oauth.openshift.io/v1
kind: OAuthClient
metadata:
  name: third-party-demo
grantMethod: auto
secret: <ocp-oauthclient-secret>
redirectURIs:
  - http://<bastion-ip>:9080/callback/ocp
  - http://localhost:8080/callback/ocp

# Create Projects
oc new-project project-a
oc new-project project-b
oc new-project project-c

# user-a (Alice): view → project-a, project-c
oc adm policy add-role-to-user view user-a -n project-a
oc adm policy add-role-to-user view user-a -n project-c

# user-b (Bob): edit → project-b
oc adm policy add-role-to-user edit user-b -n project-b

# user-c (Charlie): admin → project-c, view → project-a, project-b
oc adm policy add-role-to-user admin user-c -n project-c
oc adm policy add-role-to-user view  user-c -n project-a
oc adm policy add-role-to-user view  user-c -n project-b

# Step 2a: Token Exchange (RFC 8693)
resp = requests.post(KC_TOKEN_URL, data={
    "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
    "subject_token": kc_access_token,       # User's KC Token
    "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
    "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
    "client_id": "ocp-oauth",               # Target Client ID
    "client_secret": "<ocp-oauth-secret>",   # Target Client Secret
}, verify=TLS_VERIFY)

# Step 2b: Redirect to OCP OAuth, specifying keycloak IdP
params = {
    "client_id": "third-party-demo",
    "idp": "keycloak",           # Key: skip IdP selection page
    "response_type": "code",
    "redirect_uri": APP_URL + "/callback/ocp",
    "state": state,
}
redirect(f"{OCP_OAUTH_URL}/oauth/authorize?{urlencode(params)}")

@app.route("/logout")
def logout():
    session.clear()
    params = {
        "client_id": "third-party-app",
        "post_logout_redirect_uri": APP_URL + "/",
    }
    return redirect(f"{KC_LOGOUT_URL}?{urlencode(params)}")

FROM registry.access.redhat.com/ubi9/python-312:latest
WORKDIR /opt/app-root/src
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8080
CMD ["gunicorn", "--bind", "0.0.0.0:8080", "--workers", "2", "--timeout", "120", "app:app"]

bash baremetal/deploy-all.sh all

bash deploy-all.sh status

"""
Third-Party Demo App — Keycloak SSO + OpenShift API Integration
Demonstrates Plan B: Keycloak Token Exchange (RFC 8693) + OCP OAuth bridge

Flow:
  1. User logs in via Keycloak OIDC
  2. User clicks "Call OCP API" — server-side chain:
     a. Token Exchange (RFC 8693): swap KC token for ocp-oauth audience
     b. OCP OAuth: redirect to OCP → Keycloak SSO auto-completes → get OCP token
     c. OCP API call: GET /apis/project.openshift.io/v1/projects
  3. Result page shows all 3 sub-steps completed + project list
"""

import os
import json
import secrets
import logging
from urllib.parse import urlencode

import requests
from flask import Flask, redirect, request, session, render_template, url_for, jsonify

app = Flask(__name__)
app.secret_key = os.environ.get("FLASK_SECRET_KEY", secrets.token_hex(32))

logging.basicConfig(level=logging.DEBUG)
log = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Configuration (from environment variables)
# ---------------------------------------------------------------------------

# Keycloak
KC_BASE        = os.environ["KC_BASE_URL"]
KC_REALM       = os.environ.get("KC_REALM", "demo")
KC_CLIENT_ID   = os.environ.get("KC_CLIENT_ID", "third-party-app")
KC_CLIENT_SECRET = os.environ["KC_CLIENT_SECRET"]

# OpenShift
OCP_API_URL    = os.environ["OCP_API_URL"]
OCP_OAUTH_URL  = os.environ.get("OCP_OAUTH_URL", "")
OCP_OAUTH_CLIENT_ID = os.environ.get("OCP_OAUTH_CLIENT_ID", "third-party-demo")
OCP_OAUTH_CLIENT_SECRET = os.environ.get("OCP_OAUTH_CLIENT_SECRET", "")
OCP_IDP_NAME   = os.environ.get("OCP_IDP_NAME", "keycloak")

# Token Exchange target audience (the Keycloak client used by OCP OAuth)
KC_OCP_CLIENT_ID = os.environ.get("KC_OCP_CLIENT_ID", "ocp-oauth")
KC_OCP_CLIENT_SECRET = os.environ.get("KC_OCP_CLIENT_SECRET", "")

# Derived URLs
KC_OPENID_BASE = f"{KC_BASE}/realms/{KC_REALM}/protocol/openid-connect"
KC_AUTH_URL     = f"{KC_OPENID_BASE}/auth"
KC_TOKEN_URL    = f"{KC_OPENID_BASE}/token"
KC_USERINFO_URL = f"{KC_OPENID_BASE}/userinfo"
KC_LOGOUT_URL   = f"{KC_OPENID_BASE}/logout"

# TLS verification (set to "false" for self-signed certs in PoC)
TLS_VERIFY = os.environ.get("TLS_VERIFY", "true").lower() != "false"

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _app_url(path: str) -> str:
    """Build absolute callback URL for this app."""
    base = os.environ.get("APP_EXTERNAL_URL", request.host_url.rstrip("/"))
    return f"{base}{path}"


def _decode_jwt_payload(token: str) -> dict:
    """Decode JWT payload without signature verification (display only)."""
    import base64
    parts = token.split(".")
    if len(parts) < 2:
        return {}
    payload = parts[1]
    payload += "=" * (4 - len(payload) % 4)
    try:
        return json.loads(base64.urlsafe_b64decode(payload))
    except Exception:
        return {}

# ---------------------------------------------------------------------------
# Routes — Keycloak OIDC Login
# ---------------------------------------------------------------------------

@app.route("/")
def index():
    """Landing page."""
    return render_template("index.html",
                           user=session.get("user"),
                           kc_token_info=session.get("kc_token_info"),
                           exchanged_token_info=session.get("exchanged_token_info"),
                           ocp_token=session.get("ocp_token"),
                           ocp_projects=session.get("ocp_projects"),
                           ocp_user=session.get("ocp_user"),
                           ocp_api_called=session.get("ocp_api_called"),
                           error=session.pop("error", None))


@app.route("/login")
def login():
    """Step 1: Redirect to Keycloak for OIDC login."""
    state = secrets.token_urlsafe(32)
    session["oauth_state"] = state
    params = {
        "client_id": KC_CLIENT_ID,
        "response_type": "code",
        "scope": "openid profile email",
        "redirect_uri": _app_url("/callback/keycloak"),
        "state": state,
    }
    return redirect(f"{KC_AUTH_URL}?{urlencode(params)}")


@app.route("/callback/keycloak")
def callback_keycloak():
    """Keycloak OIDC callback — exchange code for tokens."""
    if request.args.get("state") != session.pop("oauth_state", None):
        session["error"] = "Invalid OAuth state"
        return redirect(url_for("index"))

    code = request.args.get("code")
    if not code:
        session["error"] = f"No authorization code. Error: {request.args.get('error_description', 'unknown')}"
        return redirect(url_for("index"))

    resp = requests.post(KC_TOKEN_URL, data={
        "grant_type": "authorization_code",
        "client_id": KC_CLIENT_ID,
        "client_secret": KC_CLIENT_SECRET,
        "code": code,
        "redirect_uri": _app_url("/callback/keycloak"),
    }, verify=TLS_VERIFY)

    if resp.status_code != 200:
        session["error"] = f"Token exchange failed: {resp.text}"
        return redirect(url_for("index"))

    tokens = resp.json()
    access_token = tokens["access_token"]
    token_payload = _decode_jwt_payload(access_token)

    session["kc_access_token"] = access_token
    session["user"] = token_payload.get("preferred_username", token_payload.get("sub", "unknown"))
    session["kc_token_info"] = {
        "sub": token_payload.get("sub"),
        "preferred_username": token_payload.get("preferred_username"),
        "email": token_payload.get("email"),
        "aud": token_payload.get("aud"),
        "azp": token_payload.get("azp"),
        "scope": token_payload.get("scope"),
        "exp": token_payload.get("exp"),
        "iss": token_payload.get("iss"),
    }

    # Clear previous OCP data
    for key in ("exchanged_token_info", "ocp_token", "ocp_projects", "ocp_user", "ocp_api_called"):
        session.pop(key, None)

    log.info("User %s logged in via Keycloak", session["user"])
    return redirect(url_for("index"))

# ---------------------------------------------------------------------------
# Routes — Combined "Call OCP API" (Token Exchange + OCP OAuth + API call)
# ---------------------------------------------------------------------------

@app.route("/call-ocp-api")
def call_ocp_api():
    """Single button: Token Exchange → OCP OAuth → API call."""
    kc_token = session.get("kc_access_token")
    if not kc_token:
        session["error"] = "Not logged in. Please log in via Keycloak SSO first."
        return redirect(url_for("index"))

    # --- Sub-step a: Token Exchange (RFC 8693) ---
    resp = requests.post(KC_TOKEN_URL, data={
        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
        "subject_token": kc_token,
        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
        "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
        "client_id": KC_OCP_CLIENT_ID,
        "client_secret": KC_OCP_CLIENT_SECRET,
    }, verify=TLS_VERIFY)

    if resp.status_code != 200:
        log.error("Token exchange failed: %s %s", resp.status_code, resp.text)
        session["error"] = f"Token Exchange failed ({resp.status_code}): {resp.text}"
        return redirect(url_for("index"))

    exchanged = resp.json()
    exchanged_token = exchanged["access_token"]
    payload = _decode_jwt_payload(exchanged_token)

    session["exchanged_token_info"] = {
        "sub": payload.get("sub"),
        "preferred_username": payload.get("preferred_username"),
        "aud": payload.get("aud"),
        "azp": payload.get("azp"),
        "scope": payload.get("scope"),
        "iss": payload.get("iss"),
        "token_type": exchanged.get("token_type"),
        "issued_token_type": exchanged.get("issued_token_type"),
    }

    # Clear kc_access_token — no longer needed, saves cookie space
    session.pop("kc_access_token", None)

    log.info("Token exchanged for audience=%s", KC_OCP_CLIENT_ID)

    # --- Sub-step b: Redirect to OCP OAuth ---
    if not OCP_OAUTH_URL:
        session["error"] = "OCP_OAUTH_URL not configured"
        return redirect(url_for("index"))

    state = secrets.token_urlsafe(32)
    session["ocp_oauth_state"] = state
    params = {
        "client_id": OCP_OAUTH_CLIENT_ID,
        "idp": OCP_IDP_NAME,
        "response_type": "code",
        "redirect_uri": _app_url("/callback/ocp"),
        "state": state,
    }
    authorize_url = f"{OCP_OAUTH_URL}/oauth/authorize?{urlencode(params)}"
    return redirect(authorize_url)


@app.route("/callback/ocp")
def callback_ocp():
    """OCP OAuth callback — exchange code for OCP token, then call OCP API."""
    if request.args.get("state") != session.pop("ocp_oauth_state", None):
        session["error"] = "Invalid OCP OAuth state"
        return redirect(url_for("index"))

    code = request.args.get("code")
    if not code:
        session["error"] = f"No OCP authorization code. Error: {request.args.get('error_description', 'unknown')}"
        return redirect(url_for("index"))

    # Exchange code for OCP token
    resp = requests.post(f"{OCP_OAUTH_URL}/oauth/token", data={
        "grant_type": "authorization_code",
        "client_id": OCP_OAUTH_CLIENT_ID,
        "client_secret": OCP_OAUTH_CLIENT_SECRET,
        "code": code,
        "redirect_uri": _app_url("/callback/ocp"),
    }, verify=TLS_VERIFY)

    if resp.status_code != 200:
        log.error("OCP token exchange failed: %s %s", resp.status_code, resp.text)
        session["error"] = f"OCP token exchange failed ({resp.status_code}): {resp.text}"
        return redirect(url_for("index"))

    ocp_tokens = resp.json()
    ocp_token = ocp_tokens.get("access_token", "")
    session["ocp_token"] = ocp_token[:40]  # Store truncated for display only
    log.info("Got OCP token: %s...", ocp_token[:20] if ocp_token else "empty")

    # --- Sub-step c: Call OCP API ---
    _fetch_ocp_user(ocp_token)
    _fetch_ocp_projects(ocp_token)

    # Record which API was called
    session["ocp_api_called"] = {
        "method": "GET",
        "url": f"{OCP_API_URL}/apis/project.openshift.io/v1/projects",
        "description": "List OpenShift Projects the current user has access to",
        "user_api_url": f"{OCP_API_URL}/apis/user.openshift.io/v1/users/~",
        "user_api_description": "Get current OCP user identity",
    }

    # Clear large tokens to keep session cookie under 4KB
    session.pop("kc_access_token", None)
    session.pop("exchanged_access_token", None)

    return redirect(url_for("index"))


def _fetch_ocp_user(token: str):
    """Get current OCP user identity."""
    try:
        resp = requests.get(
            f"{OCP_API_URL}/apis/user.openshift.io/v1/users/~",
            headers={"Authorization": f"Bearer {token}"},
            verify=TLS_VERIFY,
        )
        if resp.status_code == 200:
            user_data = resp.json()
            session["ocp_user"] = {
                "name": user_data.get("metadata", {}).get("name", ""),
                "uid": user_data.get("metadata", {}).get("uid", ""),
                "fullName": user_data.get("fullName", ""),
                "identities": user_data.get("identities", []),
                "groups": user_data.get("groups", []),
            }
        else:
            log.warning("Failed to fetch OCP user: %s %s", resp.status_code, resp.text)
    except Exception as e:
        log.error("Error fetching OCP user: %s", e)


def _fetch_ocp_projects(token: str):
    """Get OCP projects the user has access to."""
    try:
        resp = requests.get(
            f"{OCP_API_URL}/apis/project.openshift.io/v1/projects",
            headers={"Authorization": f"Bearer {token}"},
            verify=TLS_VERIFY,
        )
        if resp.status_code == 200:
            data = resp.json()
            projects = []
            for item in data.get("items", []):
                projects.append({
                    "name": item["metadata"]["name"],
                    "status": item.get("status", {}).get("phase", ""),
                    "display_name": item["metadata"].get("annotations", {}).get(
                        "openshift.io/display-name", ""),
                })
            session["ocp_projects"] = projects
        else:
            log.warning("Failed to fetch projects: %s %s", resp.status_code, resp.text)
            session["ocp_projects"] = []
    except Exception as e:
        log.error("Error fetching projects: %s", e)
        session["ocp_projects"] = []

# ---------------------------------------------------------------------------
# Routes — Logout & API
# ---------------------------------------------------------------------------

@app.route("/logout")
def logout():
    """Logout from the app and Keycloak (clear all sessions)."""
    session.clear()
    params = {
        "client_id": KC_CLIENT_ID,
        "post_logout_redirect_uri": _app_url("/"),
    }
    return redirect(f"{KC_LOGOUT_URL}?{urlencode(params)}")


@app.route("/health")
def health():
    return jsonify({"status": "ok"})


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    port = int(os.environ.get("PORT", "8080"))
    app.run(host="0.0.0.0", port=port, debug=True)

# Verify Keycloak startup parameters include --features=token-exchange
podman inspect keycloak | jq '.[0].Config.Cmd'

# Verify client attributes
curl -sk -H "Authorization: Bearer $TOKEN" \
  "$KC_URL/admin/realms/demo/clients?clientId=ocp-oauth" | jq '.[0].attributes'

{
  "attributes": {
    "post.logout.redirect.uris": "http://bastion:9080/*##http://localhost:8080/*"
  }
}