ในโลกของ AI API นั้น การบันทึกล็อก (Logging) เป็นสิ่งจำเป็นอย่างยิ่งสำหรับการ Debug และ Monitoring ระบบ แต่หลายคนอาจไม่ทราบว่า ล็อกเหล่านี้อาจมีข้อมูลส่วนบุคคล (PII) ที่อยู่ภายใต้กฎหมาย GDPR ซึ่งหากจัดการไม่ถูกต้อง อาจนำไปสู่บทลงโทษที่รุนแรงได้ ในบทความนี้ เราจะมาดูวิธีการจัดการ AI API Logs ให้สอดคล้องกับ GDPR พร้อมตัวอย่างโค้ดที่ใช้งานได้จริง
เหตุการณ์จริง: ปัญหาที่เกิดจากการบันทึกล็อกโดยไม่คำนึงถึง GDPR
สถานการณ์ที่เกิดขึ้นจริงกับทีมพัฒนาของผม: วันหนึ่งระบบเริ่มบันทึกล็อกข้อผิดพลาดที่มีข้อมูลผู้ใช้จำนวนมาก เช่น อีเมล ที่อยู่ และเบอร์โทรศัพท์ โดยไม่ได้ Masking หรือ Anonymize ก่อน ทันทีที่เห็นข้อความ ValueError: Invalid email format in request payload ที่มาพร้อมกับอีเมลจริงๆ ของผู้ใช้หลายพันราย ผมตระหนักว่านี่คือความผิดพลาดร้ายแรงที่ต้องแก้ไขโดยด่วน เพราะหากถูก Audit จากหน่วยงานคุ้มครองข้อมูล บริษัทอาจถูกปรับได้ถึง 4% ของรายได้ทั่วโลก หรือสูงสุด 20 ล้านยูโร
GDPR พื้นฐานสำหรับ AI API Logs
GDPR (General Data Protection Regulation) กำหนดหลักการสำคัญสำหรับการประมวลผลข้อมูลส่วนบุคคล:
- Lawfulness, Fairness, and Transparency: ต้องมีฐานทางกฎหมายในการประมวลผลข้อมูล
- Purpose Limitation: ใช้ข้อมูลเฉพาะเจาะจงตามวัตถุประสงค์ที่แจ้งไว้
- Data Minimization: เก็บเฉพาะข้อมูลที่จำเป็นต่อวัตถุประสงค์
- Storage Limitation: ไม่เก็บข้อมูลนานเกินจำเป็น
- Integrity and Confidentiality: รักษาความปลอดภัยของข้อมูล
สำหรับ AI API Logs ที่ใช้ HolySheep AI เราต้องมั่นใจว่าข้อมูลที่ผ่าน API ไม่ถูกบันทึกในรูปแบบที่ระบุตัวตนผู้ใช้ได้โดยตรง
การตั้งค่า Secure Logging สำหรับ HolySheep AI API
ตัวอย่างโค้ดต่อไปนี้แสดงวิธีการสร้าง Logger ที่ปฏิบัติตาม GDPR โดยการ Masking PII อัตโนมัติ:
"""
GDPR-Compliant Logger for HolySheep AI API
Author: HolySheep AI Technical Team
"""
import re
import logging
import json
from datetime import datetime, timedelta
from typing import Any, Optional
class GDPRLogger:
"""
Logger ที่ออกแบบมาเพื่อปฏิบัติตาม GDPR
จะทำการ Masking ข้อมูลส่วนบุคคลอัตโนมัติ
"""
# Patterns สำหรับ PII Detection
PII_PATTERNS = {
'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
'phone': r'\b\d{9,15}\b',
'credit_card': r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',
'id_card': r'\b\d{13}\b',
'ip_address': r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b'
}
def __init__(self, service_name: str, retention_days: int = 30):
self.service_name = service_name
self.retention_days = retention_days
self.logger = logging.getLogger(f'gdpr_{service_name}')
self.logger.setLevel(logging.INFO)
# สร้าง Handler พร้อม JSON Formatting
handler = logging.StreamHandler()
handler.setFormatter(self._GDPRFormatter())
self.logger.addHandler(handler)
def _mask_pii(self, text: str) -> str:
"""Mask PII ในข้อความ"""
masked = text
# Mask Email: [email protected] -> u***@e***.com
email_pattern = re.compile(self.PII_PATTERNS['email'])
def mask_email(match):
email = match.group(0)
parts = email.split('@')
if len(parts) == 2:
local = parts[0]
domain = parts[1]
masked_local = local[0] + '***' if len(local) > 1 else '***'
domain_parts = domain.split('.')
masked_domain = domain_parts[0][0] + '***'
if len(domain_parts) > 1:
masked_domain += '.' + '.'.join(domain_parts[1:])
return f'{masked_local}@{masked_domain}'
return '[REDACTED_EMAIL]'
masked = email_pattern.sub(mask_email, masked)
# Mask Phone: 0812345678 -> ***2345678
phone_pattern = re.compile(self.PII_PATTERNS['phone'])
def mask_phone(match):
phone = match.group(0)
if len(phone) >= 9:
return '***' + phone[-7:]
return '***'
masked = phone_pattern.sub(mask_phone, masked)
# Mask Credit Card
cc_pattern = re.compile(self.PII_PATTERNS['credit_card'])
masked = cc_pattern.sub('[REDACTED_CREDIT_CARD]', masked)
# Mask ID Card (Thai)
id_pattern = re.compile(self.PII_PATTERNS['id_card'])
def mask_id(match):
id_num = match.group(0)
return id_num[:1] + '-' + '***' + id_num[-4:]
masked = id_pattern.sub(mask_id, masked)
# Mask IP Address: 192.168.1.1 -> ***.***.***.1
ip_pattern = re.compile(self.PII_PATTERNS['ip_address'])
masked = ip_pattern.sub('***.***.***.***', masked)
return masked
def log_api_request(self,
user_id: str,
endpoint: str,
payload: dict,
request_id: str):
"""บันทึก API Request โดยไม่มี PII"""
# ใช้ Pseudonymized User ID แทนข้อมูลจริง
safe_user_id = self._pseudonymize(user_id)
# Mask Payload
masked_payload = self._mask_pii(json.dumps(payload))
log_entry = {
'timestamp': datetime.utcnow().isoformat(),
'request_id': request_id,
'service': self.service_name,
'user_pseudonym': safe_user_id,
'endpoint': endpoint,
'payload_size': len(payload),
'gdpr_compliant': True
}
self.logger.info(f"API_REQUEST: {json.dumps(log_entry)}")
self.logger.debug(f"PAYLOAD: {masked_payload}")
def _pseudonymize(self, identifier: str) -> str:
"""สร้าง Pseudonym สำหรับ User ID"""
import hashlib
salt = 'HOLYSHEEP_SALT_2024'
combined = f"{identifier}{salt}".encode('utf-8')
return hashlib.sha256(combined).hexdigest()[:16]
class _GDPRFormatter(logging.Formatter):
"""Formatter ที่เพิ่มข้อมูล GDPR metadata"""
def format(self, record):
record.gdpr_timestamp = datetime.utcnow().isoformat() + 'Z'
record.retention_date = (datetime.utcnow() + timedelta(days=30)).isoformat() + 'Z'
return super().format(record)
การใช้งาน
logger = GDPRLogger('holysheep-chatbot', retention_days=30)
def call_holysheep_api(user_id: str, message: str, request_id: str):
"""ตัวอย่างการเรียก HolySheep API พร้อม Logging ที่ปลอดภัย"""
import requests
endpoint = 'https://api.holysheep.ai/v1/chat/completions'
headers = {
'Authorization': f'Bearer YOUR_HOLYSHEEP_API_KEY',
'Content-Type': 'application/json'
}
payload = {
'model': 'gpt-4.1',
'messages': [{'role': 'user', 'content': message}],
'temperature': 0.7
}
# Log Request ก่อนเรียก API
logger.log_api_request(user_id, endpoint, payload, request_id)
try:
response = requests.post(endpoint, headers=headers, json=payload, timeout=30)
if response.status_code == 200:
data = response.json()
# Log Response โดยไม่บันทึกข้อมูลผู้ใช้
logger.logger.info(f"API_SUCCESS: request_id={request_id}, model={data.get('model')}")
return data
elif response.status_code == 401:
logger.logger.error(f"AUTH_ERROR: Invalid API Key for request_id={request_id}")
raise Exception("Unauthorized: ตรวจสอบ API Key ของคุณ")
elif response.status_code == 429:
logger.logger.warning(f"RATE_LIMIT: request_id={request_id}")
raise Exception("Rate Limit Exceeded")
else:
logger.logger.error(f"API_ERROR: status={response.status_code}, request_id={request_id}")
raise Exception(f"API Error: {response.status_code}")
except requests.exceptions.Timeout:
logger.logger.error(f"TIMEOUT: request_id={request_id}")
raise Exception("Request Timeout: ลองอีกครั้ง")
except requests.exceptions.ConnectionError as e:
logger.logger.error(f"CONNECTION_ERROR: {str(e)}, request_id={request_id}")
raise Exception("Connection Error: ตรวจสอบอินเทอร์เน็ตของคุณ")การจัดการ Data Retention ตาม GDPR
GDPR กำหนดให้เราต้องลบข้อมูลเมื่อไม่จำเป็นต้องเก็บรักษาไว้อีกต่อไป นี่คือระบบ Data Retention ที่ช่วยให้คุณปฏิบัติตามกฎหมาย:
"""
GDPR Data Retention Manager สำหรับ HolySheep AI Logs
จัดการการลบข้อมูลอัตโนมัติตามกรอบเวลาที่กำหนด
"""
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import json
import sqlite3
from pathlib import Path
import logging
class GDPRDataRetention:
"""
จัดการ Data Retention ตาม GDPR Article 5(1)(e)
Storage Limitation: ไม่เก็บข้อมูลนานเกินจำเป็น
"""
# กำหนดระยะเวลาเก็บข้อมูลตามประเภท
RETENTION_POLICIES = {
'api_requests': 30, # 30 วัน
'error_logs': 90, # 90 วัน
'performance_metrics': 180, # 180 วัน
'audit_logs': 365, # 1 ปี
'billing_records': 2555 # 7 ปี (ตามกฎหมายบัญชี)
}
def __init__(self, db_path: str = 'gdpr_logs.db'):
self.db_path = db_path
self.logger = logging.getLogger('gdpr_retention')
self._init_database()
def _init_database(self):
"""สร้างตารางสำหรับเก็บ Logs"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS api_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
request_id TEXT UNIQUE NOT NULL,
timestamp DATETIME NOT NULL,
user_pseudonym TEXT NOT NULL,
endpoint TEXT NOT NULL,
method TEXT,
status_code INTEGER,
response_time_ms INTEGER,
model_used TEXT,
tokens_used INTEGER,
error_type TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS data_deletion_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
deletion_date DATETIME NOT NULL,
records_deleted INTEGER NOT NULL,
data_type TEXT NOT NULL,
retention_policy_days INTEGER NOT NULL,
legal_basis TEXT,
executed_by TEXT DEFAULT 'system'
)
''')
cursor.execute('''
CREATE INDEX IF NOT EXISTS idx_timestamp
ON api_logs(timestamp)
''')
conn.commit()
conn.close()
self.logger.info("GDPR Database initialized successfully")
def record_api_call(self,
request_id: str,
user_pseudonym: str,
endpoint: str,
method: str = 'POST',
status_code: Optional[int] = None,
response_time_ms: Optional[int] = None,
model_used: Optional[str] = None,
tokens_used: Optional[int] = None,
error_type: Optional[str] = None):
"""บันทึก API Call โดยใช้ Pseudonym"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT INTO api_logs
(request_id, timestamp, user_pseudonym, endpoint, method,
status_code, response_time_ms, model_used, tokens_used, error_type)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
request_id,
datetime.utcnow(),
user_pseudonym,
endpoint,
method,
status_code,
response_time_ms,
model_used,
tokens_used,
error_type
))
conn.commit()
conn.close()
def enforce_retention_policy(self, data_type: str = 'api_requests') -> Dict:
"""
ลบข้อมูลที่เก่ากว่าระยะเวลาที่กำหนด
ส่งคืนรายงานการลบข้อมูล
"""
if data_type not in self.RETENTION_POLICIES:
raise ValueError(f"Unknown data type: {data_type}")
retention_days = self.RETENTION_POLICIES[data_type]
cutoff_date = datetime.utcnow() - timedelta(days=retention_days)
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# นับจำนวน records ที่จะถูกลบ
cursor.execute('''
SELECT COUNT(*) FROM api_logs
WHERE timestamp < ?
''', (cutoff_date,))
records_to_delete = cursor.fetchone()[0]
# ลบข้อมูล
cursor.execute('''
DELETE FROM api_logs
WHERE timestamp < ?
''', (cutoff_date,))
deleted_count = cursor.rowcount
# บันทึก Log การลบ
cursor.execute('''
INSERT INTO data_deletion_log
(deletion_date, records_deleted, data_type, retention_policy_days, legal_basis)
VALUES (?, ?, ?, ?, ?)
''', (
datetime.utcnow(),
deleted_count,
data_type,
retention_days,
'GDPR Article 5(1)(e) - Storage Limitation'
))
conn.commit()
conn.close()
result = {
'data_type': data_type,
'retention_days': retention_days,
'cutoff_date': cutoff_date.isoformat(),
'records_deleted': deleted_count,
'deletion_date': datetime.utcnow().isoformat(),
'compliant': True
}
self.logger.info(f"Data retention enforced: {json.dumps(result)}")
return result
def generate_deletion_report(self, start_date: datetime, end_date: datetime) -> str:
"""สร้างรายงานการลบข้อมูลสำหรับ GDPR Compliance"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
SELECT deletion_date, records_deleted, data_type,
retention_policy_days, legal_basis
FROM data_deletion_log
WHERE deletion_date BETWEEN ? AND ?
ORDER BY deletion_date DESC
''', (start_date, end_date))
deletions = cursor.fetchall()
conn.close()
report_lines = [
"=" * 60,
"GDPR DATA DELETION REPORT",
f"Period: {start_date.date()} to {end_date.date()}",
"=" * 60,
""
]
for deletion in deletions:
report_lines.append(f"Date: {deletion[0]}")
report_lines.append(f"Records Deleted: {deletion[1]:,}")
report_lines.append(f"Data Type: {deletion[2]}")
report_lines.append(f"Retention Policy: {deletion[3]} days")
report_lines.append(f"Legal Basis: {deletion[4]}")
report_lines.append("-" * 40)
return "\n".join(report_lines)
def process_deletion_request(self, user_pseudonym: str) -> Dict:
"""
จัดการคำขอลบข้อมูล (Right to Erasure - Article 17)
ลบข้อมูลทั้งหมดของผู้ใช้รายนี้
"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# ตรวจสอบว่ามีข้อมูลของผู้ใช้นี้หรือไม่
cursor.execute('''
SELECT COUNT(*) FROM api_logs WHERE user_pseudonym = ?
''', (user_pseudonym,))
records_count = cursor.fetchone()[0]
if records_count == 0:
conn.close()
return {
'status': 'not_found',
'message': 'ไม่พบข้อมูลของผู้ใช้นี้ในระบบ',
'records_deleted': 0
}
# ลบข้อมูลทั้งหมด
cursor.execute('''
DELETE FROM api_logs WHERE user_pseudonym = ?
''', (user_pseudonym,))
deleted_count = cursor.rowcount
# บันทึก Log การลบ
cursor.execute('''
INSERT INTO data_deletion_log
(deletion_date, records_deleted, data_type, retention_policy_days, legal_basis)
VALUES (?, ?, ?, ?, ?)
''', (
datetime.utcnow(),
deleted_count,
'user_deletion_request',
0,
'GDPR Article 17 - Right to Erasure'
))
conn.commit()
conn.close()
return {
'status': 'success',
'user_pseudonym': user_pseudonym,
'records_deleted': deleted_count,
'deletion_date': datetime.utcnow().isoformat(),
'compliant': True
}
การใช้งาน
retention_manager = GDPRDataRetention()
บันทึก API Call ตัวอย่าง
retention_manager.record_api_call(
request_id='req_abc123xyz',
user_pseudonym='a1b2c3d4e5f6g7h8',
endpoint='https://api.holysheep.ai/v1/chat/completions',
method='POST',
status_code=200,
response_time_ms=45,
model_used='gpt-4.1',
tokens_used=150
)
ลบข้อมูลเก่าตาม Retention Policy
result = retention_manager.enforce_retention_policy('api_requests')
print(f"Retention enforced: {result}")
จัดการคำขอลบข้อมูลผู้ใช้
deletion_result = retention_manager.process_deletion_request('a1b2c3d4e5f6g7h8')
print(f"Deletion request processed: {deletion_result}")การตรวจสอบ PII ใน Payload ก่อนส่งไปยัง API
ก่อนที่จะส่ง Request ไปยัง HolySheep AI ควรตรวจสอบและทำความสะอาด Payload ก่อนเสมอ:
"""
PII Scanner และ Sanitizer สำหรับ AI API Requests
ป้องกันไม่ให้ข้อมูลส่วนบุคคลรั่วไหลไปยัง Logs
"""
import re
import json
from typing import Dict, Any, List, Tuple
from dataclasses import dataclass
from enum import Enum
class PIIType(Enum):
EMAIL = 'email'
PHONE = 'phone'
CREDIT_CARD = 'credit_card'
NATIONAL_ID = 'national_id'
PASSPORT = 'passport'
IP_ADDRESS = 'ip_address'
NAME = 'name'
ADDRESS = 'address'
DATE_OF_BIRTH = 'date_of_birth'
@dataclass
class PIIMatch:
field_path: str
pii_type: PIIType
original_value: str
masked_value: str
confidence: float
class PIIScanner:
"""สแกนและ Mask PII ใน Request Payload"""
# Regex Patterns
PATTERNS = {
PIIType.EMAIL: r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
PIIType.PHONE: r'\b(?:\+?66|0)\d{8,9}\b',
PIIType.CREDIT_CARD: r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',
PIIType.NATIONAL_ID: r'\b\d{13}\b',
PIIType.IP_ADDRESS: r'\b(?:\d{1,3}\.){3}\d{1,3}\b',
PIIType.DATE_OF_BIRTH: r'\b(?:0?[1-9]|[12]\d|3[01])[\/\-](?:0?[1-9]|1[012])[\/\-](?:19|20)\d{2}\b'
}
# Sensitive Field Names
SENSITIVE_FIELDS = {
'password', 'passwd', 'secret', 'token', 'api_key', 'apikey',
'ssn', 'social_security', 'credit_card', 'card_number', 'cvv',
'name', 'fullname', 'first_name', 'last_name', 'email',
'phone', 'mobile', 'address', 'dob', 'date_of_birth'
}
def __init__(self, strict_mode: bool = True):
self.strict_mode = strict_mode
self.matches: List[PIIMatch] = []
def scan_value(self, value: str) -> List[PIIMatch]:
"""สแกนหา PII ในค่าเดียว"""
matches = []
for pii_type, pattern in self.PATTERNS.items():
for match in re.finditer(pattern, value):
masked = self._mask_pii(match.group(0), pii_type)
matches.append(PIIMatch(
field_path='value',
pii_type=pii_type,
original_value=match.group(0),
masked_value=masked,
confidence=1.0
))
return matches
def scan_payload(self, payload: Dict[str, Any], path: str = '') -> Tuple[Dict, List[PIIMatch]]:
"""
สแกนและ Mask Payload ทั้งหมด
ส่งคืน Payload ที่ถูก Mask แล้วพร้อมรายการ PII ที่พบ
"""
self.matches = []
masked_payload = self._process_dict(payload, path)
return masked_payload, self.matches
def _process_dict(self, obj: Dict[str, Any], path: str) -> Dict:
"""ประมวลผล Dictionary แบบ Recursive"""
result = {}
for key, value in obj.items():
current_path = f"{path}.{key}" if path else key
key_lower = key.lower()
if isinstance(value, str):
# ตรวจสอบว่า Key เป็น Sensitive Field หรือไม่
if key_lower in self.SENSITIVE_FIELDS:
masked_value = '[REDACTED]'
self.matches.append(PIIMatch(
field_path=current_path,
pii_type=PIIType.EMAIL if 'email' in key_lower else PIIType.EMAIL,
original_value=value,
masked_value=masked_value,
confidence=0.95
))
result[key] = masked_value
else:
# สแกนหา PII ใน Value
matches = self.scan_value(value)
if matches:
for match in matches:
match.field_path = current_path
self.matches.append(match)
result[key] = self._apply_mask(value, matches)
else:
result[key] = value
elif isinstance(value, dict):
result[key] = self._process_dict(value, current_path)
elif isinstance(value, list):
result[key] = self._process_list(value, current_path)
else:
result[key] = value
return result
def _process_list(self, obj: List, path: str) -> List:
"""ประมวลผล List แบบ Recursive"""
result = []
for i, item in enumerate(obj):
current_path = f"{path}[{i}]"
if isinstance(item, dict):
result.append(self._process_dict(item, current_path))
elif isinstance(item, str):
matches = self.scan_value(item)
if matches:
for match in matches:
match.field_path = current_path
self.matches.append(match)
result.append(self._apply_mask(item, matches))
else:
result.append(item)
else:
result.append(item)
return result
def _mask_pii(self, value: str, pii_type: PIIType) -> str:
"""Mask PII ตามประเภท"""
if pii_type == PIIType.EMAIL:
parts = value.split('@')
if len(parts) == 2:
return f"{parts[0][:2]}***@{parts[1]}"
elif pii_type == PIIType.PHONE:
if len(value) >= 9:
return f"***{value[-7:]}"
elif pii_type == PIIType.CREDIT_CARD:
return "****-****-****-****"
elif pii_type == PIIType.NATIONAL_ID:
return f"{value[:1]}-****-****-**-{value[-4:]}"
elif pii_type == PIIType.IP_ADDRESS:
return "***.***.***.***"
return '[MASKED]'
def _apply_mask(self, value: str, matches: List[PIIMatch]) -> str:
"""Apply Mask โดยเริ่มจาก Match ที่ยาวที่สุดก่อน"""
# เรียงลำดับจากยาวไปสั้น เพื่อไม่ให้การแทนที่ซ้อนทับกัน
sorted_matches = sorted(matches, key=lambda x: len(x.original_value), reverse=True)
result = value
for match in sorted_matches:
result = result.replace(match.original_value, match.masked_value)
return result
def is_compliant(self) -> bool:
"""ตรวจสอบว่า Payload ปลอดภัยหรือไม่"""
if self.strict_mode:
return len(self.matches) == 0
return True
def generate_report(self) -> str:
"""สร้างรายงาน PII ที่พบ"""
if not self.matches:
return "✅ ไม่พบ PII ใน Payload"
lines = [
f"⚠️ พบ PII {len(self.matches)} รายการ:",
"-" * 50
]
for i, match in enumerate(self.matches, 1):
lines.append(f"{i}. Path: {match.field_path}")
lines.append(f" Type: {match.pii_type.value}")
lines.append(f" Original: {match.original_value}")
lines.append(f" Masked: {match.masked_value}")
lines.append(f" Confidence: {match.confidence:.0%}")
return "\n".join(lines)
def sanitize_payload_for_api(payload: Dict, api_endpoint: str) -> Tuple[Dict, Dict]:
"""
ฟังก์ชันหลักสำ