Vấn đề thực tế mà tôi đã gặp phải
Năm ngoái, khi triển khai hệ thống RAG cho một doanh nghiệp thương mại điện tử lớn tại Việt Nam, tôi phát hiện ra một vấn đề nghiêm trọng: toàn bộ log hệ thống đang lưu trữ thông tin cá nhân của khách hàng một cách không mã hóa. Khi đội bảo mật kiểm tra, họ phát hiện hàng triệu bản ghi chứa số điện thoại, địa chỉ email, và thậm chí cả lịch sử giao dịch tài chính — tất cả đều nằm trong plain text logs.
Đó là khoảnh khắc tôi nhận ra: việc xử lý dữ liệu nhạy cảm trong các API call không chỉ là best practice mà là yêu cầu bắt buộc. Trong bài viết này, tôi sẽ chia sẻ giải pháp hoàn chỉnh mà tôi đã áp dụng thành công.
Tại sao cần log desensitization cho AI API?
Khi sử dụng
HolySheep AI hoặc bất kỳ nhà cung cấp API nào khác, request và response thường chứa:
- Thông tin cá nhân: email, số điện thoại, địa chỉ
- Dữ liệu tài chính: số thẻ, số tài khoản, số dư
- Credentials: API keys, passwords, tokens
- Thông tin doanh nghiệp: chiến lược kinh doanh, dữ liệu khách hàng
Nếu không xử lý desensitization, log files trở thành "mỏ vàng" cho attackers. Một breach có thể khiến công ty chịu phạt GDPR lên đến 4% doanh thu toàn cầu hoặc tương đương 20 triệu EUR.
Kiến trúc giải pháp desensitization
Tôi đã thiết kế một pipeline xử lý với 3 lớp bảo vệ:
┌─────────────────────────────────────────────────────────────┐
│ REQUEST PIPELINE │
├─────────────────────────────────────────────────────────────┤
│ User Input → Pre-processor → Desensitize → Log Buffer │
│ ↓ ↓ │
│ Sanitization Layer Sensitive Field Mask │
├─────────────────────────────────────────────────────────────┤
│ RESPONSE PIPELINE │
├─────────────────────────────────────────────────────────────┤
│ API Response → Parse JSON → Mask Sensitive → Store Log │
│ ↓ ↓ │
│ Field Detection Pattern Replacement │
└─────────────────────────────────────────────────────────────┘
Triển khai chi tiết với Python
1. Cấu hình cơ bản
import re
import hashlib
import logging
from typing import Any, Dict, List, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
Cấu hình logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("desensitization")
@dataclass
class SensitiveFieldConfig:
"""Cấu hình các trường nhạy cảm"""
field_names: List[str] = field(default_factory=lambda: [
"email", "phone", "password", "credit_card",
"ssn", "api_key", "token", "secret", "address"
])
patterns: List[re.Pattern] = field(default_factory=lambda: [
# Email pattern
re.compile(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'),
# Phone patterns (various formats)
re.compile(r'\+?[0-9]{1,4}?[-.\s]?\(?[0-9]{1,3}?\)?[-.\s]?[0-9]{1,4}[-.\s]?[0-9]{1,4}[-.\s]?[0-9]{1,9}'),
# Credit card patterns
re.compile(r'\b(?:\d{4}[-\s]?){3}\d{4}\b'),
# API key patterns
re.compile(r'sk-[a-zA-Z0-9]{32,}'),
# JWT tokens
re.compile(r'eyJ[a-zA-Z0-9_-]*\.eyJ[a-zA-Z0-9_-]*\.[a-zA-Z0-9_-]*'),
])
config = SensitiveFieldConfig()
print("✅ SensitiveFieldConfig initialized with", len(config.patterns), "patterns")
2. Core Desensitization Engine
import json
import copy
from functools import wraps
from typing import Union
class DesensitizationEngine:
"""Engine xử lý desensitization cho JSON data"""
def __init__(self, config: SensitiveFieldConfig):
self.config = config
self.mask_char = "*"
def _should_mask_field(self, field_name: str) -> bool:
"""Kiểm tra xem field có nên được mask không"""
field_lower = field_name.lower()
return any(sensitive in field_lower for sensitive in self.config.field_names)
def _mask_value(self, value: str, mask_type: str = "full") -> str:
"""Mask giá trị với các mức độ khác nhau"""
if not value or len(value) < 4:
return self.mask_char * 8
if mask_type == "full":
return self.mask_char * len(value)
elif mask_type == "partial":
# Giữ lại 4 ký tự cuối
return self.mask_char * (len(value) - 4) + value[-4:]
elif mask_type == "email":
# [email protected] → t***@example.com
parts = value.split("@")
if len(parts) == 2:
return parts[0][0] + self.mask_char * 3 + "@" + parts[1]
return self.mask_char * len(value)
elif mask_type == "phone":
# 0912345678 → ****5678
return self.mask_char * 4 + value[-4:]
return self.mask_char * len(value)
def _detect_value_type(self, field_name: str, value: str) -> str:
"""Phát hiện loại giá trị để áp dụng mask phù hợp"""
field_lower = field_name.lower()
if "email" in field_lower:
return "email"
elif "phone" in field_lower or "mobile" in field_lower:
return "phone"
elif "card" in field_lower or "credit" in field_lower:
return "full"
elif "password" in field_lower or "secret" in field_lower:
return "full"
elif "key" in field_lower or "token" in field_lower:
return "full"
elif "address" in field_lower:
return "partial"
return "partial"
def _process_dict(self, data: Dict) -> Dict:
"""Đệ quy xử lý dictionary"""
result = {}
for key, value in data.items():
if self._should_mask_field(key):
mask_type = self._detect_value_type(key, str(value))
result[key] = self._mask_value(str(value), mask_type)
logger.debug(f"Masked field '{key}' with type '{mask_type}'")
elif isinstance(value, dict):
result[key] = self._process_dict(value)
elif isinstance(value, list):
result[key] = self._process_list(value)
elif isinstance(value, str):
result[key] = self._apply_pattern_mask(value)
else:
result[key] = value
return result
def _process_list(self, data: List) -> List:
"""Đệ quy xử lý list"""
return [self._process_item(item) for item in data]
def _process_item(self, item: Any) -> Any:
"""Xử lý từng item trong list"""
if isinstance(item, dict):
return self._process_dict(item)
elif isinstance(item, list):
return self._process_list(item)
elif isinstance(item, str):
return self._apply_pattern_mask(item)
return item
def _apply_pattern_mask(self, text: str) -> str:
"""Áp dụng mask cho các pattern được tìm thấy trong text"""
result = text
for pattern in self.config.patterns:
matches = pattern.findall(result)
for match in matches:
if pattern == self.config.patterns[0]: # Email
masked = self._mask_value(match, "email")
elif pattern == self.config.patterns[2]: # Credit card
masked = self._mask_value(match, "full")
else:
masked = self._mask_value(match, "partial")
result = result.replace(match, masked)
return result
def desensitize(self, data: Union[Dict, str, List]) -> Union[Dict, str, List]:
"""Entry point cho desensitization"""
if isinstance(data, str):
try:
parsed = json.loads(data)
processed = self._process_dict(parsed) if isinstance(parsed, dict) else self._process_list(parsed)
return json.dumps(processed, ensure_ascii=False)
except json.JSONDecodeError:
return self._apply_pattern_mask(data)
elif isinstance(data, dict):
return self._process_dict(data)
elif isinstance(data, list):
return self._process_list(data)
return data
Test engine
engine = DesensitizationEngine(config)
test_data = {
"user": {
"email": "[email protected]",
"phone": "+84-912-345-678",
"password": "super_secret_pass_123",
"credit_card": "4532-1234-5678-9012",
"orders": [
{"order_id": "ORD001", "total": 1500000}
]
},
"api_key": "sk-abc123xyz789def456ghi012jkl345mno678",
"message": "Liên hệ với [email protected] để xác nhận"
}
result = engine.desensitize(test_data)
print("Input:", json.dumps(test_data, indent=2, ensure_ascii=False))
print("\nDesensitized:", json.dumps(result, indent=2, ensure_ascii=False))
3. Tích hợp với HolySheep AI API
import requests
import time
from datetime import datetime
from typing import Optional, Dict, Any
class HolySheepAIClient:
"""Client tích hợp HolySheep AI với desensitization logging"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, desensitization_engine: DesensitizationEngine):
self.api_key = api_key
self.engine = desensitization_engine
self.request_log = []
self.response_log = []
self._setup_logging()
def _setup_logging(self):
"""Thiết lập file logging với desensitization"""
self.file_handler = logging.FileHandler(
f"api_logs_{datetime.now().strftime('%Y%m%d')}.jsonl"
)
self.file_handler.setFormatter(
logging.Formatter('%(message)s')
)
logger.addHandler(self.file_handler)
logger.info("=== HolySheep AI Logging Session Started ===")
def _log_request(self, endpoint: str, payload: Dict):
"""Log request đã được desensitize"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"type": "request",
"endpoint": endpoint,
"data": self.engine.desensitize(payload)
}
self.request_log.append(log_entry)
logger.info(json.dumps(log_entry, ensure_ascii=False))
def _log_response(self, endpoint: str, status: int, response: Any, duration_ms: float):
"""Log response đã được desensitize"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"type": "response",
"endpoint": endpoint,
"status": status,
"duration_ms": round(duration_ms, 2),
"data": self.engine.desensitize(response) if isinstance(response, dict) else str(response)
}
self.response_log.append(log_entry)
logger.info(json.dumps(log_entry, ensure_ascii=False))
def chat_completions(self, messages: list, model: str = "gpt-4.1",
temperature: float = 0.7, max_tokens: int = 1000) -> Dict:
"""Gọi API chat completions với logging tự động"""
endpoint = f"{self.BASE_URL}/chat/completions"
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# Log request
self._log_request("/chat/completions", payload)
start_time = time.time()
try:
response = requests.post(
endpoint,
headers=headers,
json=payload,
timeout=30
)
duration_ms = (time.time() - start_time) * 1000
# Log response
self._log_response(
"/chat/completions",
response.status_code,
response.json() if response.ok else {"error": response.text},
duration_ms
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
duration_ms = (time.time() - start_time) * 1000
self._log_response("/chat/completions", 500, {"error": str(e)}, duration_ms)
raise
def embeddings(self, input_text: str, model: str = "text-embedding-3-small") -> Dict:
"""Tạo embeddings với logging"""
endpoint = f"{self.BASE_URL}/embeddings"
payload = {
"model": model,
"input": input_text
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
self._log_request("/embeddings", payload)
start_time = time.time()
try:
response = requests.post(endpoint, headers=headers, json=payload, timeout=30)
duration_ms = (time.time() - start_time) * 1000
self._log_response("/embeddings", response.status_code, response.json(), duration_ms)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
duration_ms = (time.time() - start_time) * 1000
self._log_response("/embeddings", 500, {"error": str(e)}, duration_ms)
raise
Sử dụng client
desensitization_engine = DesensitizationEngine(config)
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY", # Thay bằng API key thực tế
desensitization_engine=desensitization_engine
)
Ví dụ: Gọi API với thông tin nhạy cảm
messages = [
{"role": "system", "content": "Bạn là trợ lý hỗ trợ khách hàng thương mại điện tử."},
{"role": "user", "content": "Tôi là Nguyễn Văn A, email a***@gmail.com, SDT ****1234. Tôi cần hỗ trợ về đơn hàng ORD-2024-001."}
]
try:
# response = client.chat_completions(messages, model="gpt-4.1")
# print("Response:", response)
print("✅ Client configured. Uncomment API call to test.")
except Exception as e:
print(f"❌ Error: {e}")
Bảng so sánh chi phí: HolySheep vs OpenAI
| Model | OpenAI (Input) | OpenAI (Output) | HolySheep | Tiết kiệm |
|-------|----------------|-----------------|-----------|-----------|
| GPT-4.1 | $15/MTok | $60/MTok | $8/MTok | **85%+** |
| Claude Sonnet 4.5 | $3/MTok | $15/MTok | $15/MTok | Chi phí tương đương |
| Gemini 2.5 Flash | $0.35/MTok | $1.05/MTok | $2.50/MTok | Không khuyến nghị |
| DeepSeek V3.2 | - | - | $0.42/MTok | **Giá rẻ nhất** |
Với tỷ giá ¥1 = $1 và hỗ trợ WeChat/Alipay,
đăng ký tại đây để nhận tín dụng miễn phí khi bắt đầu.
Regex patterns nâng cao cho desensitization
import re
from typing import List, Tuple
class AdvancedPatternMatcher:
"""Advanced pattern matching cho các trường hợp phức tạp"""
PATTERNS: List[Tuple[str, re.Pattern]] = [
# Vietnamese phone numbers
("vi_phone", re.compile(
r'(?:(?:(?:\+|00)84)|0)(?:1[6-9]|2[0-9]|3[2-9]|5[6-9]|7[0-9]|8[1-9]|9[0-9])'
r'[0-9]{7}'
)),
# International phone with country codes
("intl_phone", re.compile(
r'\+?[1-9]\d{0,2}[-.\s]?\(?\d{1,4}\)?[-.\s]?\d{1,4}[-.\s]?\d{1,9}'
)),
# Vietnamese ID card
("vietnam_id", re.compile(
r'\b[0-9]{9,12}\b' # 9 or 12 digit IDs
)),
# Bank account numbers (Vietnamese format)
("bank_account", re.compile(
r'\b\d{8,20}\b(?=.*(?:bank|tài khoản|vietin|vietcom|bidv|acb|tpbank))',
re.IGNORECASE
)),
# IP addresses
("ip_address", re.compile(
r'\b(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}'
r'(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\b'
)),
# MAC addresses
("mac_address", re.compile(
r'\b(?:[0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa
Tài nguyên liên quan
Bài viết liên quan