Trong bối cảnh AI API trở thành cơ sở hạ tầng doanh nghiệp, việc quản lý truy cập, giám sát hoạt động và đảm bảo tuân thủ pháp luật không còn là tùy chọn mà là bắt buộc. Bài viết này sẽ hướng dẫn chi tiết cách triển khai hệ thống compliance audit end-to-end khi tích hợp DeepSeek V3.2 và GPT-4o vào hạ tầng doanh nghiệp.

Mở đầu: So sánh các phương án tiếp cận API AI

Là một kỹ sư đã triển khai AI infrastructure cho 15+ dự án enterprise, tôi đã thử nghiệm hầu hết các phương án hiện có. Dưới đây là bảng so sánh thực tế từ kinh nghiệm triển khai:

Tiêu chí HolySheep AI API chính thức (OpenAI/Anthropic) Dịch vụ Relay/API Proxy khác
Chi phí GPT-4o ~$8/MTok (tỷ giá ưu đãi) $15/MTok $10-13/MTok
Chi phí Claude Sonnet 4.5 $15/MTok $15/MTok $18-20/MTok
DeepSeek V3.2 $0.42/MTok 💡 Không hỗ trợ $0.80-1.20/MTok
Độ trễ trung bình <50ms (Việt Nam) 200-400ms 100-300ms
Tín dụng miễn phí ✅ Có khi đăng ký ❌ Không ❌ Không
Thanh toán WeChat/Alipay/VNPay Thẻ quốc tế Hạn chế
Enterprise Audit Logs ✅ Dashboard tích hợp ✅ Có nhưng đắt đỏ ⚠️ Thường không có
Key Isolation ✅ Native support ✅ Có ❌ Thường shared key

Phù hợp / Không phù hợp với ai

✅ NÊN sử dụng HolySheep AI khi:

❌ KHÔNG phù hợp khi:

Kiến trúc tổng quan: Compliance Audit Framework

Trước khi đi vào chi tiết implementation, hãy xem xét kiến trúc tổng thể mà tôi đã triển khai cho một dự án fintech:

┌─────────────────────────────────────────────────────────────────┐
│                    ENTERPRISE AI GATEWAY                         │
├─────────────────────────────────────────────────────────────────┤
│  ┌─────────────┐    ┌─────────────┐    ┌─────────────────────┐ │
│  │ Rate Limiter│───▶│ Auth Layer  │───▶│ Permission Checker │ │
│  │ (100 req/m) │    │ (JWT/RBAC)  │    │ (Role-based access) │ │
│  └─────────────┘    └─────────────┘    └─────────────────────┘ │
│                                              │                   │
│                                              ▼                   │
│  ┌─────────────────────────────────────────────────────────────┐│
│  │                   AUDIT LOGGER                              ││
│  │  - Timestamp, UserID, API Key (masked), Model, Tokens      ││
│  │  - Request/Response Hash, Latency, Cost                    ││
│  └─────────────────────────────────────────────────────────────┘│
│                                              │                   │
│        ┌────────────────┐    ┌────────────────┐                 │
│        │   DeepSeek     │    │    GPT-4o      │                 │
│        │   V3.2         │    │                │                 │
│        │   $0.42/MTok   │    │   $8/MTok      │                 │
│        └────────────────┘    └────────────────┘                 │
└─────────────────────────────────────────────────────────────────┘

Giá và ROI: Phân tích chi phí thực tế

Model API chính thức HolySheep AI Tiết kiệm Use case phù hợp
GPT-4.1 $30/MTok $8/MTok 73% Complex reasoning, coding
Claude Sonnet 4.5 $15/MTok $15/MTok ~0% Long context analysis
Gemini 2.5 Flash $2.50/MTok $2.50/MTok ~0% High volume, low latency
DeepSeek V3.2 N/A $0.42/MTok 85%+ Bulk processing, RAG

Tính toán ROI thực tế:

Triển khai chi tiết: Phần 1 - Key Isolation

Key isolation là nền tảng của mọi compliance framework. Mỗi team/customer phải có API key riêng biệt với tracking độc lập.

Bước 1: Cấu trúc Database cho Multi-tenant Key Management

-- Schema cho enterprise key management
CREATE TABLE api_keys (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    key_hash VARCHAR(64) NOT NULL UNIQUE,  -- SHA-256 hash of actual key
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    user_id UUID REFERENCES users(id),
    model_access JSONB NOT NULL DEFAULT '["gpt-4o", "deepseek-v3"]',
    daily_limit_tokens INTEGER DEFAULT 1000000,
    monthly_limit_usd DECIMAL(10,2) DEFAULT 5000.00,
    is_active BOOLEAN DEFAULT true,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    expires_at TIMESTAMP WITH TIME ZONE,
    last_used_at TIMESTAMP WITH TIME ZONE,
    metadata JSONB DEFAULT '{}'
);

CREATE TABLE key_usage_log (
    id BIGSERIAL PRIMARY KEY,
    key_id UUID NOT NULL REFERENCES api_keys(id),
    model VARCHAR(50) NOT NULL,
    input_tokens INTEGER NOT NULL,
    output_tokens INTEGER NOT NULL,
    cost_usd DECIMAL(10,2) NOT NULL,
    latency_ms INTEGER NOT NULL,
    request_hash VARCHAR(64) NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

-- Indexes cho query hiệu suất cao
CREATE INDEX idx_key_usage_key_id ON key_usage_log(key_id, created_at DESC);
CREATE INDEX idx_key_usage_daily ON key_usage_log(key_id, DATE(created_at));

-- Trigger tự động update last_used_at
CREATE OR REPLACE FUNCTION update_last_used()
RETURNS TRIGGER AS $$
BEGIN
    UPDATE api_keys SET last_used_at = NOW() WHERE id = NEW.key_id;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER tr_update_last_used
AFTER INSERT ON key_usage_log
FOR EACH ROW EXECUTE FUNCTION update_last_used();

Bước 2: API Gateway Implementation với HolySheep

#!/usr/bin/env python3
"""
Enterprise AI Gateway với Key Isolation và Audit Logging
Base URL: https://api.holysheep.ai/v1
"""

import hashlib
import time
from datetime import datetime, timedelta
from typing import Optional
from dataclasses import dataclass
import httpx


Configuration - Thay YOUR_HOLYSHEEP_API_KEY bằng key thực tế

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
ADMIN_API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # Key quản trị viên

@dataclass
class KeyInfo:
    key_hash: str
    tenant_id: str
    user_id: str
    model_access: list
    daily_limit: int
    monthly_limit: float
    is_active: bool

class EnterpriseAIGateway:
    """
    Gateway cho phép doanh nghiệp quản lý nhiều API keys
    với isolation hoàn chỉnh và audit logging
    """
    
    def __init__(self):
        self.client = httpx.Client(
            base_url=HOLYSHEEP_BASE_URL,
            headers={
                "Authorization": f"Bearer {ADMIN_API_KEY}",
                "Content-Type": "application/json"
            },
            timeout=30.0
        )
        # Cache cho rate limiting
        self.daily_usage = {}  # {key_hash: (date, tokens_used)}
    
    def _hash_api_key(self, api_key: str) -> str:
        """Tạo hash SHA-256 từ API key - không bao giờ lưu key thực"""
        return hashlib.sha256(api_key.encode()).hexdigest()
    
    def _check_daily_limit(self, key_info: KeyInfo, requested_tokens: int) -> bool:
        """Kiểm tra giới hạn daily token"""
        today = datetime.now().date()
        key = key_info.key_hash
        
        if key not in self.daily_usage or self.daily_usage[key][0] != today:
            self.daily_usage[key] = (today, 0)
        
        _, used = self.daily_usage[key]
        return (used + requested_tokens) <= key_info.daily_limit
    
    def _update_daily_usage(self, key_info: KeyInfo, tokens_used: int):
        """Cập nhật usage counter"""
        today = datetime.now().date()
        key = key_info.key_hash
        
        if key in self.daily_usage and self.daily_usage[key][0] == today:
            _, used = self.daily_usage[key]
            self.daily_usage[key] = (today, used + tokens_used)
        else:
            self.daily_usage[key] = (today, tokens_used)
    
    def _log_to_audit(self, key_info: KeyInfo, model: str, 
                      input_tokens: int, output_tokens: int,
                      cost: float, latency_ms: int, request_id: str):
        """
        Ghi log vào audit system
        Trong production, đây sẽ gọi đến database hoặc log service
        """
        audit_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "key_id": key_info.key_hash[:16] + "...",  # Masked
            "tenant_id": key_info.tenant_id,
            "user_id": key_info.user_id,
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost_usd": round(cost, 4),
            "latency_ms": latency_ms,
            "request_id": request_id
        }
        print(f"[AUDIT] {audit_entry}")
        # Trong production: lưu vào PostgreSQL, Elasticsearch, hoặc S3
    
    async def chat_completion(
        self,
        api_key: str,
        model: str,
        messages: list,
        max_tokens: int = 1000
    ) -> dict:
        """
        Gửi request đến HolySheep với key isolation và audit logging
        """
        start_time = time.time()
        request_id = f"req_{datetime.utcnow().timestamp()}"
        
        # Bước 1: Xác thực và lấy thông tin key
        key_hash = self._hash_api_key(api_key)
        key_info = self._get_key_info(key_hash)  # Implement từ database
        
        # Bước 2: Kiểm tra key có active không
        if not key_info.is_active:
            raise PermissionError("API key đã bị vô hiệu hóa")
        
        # Bước 3: Kiểm tra quyền truy cập model
        if model not in key_info.model_access:
            raise PermissionError(f"Không có quyền truy cập model: {model}")
        
        # Bước 4: Kiểm tra daily limit (ước tính)
        estimated_tokens = self._estimate_tokens(messages) + max_tokens
        if not self._check_daily_limit(key_info, estimated_tokens):
            raise PermissionError("Đã vượt quá giới hạn daily token")
        
        # Bước 5: Gửi request đến HolySheep
        payload = {
            "model": model,
            "messages": messages,
            "max_tokens": max_tokens
        }
        
        response = self.client.post("/chat/completions", json=payload)
        
        # Bước 6: Tính toán chi phí và log
        latency_ms = int((time.time() - start_time) * 1000)
        result = response.json()
        
        usage = result.get("usage", {})
        input_tokens = usage.get("prompt_tokens", 0)
        output_tokens = usage.get("completion_tokens", 0)
        
        # Tính chi phí theo bảng giá HolySheep 2026
        cost = self._calculate_cost(model, input_tokens, output_tokens)
        
        # Cập nhật usage
        self._update_daily_usage(key_info, input_tokens + output_tokens)
        
        # Log audit
        self._log_to_audit(
            key_info, model, input_tokens, output_tokens,
            cost, latency_ms, request_id
        )
        
        return result
    
    def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        """Tính chi phí theo bảng giá HolySheep 2026"""
        pricing = {
            "gpt-4.1": {"input": 0.002, "output": 0.008},  # $8/MTok input
            "gpt-4o": {"input": 0.002, "output": 0.008},
            "claude-sonnet-4.5": {"input": 0.003, "output": 0.015},
            "gemini-2.5-flash": {"input": 0.000125, "output": 0.0005},
            "deepseek-v3": {"input": 0.00007, "output": 0.00028}  # $0.42/MTok
        }
        
        if model not in pricing:
            model = "deepseek-v3"  # Default về rẻ nhất
        
        p = pricing[model]
        return (input_tokens / 1_000_000 * p["input"] * 1_000_000 / 1000 + 
                output_tokens / 1_000_000 * p["output"] * 1_000_000 / 1000)
    
    def _estimate_tokens(self, messages: list) -> int:
        """Ước tính tokens từ messages - heuristic đơn giản"""
        total_chars = sum(len(str(m)) for m in messages)
        return int(total_chars * 0.25)  # ~4 characters per token
    
    def _get_key_info(self, key_hash: str) -> KeyInfo:
        """
        Lấy thông tin key từ database
        Implement thực tế sẽ query PostgreSQL
        """
        # Mock - implement thực tế với database
        return KeyInfo(
            key_hash=key_hash,
            tenant_id="tenant_001",
            user_id="user_001",
            model_access=["gpt-4o", "deepseek-v3", "claude-sonnet-4.5"],
            daily_limit=5_000_000,
            monthly_limit=10000.0,
            is_active=True
        )


Ví dụ sử dụng

gateway = EnterpriseAIGateway()


Request với key isolation

try:
    response = gateway.chat_completion(
        api_key="sk_test_xxx_your_customer_key",  # Key của customer
        model="deepseek-v3",  # Model rẻ, phù hợp cho internal tools
        messages=[
            {"role": "system", "content": "Bạn là trợ lý AI"},
            {"role": "user", "content": "Giải thích về compliance audit"}
        ],
        max_tokens=500
    )
    print(f"Response ID: {response.get('id')}")
except PermissionError as e:
    print(f"Permission denied: {e}")

Triển khai chi tiết: Phần 2 - Log Retention và Compliance

Yêu cầu log retention thường là 6 tháng đến 7 năm tùy ngành. Dưới đây là hệ thống log hoàn chỉnh:

#!/usr/bin/env python3
"""
Compliance Log Retention System
Đảm bảo tuân thủ các quy định về lưu trữ log
"""

import json
import gzip
import hashlib
from datetime import datetime, timedelta
from pathlib import Path
from typing import Iterator
import boto3
from botocore.exceptions import ClientError

class ComplianceLogManager:
    """
    Quản lý log theo tiêu chuẩn compliance:
    - Retention: 7 năm (tài chính), 2 năm (chung)
    - Encryption at rest (AES-256)
    - Tamper detection (hash chain)
    """
    
    def __init__(self, retention_years: int = 7, region: str = "ap-southeast-1"):
        self.retention_years = retention_years
        self.region = region
        self.s3_client = boto3.client('s3', region_name=region)
        self.bucket_name = "enterprise-ai-compliance-logs"
        self.last_hash = None  # Cho hash chain integrity
    
    def _calculate_retention_date(self) -> datetime:
        """Tính ngày hết hạn retention"""
        return datetime.now() - timedelta(days=self.retention_years * 365)
    
    def _generate_log_entry(
        self,
        event_type: str,
        api_key_hash: str,
        user_id: str,
        tenant_id: str,
        model: str,
        request_data: dict,
        response_data: dict,
        metadata: dict
    ) -> dict:
        """
        Tạo log entry với đầy đủ thông tin compliance
        """
        timestamp = datetime.utcnow().isoformat() + "Z"
        
        # Hash of request content (PII protection)
        request_hash = hashlib.sha256(
            json.dumps(request_data, sort_keys=True).encode()
        ).hexdigest()
        
        # Hash of response content
        response_hash = hashlib.sha256(
            json.dumps(response_data, sort_keys=True).encode()
        ).hexdigest()
        
        # Chain hash cho tamper detection
        chain_input = f"{self.last_hash or 'genesis'}:{request_hash}:{timestamp}"
        chain_hash = hashlib.sha256(chain_input.encode()).hexdigest()
        self.last_hash = chain_hash
        
        return {
            "log_id": f"log_{datetime.utcnow().timestamp()}_{hashlib.md5(user_id.encode()).hexdigest()[:8]}",
            "timestamp": timestamp,
            "event_type": event_type,
            
            # Audit trail
            "api_key_hash": api_key_hash,  # Không lưu key thực
            "user_id_hash": hashlib.sha256(user_id.encode()).hexdigest(),
            "tenant_id": tenant_id,
            
            # Request/Response (không có PII)
            "model": model,
            "request_hash": request_hash,
            "response_hash": response_hash,
            "tokens_used": metadata.get("tokens", {}),
            "cost_usd": metadata.get("cost", 0.0),
            "latency_ms": metadata.get("latency", 0),
            
            # Integrity
            "chain_hash": chain_hash,
            "previous_hash": self.last_hash,
            
            # Compliance metadata
            "data_classification": metadata.get("classification", "internal"),
            "processing_region": self.region,
            "retention_until": (
                datetime.utcnow() + timedelta(days=self.retention_years * 365)
            ).isoformat() + "Z"
        }
    
    def write_log_batch(self, logs: list, partition: str = "daily") -> str:
        """
        Ghi batch log với compression và encryption
        partition: 'daily' | 'monthly' | 'tenant'
        """
        timestamp = datetime.utcnow()
        
        # Compress data
        log_json = json.dumps(logs, indent=2)
        compressed = gzip.compress(log_json.encode('utf-8'))
        
        # Determine S3 key path
        if partition == "daily":
            s3_key = f"logs/{timestamp.year}/{timestamp.month:02d}/{timestamp.day:02d}/logs_{timestamp.timestamp()}.json.gz"
        elif partition == "tenant":
            # Group by tenant for easier access
            s3_key = f"tenant-logs/{logs[0]['tenant_id']}/{timestamp.year}/{timestamp.month:02d}/logs.json.gz"
        else:
            s3_key = f"logs/{timestamp.year}/{timestamp.month:02d}/logs_{timestamp.timestamp()}.json.gz"
        
        # Write to S3 with appropriate storage class
        try:
            self.s3_client.put_object(
                Bucket=self.bucket_name,
                Key=s3_key,
                Body=compressed,
                ServerSideEncryption='AES-256',
                StorageClass='GLACIER',  # Cheap storage for old logs
                Metadata={
                    'retention-years': str(self.retention_years),
                    'log-count': str(len(logs)),
                    'first-timestamp': logs[0]['timestamp'],
                    'last-timestamp': logs[-1]['timestamp']
                }
            )
            return s3_key
        except ClientError as e:
            print(f"S3 write error: {e}")
            # Fallback to local storage
            local_path = Path(f"/compliance-logs/{s3_key}")
            local_path.parent.mkdir(parents=True, exist_ok=True)
            with gzip.open(local_path, 'wb') as f:
                f.write(compressed)
            return str(local_path)
    
    def query_logs(
        self,
        tenant_id: str,
        start_date: datetime,
        end_date: datetime,
        event_types: list = None
    ) -> Iterator[dict]:
        """
        Query logs với filtering - hỗ trợ compliance audit
        """
        prefix = f"tenant-logs/{tenant_id}/"
        
        paginator = self.s3_client.get_paginator('list_objects_v2')
        
        for page in paginator.paginate(
            Bucket=self.bucket_name,
            Prefix=prefix
        ):
            for obj in page.get('Contents', []):
                # Check date range
                if start_date <= obj['LastModified'].replace(tzinfo=None) <= end_date:
                    # Download and decompress
                    response = self.s3_client.get_object(
                        Bucket=self.bucket_name,
                        Key=obj['Key']
                    )
                    compressed = response['Body'].read()
                    logs = json.loads(gzip.decompress(compressed).decode('utf-8'))
                    
                    # Filter by event type
                    for log in logs:
                        if event_types is None or log['event_type'] in event_types:
                            yield log
    
    def generate_compliance_report(
        self,
        tenant_id: str,
        start_date: datetime,
        end_date: datetime
    ) -> dict:
        """
        Tạo báo cáo compliance theo yêu cầu audit
        """
        logs = list(self.query_logs(tenant_id, start_date, end_date))
        
        # Calculate statistics
        total_requests = len(logs)
        total_tokens = sum(
            log['tokens_used'].get('total', 0) for log in logs
        )
        total_cost = sum(log['cost_usd'] for log in logs)
        
        # Model usage breakdown
        model_usage = {}
        for log in logs:
            model = log['model']
            if model not in model_usage:
                model_usage[model] = {"requests": 0, "tokens": 0, "cost": 0}
            model_usage[model]["requests"] += 1
            model_usage[model]["tokens"] += log['tokens_used'].get('total', 0)
            model_usage[model]["cost"] += log['cost_usd']
        
        # Check for integrity issues
        integrity_issues = []
        for i, log in enumerate(logs):
            if i > 0 and log.get('previous_hash') != logs[i-1].get('chain_hash'):
                integrity_issues.append({
                    "log_id": log['log_id'],
                    "issue": "Chain hash mismatch"
                })
        
        return {
            "report_id": f"comp_report_{datetime.utcnow().timestamp()}",
            "generated_at": datetime.utcnow().isoformat() + "Z",
            "tenant_id": tenant_id,
            "period": {
                "start": start_date.isoformat(),
                "end": end_date.isoformat()
            },
            "summary": {
                "total_requests": total_requests,
                "total_tokens": total_tokens,
                "total_cost_usd": round(total_cost, 4)
            },
            "model_breakdown": model_usage,
            "integrity": {
                "total_logs": total_requests,
                "integrity_issues": len(integrity_issues),
                "issues_detail": integrity_issues[:10]  # First 10
            },
            "compliance": {
                "retention_policy_years": self.retention_years,
                "encryption": "AES-256",
                "storage_class": "GLACIER"
            }
        }


Ví dụ sử dụng

log_manager = ComplianceLogManager(retention_years=7)


Tạo log entry cho mỗi request

log_entry = log_manager._generate_log_entry(
    event_type="chat_completion",
    api_key_hash="abc123...",
    user_id="user_001",
    tenant_id="tenant_finance",
    model="deepseek-v3",
    request_data={"messages": [{"role": "user", "content": "[REDACTED]"}]},
    response_data={"content": "[REDACTED]"},
    metadata={
        "tokens": {"input": 150, "output": 300, "total": 450},
        "cost": 0.000189,  # $0.42/MTok * 450 tokens
        "latency": 45,
        "classification": "internal"
    }
)


Ghi batch log

batch_key = log_manager.write_log_batch([log_entry], partition="tenant")


Generate compliance report

report = log_manager.generate_compliance_report(
    tenant_id="tenant_finance",
    start_date=datetime.now() - timedelta(days=30),
    end_date=datetime.now()
)
print(json.dumps(report, indent=2))

Triển khai chi tiết: Phần 3 - Permission Layering với RBAC

#!/usr/bin/env python3
"""
Role-Based Access Control (RBAC) cho Enterprise AI Gateway
Triển khai permission layering theo nguyên tắc least privilege
"""

from enum import Enum
from typing import Set, Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import jwt
import hashlib

class Role(Enum):
    ADMIN = "admin"
    DEVELOPER = "developer"
    ANALYST = "analyst"
    END_USER = "end_user"
    AUDITOR = "auditor"

class Permission(Enum):
    # Model permissions
    CAN_ACCESS_GPT4 = "can_access_gpt4"
    CAN_ACCESS_CLAUDE = "can_access_claude"
    CAN_ACCESS_DEEPSEEK = "can_access_deepseek"
    CAN_ACCESS_GEMINI = "can_access_gemini"
    
    # Action permissions
    CAN_CREATE_KEYS = "can_create_keys"
    CAN_REVOKE_KEYS = "can_revoke_keys"
    CAN_VIEW_AUDIT = "can_view_audit"
    CAN_EXPORT_REPORTS = "can_export_reports"
    CAN_MANAGE_BILLING = "can_manage_billing"
    
    # Cost limits
    CAN_SPEND_UNLIMITED = "can_spend_unlimited"


Role-to-Permission mapping

ROLE_PERMISSIONS: Dict[Role, Set[Permission]] = {
    Role.ADMIN: {
        Permission.CAN_ACCESS_GPT4,
        Permission.CAN_ACCESS_CLAUDE,
        Permission.CAN_ACCESS_DEEPSEEK,
        Permission.CAN_ACCESS_GEMINI,
        Permission.CAN_CREATE_KEYS,
        Permission.CAN_REVOKE_KEYS,
        Permission.CAN_VIEW_AUDIT,
        Permission.CAN_EXPORT_REPORTS,
        Permission.CAN_MANAGE_BILLING,
        Permission.CAN_SPEND_UNLIMITED,
    },
    Role.DEVELOPER: {
        Permission.CAN_ACCESS_GPT4,
        Permission.CAN_ACCESS_DEEPSEEK,
        Permission.CAN_VIEW_AUDIT,
    },
    Role.ANALYST: {
        Permission.CAN_ACCESS_DEEPSEEK,  # Chỉ model rẻ
        Permission.CAN_VIEW_AUDIT,
        Permission.CAN_EXPORT_REPORTS,
    },
    Role.END_USER: {
        Permission.CAN_ACCESS_DEEPSEEK,
    },
    Role.AUDITOR: {
        Permission.CAN_VIEW_AUDIT,
        Permission.CAN_EXPORT_REPORTS,
    },
}

@dataclass
class User:
    user_id: str
    email: str
    role: Role
    tenant_id: str
    department: Optional[str] = None
    restrictions: Dict = field(default_factory=dict)

@dataclass
class APIKey:
    key_id: str
    key_hash: str
    user_id: str
    tenant_id: str
    role: Role
    allowed_models: Set[str]
    max_tokens_per_request: int = 100000
    max_requests_per_day: int = 10000
    max_spend_per_month_usd: float = 10000.0
    expires_at: Optional[datetime] = None
    is_active: bool = True

class RBACEngine:

Tài nguyên liên quan
Bài viết liên quan