การปฏิบัติตามข้อกำหนด: GDPR และ SOC2

เหตุใดข้อกำหนดจึงใช้กับเอเจนต์ปัญญาประดิษฐ์

GDPR: การลดทอนข้อมูล

def minimized_log_entry(conversation: list[dict], user_id: str) -> dict:
    '''Store only what is needed — not the full transcript.'''
    return {
        'user_id':       user_id,
        'timestamp':     conversation[-1].get('timestamp'),
        'turns':         len(conversation),
        'resolved':      conversation[-1].get('resolved', False),
        'intent':        conversation[-1].get('intent'),
        'satisfaction':  conversation[-1].get('csat_score'),
        # NOT stored: full message text, PII mentioned in conversation
    }

# WRONG — stores raw messages which may contain PII
# {'messages': conversation, 'user_id': user_id}

if __name__ == '__main__':
    conversation = [
        {'timestamp': 1717000000, 'resolved': True, 'intent': 'reset_password', 'csat_score': 5}
    ]
    print('Minimized log entry:', minimized_log_entry(conversation, user_id='u42'))

GDPR: การติดตามความยินยอม

from datetime import datetime, timezone

consent_store: dict[str, dict] = {}  # user_id -> consent record

def record_consent(user_id: str, purpose: str, version: str = '1.0'):
    if user_id not in consent_store:
        consent_store[user_id] = {'purposes': {}}
    consent_store[user_id]['purposes'][purpose] = {
        'granted':     True,
        'timestamp':   datetime.now(timezone.utc).isoformat(),
        'version':     version
    }

def withdraw_consent(user_id: str, purpose: str):
    if user_id in consent_store:
        consent_store[user_id]['purposes'].pop(purpose, None)

def has_consent(user_id: str, purpose: str) -> bool:
    return bool(
        consent_store.get(user_id, {}).get('purposes', {}).get(purpose, {}).get('granted')
    )

if __name__ == '__main__':
    record_consent('u42', 'marketing_emails')
    print('Has consent:', has_consent('u42', 'marketing_emails'))
    withdraw_consent('u42', 'marketing_emails')
    print('Has consent after withdrawal:', has_consent('u42', 'marketing_emails'))

GDPR: สิทธิในการลบ (สิทธิที่จะถูกลืม)

import os

def erase_user_data(user_id: str, conn) -> dict:
    '''Delete all personal data for a user across all storage systems.'''
    deleted = {}

    # 1. Delete from audit log (if not using WORM storage)
    with conn.cursor() as cur:
        cur.execute('DELETE FROM agent_audit_log WHERE user_id = %s', (user_id,))
        deleted['audit_log_rows'] = cur.rowcount

    # 2. Delete conversation history
    with conn.cursor() as cur:
        cur.execute('DELETE FROM conversation_history WHERE user_id = %s', (user_id,))
        deleted['conversations'] = cur.rowcount

    # 3. Remove consent record
    consent_store.pop(user_id, None)
    deleted['consent_record'] = 1

    # 4. Clear cached profile
    profile_cache = {}  # reference to your cache
    profile_cache.pop(user_id, None)
    deleted['profile_cache'] = 1

    conn.commit()
    return {'user_id': user_id, 'erased': deleted}

GDPR: การเคลื่อนย้ายข้อมูล

import json

def export_user_data(user_id: str, conn) -> str:
    export = {'user_id': user_id, 'data': {}}

    with conn.cursor() as cur:
        cur.execute(
            'SELECT timestamp, action_type, result FROM agent_audit_log '
            'WHERE user_id = %s ORDER BY timestamp',
            (user_id,)
        )
        rows = cur.fetchall()
        export['data']['audit_log'] = [
            {'timestamp': r[0], 'action': r[1], 'result': r[2]}
            for r in rows
        ]

    export['data']['consent'] = consent_store.get(user_id, {})
    return json.dumps(export, indent=2)

SOC 2: การควบคุมการเข้าถึง

ROLE_PERMISSIONS = {
    'agent_basic':    {'web_search', 'read_file', 'query_database'},
    'agent_email':    {'web_search', 'read_file', 'query_database', 'send_email'},
    'agent_admin':    {'web_search', 'read_file', 'write_file', 'query_database',
                       'send_email', 'create_ticket', 'execute_python_sandbox'},
    'human_admin':    {'*'}  # all permissions
}

def get_permitted_actions(role: str) -> set:
    perms = ROLE_PERMISSIONS.get(role, set())
    if '*' in perms:
        return ALLOWED_ACTIONS  # reference to full allowlist
    return perms

def check_role_permission(role: str, action: str) -> bool:
    return action in get_permitted_actions(role)

SOC 2: การเข้ารหัสขณะจัดเก็บและระหว่างส่งข้อมูล

from cryptography.fernet import Fernet
import os

# Key stored in environment variable, never in code
ENCRYPTION_KEY = os.environ.get('LOG_ENCRYPTION_KEY', Fernet.generate_key())
cipher = Fernet(ENCRYPTION_KEY)

def encrypt_sensitive_field(value: str) -> str:
    return cipher.encrypt(value.encode()).decode()

def decrypt_sensitive_field(encrypted: str) -> str:
    return cipher.decrypt(encrypted.encode()).decode()

# Usage: encrypt PII before storing in audit log
encrypted_email = encrypt_sensitive_field('user@example.com')
print(encrypted_email[:30], '...')  # starts with gAAAA...

SOC 2: บันทึกการตรวจสอบเหตุการณ์ด้านความปลอดภัย

import logging, json, time

security_logger = logging.getLogger('agent.security')

SECURITY_EVENTS = {
    'auth_success', 'auth_failure', 'policy_violation',
    'data_access_pii', 'admin_action', 'config_change'
}

def log_security_event(event_type: str, agent_id: str,
                       user_id: str, details: dict):
    assert event_type in SECURITY_EVENTS, f'Unknown security event: {event_type}'
    entry = {
        'timestamp':  time.time(),
        'event_type': event_type,
        'agent_id':   agent_id,
        'user_id':    user_id,
        'details':    details
    }
    security_logger.warning(json.dumps(entry))

if __name__ == '__main__':
    import sys
    security_logger.addHandler(logging.StreamHandler(sys.stdout))
    log_security_event('auth_failure', 'agent-1', 'user-42', {'reason': 'bad_token'})

การติดตาม PII ในผลลัพธ์ของเอเจนต์

import re

PII_PATTERNS = [
    (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'email'),
    (r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',                         'phone'),
    (r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',           'credit_card'),
    (r'\b\d{3}-\d{2}-\d{4}\b',                                   'ssn')
]

def redact_pii(text: str) -> tuple[str, list[str]]:
    redacted = text
    found_types = []
    for pattern, pii_type in PII_PATTERNS:
        if re.search(pattern, redacted):
            found_types.append(pii_type)
            redacted = re.sub(pattern, f'[{pii_type.upper()}_REDACTED]', redacted)
    return redacted, found_types

text, types = redact_pii('Contact john@example.com or call 555-867-5309')
print(text)
print('PII found:', types)

การสร้างรายการตรวจสอบการปฏิบัติตามข้อกำหนด

COMPLIANCE_CHECKLIST = [
    ('GDPR',  'Consent captured before processing personal data'),
    ('GDPR',  'Data minimization: only necessary data is collected'),
    ('GDPR',  'Right-to-erasure endpoint implemented and tested'),
    ('GDPR',  'Data portability export available'),
    ('GDPR',  'Data retention policy enforced (auto-delete after N days)'),
    ('SOC2',  'Role-based access control implemented for all agent tools'),
    ('SOC2',  'All data encrypted at rest and in transit (TLS 1.2+)'),
    ('SOC2',  'Security events logged and retained for 12+ months'),
    ('SOC2',  'Access reviews scheduled quarterly'),
    ('SOC2',  'Incident response plan documented'),
    ('BOTH',  'PII redaction applied to logs and analytics exports'),
    ('BOTH',  'Vendor DPAs signed for all third-party APIs used'),
]

for framework, control in COMPLIANCE_CHECKLIST:
    print(f'[ ] [{framework}] {control}')

การทำงานอัตโนมัติเพื่อเก็บรักษาข้อมูล

from datetime import datetime, timedelta, timezone

RETENTION_POLICIES = {
    'conversation_history': 90,    # days
    'agent_audit_log':      365,
    'cached_profiles':      1,
    'session_data':         30
}

def purge_expired_data(conn):
    purged = {}
    for table, retention_days in RETENTION_POLICIES.items():
        cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
        try:
            with conn.cursor() as cur:
                cur.execute(
                    f'DELETE FROM {table} WHERE timestamp < %s',
                    (cutoff.timestamp(),)
                )
                purged[table] = cur.rowcount
        except Exception as e:
            purged[table] = f'Error: {e}'
    conn.commit()
    return purged

if __name__ == '__main__':
    class FakeCursor:
        def __enter__(self):
            return self
        def __exit__(self, *exc):
            return False
        def execute(self, sql, params):
            self.rowcount = 3

    class FakeConn:
        def cursor(self):
            return FakeCursor()
        def commit(self):
            pass

    result = purge_expired_data(FakeConn())
    print('Rows purged per table:')
    for table, count in result.items():
        print(f'  {table}: {count}')

หลักการลดข้อมูลให้น้อยที่สุดของ GDPR คืออะไร

ทบทวนการปฏิบัติตาม GDPR และ SOC 2