OpenID Connect Identity Access Management — ระบบยืนยันตัวตนและจัดการสิทธิ์

OpenID Connect คืออะไรและทำงานอย่างไร

OpenID Connect (OIDC) เป็น identity layer ที่สร้างบน OAuth 2.0 protocol ออกแบบมาสำหรับ authentication (ยืนยันตัวตน) ในขณะที่ OAuth 2.0 เน้นเรื่อง authorization (การอนุญาต) OIDC เพิ่ม ID Token ที่เป็น JWT ซึ่งบรรจุข้อมูลของผู้ใช้เช่น ชื่อ อีเมล และ roles

OIDC Flow หลักมี 3 แบบคือ Authorization Code Flow ที่เหมาะสำหรับ server-side applications เป็น flow ที่ปลอดภัยที่สุด, Implicit Flow ที่เหมาะสำหรับ SPA แต่ปัจจุบันแนะนำให้ใช้ Authorization Code Flow with PKCE แทน และ Client Credentials Flow สำหรับ machine-to-machine communication

ส่วนประกอบหลักของ OIDC ได้แก่ OpenID Provider (OP) ที่เป็น identity server เช่น Keycloak, Auth0, Okta, Relying Party (RP) ที่เป็น application ที่ต้องการยืนยันตัวตนผู้ใช้, ID Token ที่เป็น JWT บรรจุข้อมูลผู้ใช้, Access Token สำหรับเข้าถึง API, Refresh Token สำหรับขอ token ใหม่ และ UserInfo Endpoint สำหรับดึงข้อมูลผู้ใช้เพิ่มเติม

OIDC Discovery Document อยู่ที่ URL .well-known/openid-configuration ซึ่งบอก endpoints ทั้งหมดที่ client ต้องใช้ ทำให้การตั้งค่าง่ายขึ้นมาก client แค่รู้ issuer URL ก็สามารถค้นหา endpoints ทั้งหมดได้อัตโนมัติ

ตั้งค่า Keycloak เป็น OpenID Connect Provider

ติดตั้ง Keycloak ด้วย Docker Compose พร้อม PostgreSQL

# docker-compose.yml — Keycloak with PostgreSQL
version: '3.8'

services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: keycloak
      POSTGRES_USER: keycloak
      POSTGRES_PASSWORD: keycloak_db_pass
    volumes:
      - pgdata:/var/lib/postgresql/data
    networks:
      - keycloak-net
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U keycloak"]
      interval: 10s
      timeout: 5s
      retries: 5

  keycloak:
    image: quay.io/keycloak/keycloak:24.0
    command: start
    environment:
      KC_DB: postgres
      KC_DB_URL: jdbc:postgresql://postgres:5432/keycloak
      KC_DB_USERNAME: keycloak
      KC_DB_PASSWORD: keycloak_db_pass
      KC_HOSTNAME: auth.example.com
      KC_HOSTNAME_STRICT: false
      KC_HTTP_ENABLED: true
      KC_PROXY_HEADERS: xforwarded
      KEYCLOAK_ADMIN: admin
      KEYCLOAK_ADMIN_PASSWORD: AdminPass123!
    ports:
      - "8080:8080"
    depends_on:
      postgres:
        condition: service_healthy
    networks:
      - keycloak-net

volumes:
  pgdata:

networks:
  keycloak-net:

# docker compose up -d

# ตั้งค่า Realm และ Client ผ่าน Keycloak Admin CLI
docker exec keycloak /opt/keycloak/bin/kcadm.sh config credentials \
  --server http://localhost:8080 \
  --realm master \
  --user admin \
  --password AdminPass123!

# สร้าง Realm
docker exec keycloak /opt/keycloak/bin/kcadm.sh create realms \
  -s realm=myapp \
  -s enabled=true \
  -s displayName="My Application"

# สร้าง Client (Authorization Code Flow)
docker exec keycloak /opt/keycloak/bin/kcadm.sh create clients \
  -r myapp \
  -s clientId=web-app \
  -s enabled=true \
  -s publicClient=false \
  -s secret=web-app-secret-123 \
  -s 'redirectUris=["https://app.example.com/callback","http://localhost:3000/callback"]' \
  -s 'webOrigins=["https://app.example.com","http://localhost:3000"]' \
  -s directAccessGrantsEnabled=true \
  -s standardFlowEnabled=true

# สร้าง Roles
docker exec keycloak /opt/keycloak/bin/kcadm.sh create roles -r myapp -s name=admin
docker exec keycloak /opt/keycloak/bin/kcadm.sh create roles -r myapp -s name=editor
docker exec keycloak /opt/keycloak/bin/kcadm.sh create roles -r myapp -s name=viewer

# สร้าง User
docker exec keycloak /opt/keycloak/bin/kcadm.sh create users -r myapp \
  -s username=john \
  -s email=john@example.com \
  -s firstName=John \
  -s lastName=Doe \
  -s enabled=true

docker exec keycloak /opt/keycloak/bin/kcadm.sh set-password -r myapp \
  --username john --new-password JohnPass123!

เชื่อมต่อ Application กับ OIDC Provider

เชื่อมต่อ Node.js Application กับ Keycloak ผ่าน OIDC

// app.js — Node.js Express with OpenID Connect
const express = require('express');
const session = require('express-session');
const { Issuer, Strategy } = require('openid-client');
const passport = require('passport');

const app = express();

// Session
app.use(session({
  secret: 'session-secret-key',
  resave: false,
  saveUninitialized: false,
  cookie: { secure: process.env.NODE_ENV === 'production' }
}));

app.use(passport.initialize());
app.use(passport.session());

passport.serializeUser((user, done) => done(null, user));
passport.deserializeUser((user, done) => done(null, user));

async function setupOIDC() {
  // Discover OIDC configuration
  const keycloakIssuer = await Issuer.discover(
    'http://localhost:8080/realms/myapp'
  );
  console.log('Discovered issuer:', keycloakIssuer.issuer);

  const client = new keycloakIssuer.Client({
    client_id: 'web-app',
    client_secret: 'web-app-secret-123',
    redirect_uris: ['http://localhost:3000/callback'],
    response_types: ['code'],
  });

  passport.use('oidc', new Strategy({ client, params: {
    scope: 'openid email profile roles',
  }}, (tokenSet, userinfo, done) => {
    const user = {
      id: userinfo.sub,
      email: userinfo.email,
      name: userinfo.name,
      roles: tokenSet.claims().realm_access?.roles || [],
      accessToken: tokenSet.access_token,
      idToken: tokenSet.id_token,
    };
    return done(null, user);
  }));

  return client;
}

// Routes
app.get('/login', passport.authenticate('oidc'));

app.get('/callback',
  passport.authenticate('oidc', { failureRedirect: '/error' }),
  (req, res) => res.redirect('/profile')
);

app.get('/profile', ensureAuth, (req, res) => {
  res.json({
    user: req.user.name,
    email: req.user.email,
    roles: req.user.roles,
  });
});

app.get('/admin', ensureAuth, ensureRole('admin'), (req, res) => {
  res.json({ message: 'Admin area', user: req.user.name });
});

app.get('/logout', (req, res) => {
  req.logout(() => {
    res.redirect(`http://localhost:8080/realms/myapp/protocol/openid-connect/logout?redirect_uri=http://localhost:3000`);
  });
});

function ensureAuth(req, res, next) {
  if (req.isAuthenticated()) return next();
  res.status(401).json({ error: 'Unauthorized' });
}

function ensureRole(role) {
  return (req, res, next) => {
    if (req.user.roles.includes(role)) return next();
    res.status(403).json({ error: 'Forbidden' });
  };
}

setupOIDC().then(() => {
  app.listen(3000, () => console.log('Server on http://localhost:3000'));
});

// package.json dependencies:
// "express": "^4.18",
// "express-session": "^1.17",
// "openid-client": "^5.6",
// "passport": "^0.7"

จัดการ Roles และ Permissions ด้วย OIDC Claims

ตั้งค่า fine-grained access control ด้วย OIDC claims และ Keycloak roles

#!/usr/bin/env python3
# oidc_rbac.py — Role-Based Access Control with OIDC
import jwt
import requests
from functools import wraps
from flask import Flask, request, jsonify, g

app = Flask(__name__)

OIDC_ISSUER = "http://localhost:8080/realms/myapp"
OIDC_JWKS_URI = f"{OIDC_ISSUER}/protocol/openid-connect/certs"

# Cache JWKS keys
_jwks_client = jwt.PyJWKClient(OIDC_JWKS_URI)

PERMISSIONS = {
    "admin": ["read", "write", "delete", "manage_users"],
    "editor": ["read", "write"],
    "viewer": ["read"],
}

def decode_token(token):
    signing_key = _jwks_client.get_signing_key_from_jwt(token)
    return jwt.decode(
        token,
        signing_key.key,
        algorithms=["RS256"],
        audience="web-app",
        issuer=OIDC_ISSUER,
    )

def require_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        auth_header = request.headers.get("Authorization", "")
        if not auth_header.startswith("Bearer "):
            return jsonify({"error": "Missing token"}), 401
        
        token = auth_header.split(" ")[1]
        try:
            claims = decode_token(token)
            g.user = {
                "sub": claims["sub"],
                "email": claims.get("email"),
                "name": claims.get("name"),
                "roles": claims.get("realm_access", {}).get("roles", []),
            }
        except jwt.ExpiredSignatureError:
            return jsonify({"error": "Token expired"}), 401
        except jwt.InvalidTokenError as e:
            return jsonify({"error": f"Invalid token: {str(e)}"}), 401
        
        return f(*args, **kwargs)
    return decorated

def require_role(*roles):
    def decorator(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            user_roles = g.user.get("roles", [])
            if not any(r in user_roles for r in roles):
                return jsonify({"error": "Insufficient permissions"}), 403
            return f(*args, **kwargs)
        return decorated
    return decorator

def require_permission(permission):
    def decorator(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            user_roles = g.user.get("roles", [])
            user_permissions = set()
            for role in user_roles:
                user_permissions.update(PERMISSIONS.get(role, []))
            
            if permission not in user_permissions:
                return jsonify({"error": f"Missing permission: {permission}"}), 403
            return f(*args, **kwargs)
        return decorated
    return decorator

@app.route("/api/profile")
@require_auth
def profile():
    return jsonify(g.user)

@app.route("/api/articles", methods=["GET"])
@require_auth
@require_permission("read")
def list_articles():
    return jsonify({"articles": [{"id": 1, "title": "Hello"}]})

@app.route("/api/articles", methods=["POST"])
@require_auth
@require_permission("write")
def create_article():
    return jsonify({"message": "Article created"}), 201

@app.route("/api/articles/", methods=["DELETE"])
@require_auth
@require_permission("delete")
def delete_article(id):
    return jsonify({"message": f"Article {id} deleted"})

@app.route("/api/users")
@require_auth
@require_role("admin")
def manage_users():
    return jsonify({"users": [{"id": "1", "name": "John"}]})

if __name__ == "__main__":
    app.run(port=5000, debug=True)
# pip install flask PyJWT cryptography requests

ตั้งค่า Multi-Factor Authentication และ Security

เปิดใช้ MFA และตั้งค่า security policies บน Keycloak

# ตั้งค่า MFA (TOTP) ผ่าน Keycloak Admin CLI

# เปิด OTP Policy
docker exec keycloak /opt/keycloak/bin/kcadm.sh update realms/myapp \
  -s 'otpPolicyType=totp' \
  -s 'otpPolicyAlgorithm=HmacSHA256' \
  -s 'otpPolicyDigits=6' \
  -s 'otpPolicyPeriod=30' \
  -s 'otpPolicyInitialCounter=0'

# สร้าง Authentication Flow ที่บังคับ MFA
# ใน Keycloak Admin Console:
# Authentication > Flows > Browser > Copy
# เพิ่ม "OTP Form" หลัง "Username Password Form"
# ตั้งเป็น Required

# ตั้งค่า Password Policy
docker exec keycloak /opt/keycloak/bin/kcadm.sh update realms/myapp \
  -s 'passwordPolicy="length(12) and digits(1) and upperCase(1) and lowerCase(1) and specialChars(1) and notUsername and passwordHistory(5)"'

# ตั้งค่า Brute Force Protection
docker exec keycloak /opt/keycloak/bin/kcadm.sh update realms/myapp \
  -s bruteForceProtected=true \
  -s permanentLockout=false \
  -s maxFailureWaitSeconds=900 \
  -s minimumQuickLoginWaitSeconds=60 \
  -s waitIncrementSeconds=60 \
  -s quickLoginCheckMilliSeconds=1000 \
  -s maxDeltaTimeSeconds=43200 \
  -s failureFactor=5

# ตั้งค่า Token Lifetimes
docker exec keycloak /opt/keycloak/bin/kcadm.sh update realms/myapp \
  -s 'accessTokenLifespan=300' \
  -s 'ssoSessionIdleTimeout=1800' \
  -s 'ssoSessionMaxLifespan=36000' \
  -s 'accessTokenLifespanForImplicitFlow=900'

# ตั้งค่า CORS
docker exec keycloak /opt/keycloak/bin/kcadm.sh update clients/CLIENT_ID -r myapp \
  -s 'webOrigins=["https://app.example.com"]'

# ทดสอบ Token
curl -X POST "http://localhost:8080/realms/myapp/protocol/openid-connect/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "grant_type=password" \
  -d "client_id=web-app" \
  -d "client_secret=web-app-secret-123" \
  -d "username=john" \
  -d "password=JohnPass123!" \
  -d "scope=openid email profile" | python3 -m json.tool

# Decode ID Token
# python3 -c "
# import jwt, sys, json
# token = sys.argv[1]
# claims = jwt.decode(token, options={'verify_signature': False})
# print(json.dumps(claims, indent=2))
# " "eyJhbGci..."

Monitoring และ Audit Log สำหรับ IAM

ตั้งค่า event logging และ monitoring สำหรับ Keycloak

# เปิด Event Logging
docker exec keycloak /opt/keycloak/bin/kcadm.sh update events/config -r myapp \
  -s 'eventsEnabled=true' \
  -s 'eventsExpiration=2592000' \
  -s 'eventsListeners=["jboss-logging"]' \
  -s 'enabledEventTypes=["LOGIN","LOGIN_ERROR","LOGOUT","REGISTER","TOKEN_EXCHANGE","REFRESH_TOKEN"]' \
  -s 'adminEventsEnabled=true' \
  -s 'adminEventsDetailsEnabled=true'

# ดู Events ผ่าน Admin API
curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  "http://localhost:8080/admin/realms/myapp/events?type=LOGIN_ERROR&max=10" | \
  python3 -m json.tool

# สร้าง Event Monitor ด้วย Python
#!/usr/bin/env python3
# keycloak_monitor.py
import requests
import time
from datetime import datetime

KEYCLOAK_URL = "http://localhost:8080"
REALM = "myapp"
ADMIN_USER = "admin"
ADMIN_PASS = "AdminPass123!"

def get_admin_token():
    r = requests.post(f"{KEYCLOAK_URL}/realms/master/protocol/openid-connect/token",
        data={
            "grant_type": "password",
            "client_id": "admin-cli",
            "username": ADMIN_USER,
            "password": ADMIN_PASS,
        })
    return r.json()["access_token"]

def get_events(token, event_type=None, max_results=50):
    params = {"max": max_results}
    if event_type:
        params["type"] = event_type
    
    r = requests.get(
        f"{KEYCLOAK_URL}/admin/realms/{REALM}/events",
        headers={"Authorization": f"Bearer {token}"},
        params=params
    )
    return r.json()

def monitor():
    token = get_admin_token()
    print(f"[{datetime.now()}] Keycloak Event Monitor started")
    
    seen_events = set()
    
    while True:
        try:
            events = get_events(token, max_results=20)
            
            for event in events:
                event_id = f"{event.get('time')}_{event.get('type')}_{event.get('userId', '')}"
                if event_id in seen_events:
                    continue
                seen_events.add(event_id)
                
                ts = datetime.fromtimestamp(event["time"] / 1000)
                etype = event.get("type", "UNKNOWN")
                user = event.get("userId", "N/A")[:8]
                ip = event.get("ipAddress", "N/A")
                
                if "ERROR" in etype:
                    print(f"[ALERT] {ts} {etype} user={user} ip={ip}")
                else:
                    print(f"[INFO]  {ts} {etype} user={user} ip={ip}")
            
            if len(seen_events) > 1000:
                seen_events = set(list(seen_events)[-500:])
                
        except Exception as e:
            print(f"[ERROR] {e}")
            token = get_admin_token()
        
        time.sleep(30)

if __name__ == "__main__":
    monitor()

FAQ คำถามที่พบบ่อย

Q: OpenID Connect กับ OAuth 2.0 ต่างกันอย่างไร?

A: OAuth 2.0 เป็น authorization framework สำหรับให้สิทธิ์เข้าถึง resources ส่วน OpenID Connect เป็น identity layer ที่สร้างบน OAuth 2.0 เพิ่ม ID Token สำหรับ authentication OIDC ตอบคำถามว่าผู้ใช้คือใคร ส่วน OAuth 2.0 ตอบคำถามว่าผู้ใช้มีสิทธิ์ทำอะไรได้บ้าง ในทางปฏิบัติมักใช้ทั้งสองร่วมกัน

Q: Keycloak กับ Auth0 กับ Okta ควรเลือกอันไหน?

A: Keycloak เป็น Open Source ไม่มีค่าใช้จ่าย ปรับแต่งได้ทุกอย่าง แต่ต้อง host และดูแลเอง Auth0 เป็น managed service ใช้งานง่าย มี free tier สำหรับ 7,000 users แต่ราคาสูงเมื่อ scale Okta เหมาะสำหรับองค์กรใหญ่ มี enterprise features ครบ แต่ราคาแพงที่สุด สำหรับ startup และ SME แนะนำ Keycloak

Q: PKCE คืออะไรและจำเป็นไหม?

A: PKCE (Proof Key for Code Exchange) เป็น security extension สำหรับ Authorization Code Flow ป้องกัน authorization code interception attack จำเป็นสำหรับ public clients เช่น SPA และ mobile apps ที่ไม่สามารถเก็บ client secret ได้อย่างปลอดภัย ปัจจุบันแนะนำให้ใช้ PKCE สำหรับทุก client แม้แต่ confidential clients

Q: JWT Token ควรมีอายุเท่าไหร่?

A: Access Token ควรมีอายุสั้น 5-15 นาที เพราะถ้าถูกขโมยจะใช้ได้แค่ช่วงสั้น Refresh Token ควรมีอายุยาวกว่า 1-30 วัน ขึ้นอยู่กับ security requirements ID Token ใช้แค่ตอน authentication ไม่ควรใช้เป็น session token ใช้ Refresh Token สำหรับขอ Access Token ใหม่