Model Registry กับ RBAC ABAC Policy —

Model Registry และ Access Control

Model Registry เป็นระบบจัดการ ML Models แบบรวมศูนย์ เมื่อองค์กรมี Models หลายสิบหรือหลายร้อยตัว การจัดการสิทธิ์เข้าถึงเป็นสิ่งจำเป็น ใครสร้าง Model ได้ ใครดูได้ ใคร Deploy ได้ ใครลบได้ ต้องมี Policy ที่ชัดเจน

RBAC (Role-Based Access Control) กำหนดสิทธิ์ตามบทบาท เช่น Data Scientist, ML Engineer, Admin ส่วน ABAC (Attribute-Based Access Control) กำหนดสิทธิ์ตาม Attributes หลายมิติ เช่น Department, Model Risk Level, Environment ใช้ทั้งคู่ร่วมกันให้ครอบคลุมทุกสถานการณ์

เนื้อหาเกี่ยวข้อง — บทความที่เกี่ยวข้อง: คริปโตเคือเรนซี่

RBAC สำหรับ Model Registry

# rbac_model_registry.py — RBAC สำหรับ Model Registry
from enum import Enum
from dataclasses import dataclass, field
from typing import Set, Dict, Optional
import functools

class Permission(Enum):
    """Permissions สำหรับ Model Registry"""
    MODEL_CREATE       = "model:create"
    MODEL_READ         = "model:read"
    MODEL_UPDATE       = "model:update"
    MODEL_DELETE       = "model:delete"
    MODEL_DEPLOY       = "model:deploy"
    MODEL_PROMOTE      = "model:promote"       # เลื่อน Stage
    MODEL_ARCHIVE      = "model:archive"
    EXPERIMENT_CREATE   = "experiment:create"
    EXPERIMENT_READ     = "experiment:read"
    EXPERIMENT_DELETE   = "experiment:delete"
    REGISTRY_ADMIN      = "registry:admin"
    AUDIT_READ          = "audit:read"

class Role(Enum):
    """Roles สำหรับ ML Team"""
    VIEWER          = "viewer"
    DATA_SCIENTIST  = "data_scientist"
    ML_ENGINEER     = "ml_engineer"
    REVIEWER        = "reviewer"
    ADMIN           = "admin"

# Role-Permission Mapping
ROLE_PERMISSIONS: Dict[Role, Set[Permission]] = {
    Role.VIEWER: {
        Permission.MODEL_READ,
        Permission.EXPERIMENT_READ,
    },
    Role.DATA_SCIENTIST: {
        Permission.MODEL_CREATE,
        Permission.MODEL_READ,
        Permission.MODEL_UPDATE,
        Permission.EXPERIMENT_CREATE,
        Permission.EXPERIMENT_READ,
    },
    Role.ML_ENGINEER: {
        Permission.MODEL_CREATE,
        Permission.MODEL_READ,
        Permission.MODEL_UPDATE,
        Permission.MODEL_DEPLOY,
        Permission.MODEL_PROMOTE,
        Permission.EXPERIMENT_CREATE,
        Permission.EXPERIMENT_READ,
    },
    Role.REVIEWER: {
        Permission.MODEL_READ,
        Permission.MODEL_PROMOTE,
        Permission.EXPERIMENT_READ,
        Permission.AUDIT_READ,
    },
    Role.ADMIN: {p for p in Permission},  # ทุก Permission
}

@dataclass
class User:
    """User ในระบบ"""
    id: str
    name: str
    email: str
    roles: Set[Role] = field(default_factory=set)
    department: str = ""
    team: str = ""

    @property
    def permissions(self) -> Set[Permission]:
        """รวม Permissions จากทุก Roles"""
        perms = set()
        for role in self.roles:
            perms.update(ROLE_PERMISSIONS.get(role, set()))
        return perms

    def has_permission(self, perm: Permission) -> bool:
        return perm in self.permissions

class RBACModelRegistry:
    """Model Registry พร้อม RBAC"""

    def __init__(self):
        self.models: Dict[str, dict] = {}
        self.audit_log = []

    def _check_permission(self, user: User, perm: Permission):
        if not user.has_permission(perm):
            self._log(user, perm.value, "DENIED")
            raise PermissionError(
                f"User '{user.name}' ({', '.join(r.value for r in user.roles)}) "
                f"does not have permission '{perm.value}'"
            )
        self._log(user, perm.value, "ALLOWED")

    def _log(self, user, action, result):
        self.audit_log.append({
            "user": user.id, "action": action,
            "result": result,
        })

    def create_model(self, user: User, model_name: str, metadata: dict):
        self._check_permission(user, Permission.MODEL_CREATE)
        self.models[model_name] = {
            "name": model_name, "owner": user.id,
            "stage": "None", "metadata": metadata,
        }
        print(f"Model '{model_name}' created by {user.name}")

    def deploy_model(self, user: User, model_name: str, env: str):
        self._check_permission(user, Permission.MODEL_DEPLOY)
        if model_name not in self.models:
            raise ValueError(f"Model '{model_name}' not found")
        self.models[model_name]["stage"] = "Production"
        self.models[model_name]["environment"] = env
        print(f"Model '{model_name}' deployed to {env} by {user.name}")

    def promote_model(self, user: User, model_name: str, stage: str):
        self._check_permission(user, Permission.MODEL_PROMOTE)
        self.models[model_name]["stage"] = stage
        print(f"Model '{model_name}' promoted to {stage} by {user.name}")

    def delete_model(self, user: User, model_name: str):
        self._check_permission(user, Permission.MODEL_DELETE)
        del self.models[model_name]
        print(f"Model '{model_name}' deleted by {user.name}")

# ตัวอย่าง
registry = RBACModelRegistry()

alice = User("u1", "Alice", "alice@corp.com",
             roles={Role.DATA_SCIENTIST}, department="ML")
bob = User("u2", "Bob", "bob@corp.com",
           roles={Role.ML_ENGINEER}, department="Platform")
admin = User("u3", "Admin", "admin@corp.com",
             roles={Role.ADMIN})

registry.create_model(alice, "fraud_detector", {"version": "1.0"})
# alice.deploy_model → PermissionError (Data Scientist ไม่มีสิทธิ์ Deploy)
registry.deploy_model(bob, "fraud_detector", "production")

ABAC Policy Engine

# abac_policy.py — Attribute-Based Access Control
import json
from dataclasses import dataclass
from typing import Any, Dict, List, Callable
from datetime import datetime

@dataclass
class PolicyRule:
    """ABAC Policy Rule"""
    name: str
    description: str
    effect: str              # "allow" or "deny"
    conditions: Dict[str, Any]
    priority: int = 0        # สูงกว่า = สำคัญกว่า

class ABACEngine:
    """ABAC Policy Engine สำหรับ Model Registry"""

    def __init__(self):
        self.policies: List[PolicyRule] = []

    def add_policy(self, policy: PolicyRule):
        self.policies.append(policy)
        self.policies.sort(key=lambda p: p.priority, reverse=True)

    def evaluate(self, subject: dict, resource: dict,
                 action: str, environment: dict) -> bool:
        """ประเมิน Policy"""
        context = {
            "subject": subject,
            "resource": resource,
            "action": action,
            "environment": environment,
        }

        for policy in self.policies:
            if self._match_conditions(policy.conditions, context):
                allowed = policy.effect == "allow"
                print(f"Policy '{policy.name}': {policy.effect} "
                      f"(action={action})")
                return allowed

        # Default Deny
        print(f"No matching policy: DENY (action={action})")
        return False

    def _match_conditions(self, conditions: dict, context: dict) -> bool:
        """ตรวจสอบว่า Context ตรงกับ Conditions หรือไม่"""
        for key, expected in conditions.items():
            parts = key.split(".")
            value = context
            for part in parts:
                value = value.get(part, None) if isinstance(value, dict) else None
                if value is None:
                    return False

            if isinstance(expected, list):
                if value not in expected:
                    return False
            elif isinstance(expected, dict):
                op = list(expected.keys())[0]
                exp_val = expected[op]
                if op == "$gt" and not (value > exp_val):
                    return False
                elif op == "$lt" and not (value < exp_val):
                    return False
                elif op == "$gte" and not (value >= exp_val):
                    return False
                elif op == "$in" and value not in exp_val:
                    return False
                elif op == "$not" and value == exp_val:
                    return False
            else:
                if value != expected:
                    return False
        return True

# กำหนด Policies
engine = ABACEngine()

# Policy 1: Admin ทำได้ทุกอย่าง
engine.add_policy(PolicyRule(
    name="admin_full_access",
    description="Admin has full access",
    effect="allow",
    conditions={"subject.role": "admin"},
    priority=100,
))

# Policy 2: Data Scientist สร้าง Model ได้เฉพาะ Department ตัวเอง
engine.add_policy(PolicyRule(
    name="ds_create_own_dept",
    description="DS can create models in own department",
    effect="allow",
    conditions={
        "subject.role": "data_scientist",
        "action": "create",
        "subject.department": {"$in": ["ml", "data"]},
    },
    priority=50,
))

# Policy 3: ML Engineer Deploy ได้เฉพาะ Low/Medium Risk Models
engine.add_policy(PolicyRule(
    name="mle_deploy_low_risk",
    description="MLE can deploy low/medium risk models",
    effect="allow",
    conditions={
        "subject.role": "ml_engineer",
        "action": "deploy",
        "resource.risk_level": {"$in": ["low", "medium"]},
    },
    priority=50,
))

# Policy 4: High Risk Models ต้อง Reviewer Approve ก่อน Deploy
engine.add_policy(PolicyRule(
    name="high_risk_require_review",
    description="High risk models require reviewer approval",
    effect="deny",
    conditions={
        "action": "deploy",
        "resource.risk_level": "high",
        "resource.approved": {"$not": True},
    },
    priority=80,
))

# Policy 5: ห้าม Deploy นอกเวลาทำงาน
engine.add_policy(PolicyRule(
    name="no_deploy_off_hours",
    description="No deployments outside business hours",
    effect="deny",
    conditions={
        "action": "deploy",
        "environment.business_hours": False,
    },
    priority=90,
))

# ทดสอบ
subject = {"id": "u2", "role": "ml_engineer", "department": "platform"}
resource = {"name": "fraud_model", "risk_level": "low", "approved": False}
env = {"business_hours": True, "region": "ap-southeast-1"}

result = engine.evaluate(subject, resource, "deploy", env)
print(f"Access: {'ALLOWED' if result else 'DENIED'}")

Integration กับ MLflow

# mlflow_rbac_middleware.py — RBAC Middleware สำหรับ MLflow from flask import Flask, request, jsonify, g import functools import jwt import os app = Flask(__name__) JWT_SECRET = os.environ.get("JWT_SECRET", "secret") # RBAC Decorator def require_permission(permission): def decorator(f): @functools.wraps(f) def wrapped(*args, **kwargs): token = request.headers.get("Authorization", "").replace("Bearer ", "") if not token: return jsonify({"error": "No token"}), 401 try: payload = jwt.decode(token, JWT_SECRET, algorithms=["HS256"]) except jwt.InvalidTokenError: return jsonify({"error": "Invalid token"}), 401 user_roles = set(payload.get("roles", [])) user_perms = set() for role_name in user_roles: try: role = Role(role_name) user_perms.update(ROLE_PERMISSIONS.get(role, set())) except ValueError: pass perm = Permission(permission) if perm not in user_perms: return jsonify({ "error": f"Permission denied: {permission}", "user": payload.get("sub"), "roles": list(user_roles), }), 403 g.user = payload return f(*args, **kwargs) return wrapped return decorator # Protected Endpoints @app.route("/api/models", methods=["POST"]) @require_permission("model:create") def create_model(): data = request.json # Forward to MLflow return jsonify({"status": "created", "model": data.get("name")}) @app.route("/api/models//deploy", methods=["POST"]) @require_permission("model:deploy") def deploy_model(name): env = request.json.get("environment", "staging") return jsonify({"status": "deployed", "model": name, "env": env}) @app.route("/api/models//promote", methods=["POST"]) @require_permission("model:promote") def promote_model(name): stage = request.json.get("stage", "Staging") return jsonify({"status": "promoted", "model": name, "stage": stage}) @app.route("/api/models", methods=["GET"]) @require_permission("model:read") def list_models(): return jsonify({"models": []}) # Kubernetes RBAC สำหรับ Model Serving # --- # apiVersion: rbac.authorization.k8s.io/v1 # kind: Role # metadata: # name: ml-engineer-role # namespace: ml-serving # rules: # - apiGroups: ["serving.kubeflow.org"] # resources: ["inferenceservices"] # verbs: ["get", "list", "create", "update"] # - apiGroups: [""] # resources: ["pods", "services"] # verbs: ["get", "list"] # --- # kind: RoleBinding # apiVersion: rbac.authorization.k8s.io/v1 # metadata: # name: ml-engineer-binding # namespace: ml-serving # subjects: # - kind: Group # name: ml-engineers # apiGroup: rbac.authorization.k8s.io # roleRef: # kind: Role # name: ml-engineer-role # apiGroup: rbac.authorization.k8s.io

Best Practices

• Principle of Least Privilege: ให้สิทธิ์น้อยที่สุดที่จำเป็น Data Scientist ไม่ต้อง Deploy ได้
• Separation of Duties: คนสร้าง Model ไม่ควรเป็นคน Approve Deploy ต้องมี Reviewer
• Audit Trail: บันทึกทุกการกระทำ ใครทำอะไร เมื่อไร สำหรับ Compliance และ Forensics
• High Risk Models: กำหนด Risk Level สำหรับ Models ที่กระทบลูกค้าโดยตรง ต้อง Review ก่อน Deploy
• Time-based Policies: ห้าม Deploy นอกเวลาทำงาน หรือต้อง Senior Approve
• Environment Isolation: Dev/Staging ให้สิทธิ์กว้าง Production จำกัดเฉพาะ ML Engineer และ Admin

Model Registry คืออะไร

Model Registry เป็นระบบจัดการ ML Models แบบรวมศูนย์ เก็บ Model Versions, Metadata, Artifacts, Lineage Track Model Stage (Staging, Production, Archived) เครื่องมือนิยม เช่น MLflow, W&B, Neptune

สรุป

การจัดการสิทธิ์เข้าถึง Model Registry เป็นสิ่งจำเป็นเมื่อองค์กรมี ML Models จำนวนมาก RBAC ให้พื้นฐานที่ง่ายต่อการจัดการด้วย Roles และ Permissions ABAC เสริมด้วย Fine-grained Policies ตาม Attributes หลายมิติ สิ่งสำคัญคือ Least Privilege, Separation of Duties, Audit Trail และ Risk-based Policies สำหรับ High Risk Models

เนื้อหาเกี่ยวข้อง — แนะนำให้อ่าน Href Html คืออะไร — คู่มือฉบับสมบูรณ์ 2026 — คู่มือฉบับสมบูรณ์ 2026