ในยุคที่ Edge AI กำลังเติบโตอย่างรวดเร็ว การรักษาความปลอดภัยโมเดล AI ในสภาพแวดล้อมที่ไม่มีการเชื่อมต่ออินเทอร์เน็ตกลายเป็นความท้าทายสำคัญสำหรับวิศวกร ในบทความนี้ผมจะแชร์ประสบการณ์ตรงจากการ deploy ระบบ Edge AI สำหรับโรงงานอุตสาหกรรมที่ต้องการความปลอดภัยระดับสูงสุด
สถาปัตยกรรม Edge AI Security พื้นฐาน
สถาปัตยกรรมที่แนะนำสำหรับ Edge AI ที่ปลอดภัยประกอบด้วย 3 ชั้นหลัก ได้แก่ Hardware Security Module (HSM) สำหรับจัดเก็บคีย์, Secure Boot สำหรับยืนยันความถูกต้องของโค้ด และ Encrypted Model Storage สำหรับป้องกันการขโมยโมเดล
การติดตั้ง Secure Boot สำหรับ Edge Device
การตั้งค่า Secure Boot บน Raspberry Pi หรือ NVIDIA Jetson ต้องทำผ่าน TPM (Trusted Platform Module) เพื่อให้มั่นใจว่าเฉพาะโค้ดที่ผ่านการ sign ด้วยคีย์ที่เชื่อถือได้เท่านั้นที่จะถูก execute
#!/bin/bash
Secure Boot Setup Script สำหรับ ARM64 Edge Devices
set -e
ติดตั้ง TPM 2.0 tools
apt-get update && apt-get install -y \
tpm2-tools \
tpm2-pkcs11 \
trousers
Generate secure boot keys
mkdir -p /etc/secureboot/keys
cd /etc/secureboot/keys
Platform Key (PK) - สูงสุด
tpm2_createprimary -C o -g sha256 -G rsa -c pk.ctx
tpm2_evictcontrol -C o -c pk.ctx 0x81000001
Key Exchange Key (KEK)
tpm2_createprimary -C o -g sha256 -G rsa -c kek.ctx
tpm2_evictcontrol -C o -c kek.ctx 0x81000002
Signature Database (db) - ไดรเวอร์และ UEFI apps ที่อนุญาต
openssl req -new -x509 -sha256 -newkey rsa:4096 \
-keyout db.key -out db.crt -days 3650 -nodes \
-subj "/CN=EdgeAI-SecureBoot-db"
tpm2_evictcontrol -C o -c db.crt 0x81000003
echo "Secure Boot Keys Generated Successfully"
echo "Next: Enroll keys in UEFI firmware"Encrypted Model Storage ด้วย LUKS2 และ Model Encryption
การเข้ารหัสโมเดล AI ต้องทำทั้ง at-rest และ in-use encryption เพื่อป้องกันการ reverse engineering
#!/usr/bin/env python3
"""
Edge AI Model Encryption Module
รองรับ AES-256-GCM + Argon2id key derivation
"""
import os
import hashlib
import hmac
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.argon2 import Argon2id
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.backends import default_backend
import json
class EdgeModelEncryptor:
def __init__(self, salt: bytes = None):
self.salt = salt or os.urandom(16)
self.nonce = os.urandom(12)
def derive_key(self, password: str, memory_cost: int = 65536) -> bytes:
"""Argon2id key derivation - ทนต่อ GPU/ASIC attacks"""
kdf = Argon2id(
length=32,
salt=self.salt,
iterations=3,
memory_cost=memory_cost, # 64MB memory
parallelism=4
)
return kdf.derive(password.encode('utf-8'))
def encrypt_model(self, model_path: str, password: str) -> bytes:
"""เข้ารหัสโมเดลด้วย AES-256-GCM"""
with open(model_path, 'rb') as f:
model_data = f.read()
key = self.derive_key(password)
aesgcm = AESGCM(key)
# เพิ่ม metadata header (model hash, version, timestamp)
metadata = {
'hash': hashlib.sha256(model_data).hexdigest(),
'version': '1.0',
'encrypted_at': str(int(__import__('time').time()))
}
header = json.dumps(metadata).encode('utf-8')
header_len = len(header).to_bytes(4, 'big')
# Encrypt: header + model data
plaintext = header_len + header + model_data
ciphertext = aesgcm.encrypt(self.nonce, plaintext, None)
return self.salt + self.nonce + ciphertext
def decrypt_model(self, encrypted_data: bytes, password: str) -> tuple:
"""ถอดรหัสโมเดลและตรวจสอบ integrity"""
salt = encrypted_data[:16]
nonce = encrypted_data[16:28]
ciphertext = encrypted_data[28:]
kdf = Argon2id(
length=32, salt=salt, iterations=3,
memory_cost=65536, parallelism=4
)
key = kdf.derive(password.encode('utf-8'))
aesgcm = AESGCM(key)
try:
plaintext = aesgcm.decrypt(nonce, ciphertext, None)
except Exception as e:
raise ValueError(f"Decryption failed: {e}")
# Parse header
header_len = int.from_bytes(plaintext[:4], 'big')
header = json.loads(plaintext[4:4+header_len])
model_data = plaintext[4+header_len:]
# Verify integrity
if hashlib.sha256(model_data).hexdigest() != header['hash']:
raise ValueError("Model integrity check failed!")
return model_data, header
Usage example
if __name__ == "__main__":
encryptor = EdgeModelEncryptor()
# เข้ารหัสโมเดล
encrypted = encryptor.encrypt_model(
'/models/yolov8n.onnx',
'SecureP@ssw0rd!2024'
)
with open('/secure-models/yolov8n.enc', 'wb') as f:
f.write(encrypted)
# ถอดรหัสเมื่อ load เข้า inference engine
decrypted, meta = encryptor.decrypt_model(encrypted, 'SecureP@ssw0rd!2024')
print(f"Model version: {meta['version']}, Hash: {meta['hash'][:16]}...")Over-The-Air Model Update ผ่าน Signed Package
การอัปเดตโมเดลใน Edge device แบบ offline ต้องมีการตรวจสอบความถูกต้องของ package ก่อน deploy ทุกครั้ง
#!/usr/bin/env python3
"""
Secure OTA Model Update System
รองรับ Ed25519 signature verification และ rollback protection
"""
import os
import hashlib
import tempfile
import shutil
from pathlib import Path
from nacl.signing import SigningKey
from nacl.encoding import RawEncoder
import json
import tarfile
import gzip
class SecureOTAUpdater:
SIGNATURE_BYTES = 64
HASH_BYTES = 32
def __init__(self, device_id: str, verify_key: bytes):
self.device_id = device_id
self.verify_key = verify_key
self.update_dir = Path(f"/var/edgeai/updates/{device_id}")
self.update_dir.mkdir(parents=True, exist_ok=True)
def create_signed_package(self, model_dir: str, output_path: str,
signing_key: bytes, version: str):
"""สร้าง signed update package"""
# คำนวณ hash ของไฟล์ทั้งหมดในโฟลเดอร์โมเดล
file_hashes = {}
model_path = Path(model_dir)
for file in sorted(model_path.rglob('*')):
if file.is_file():
with open(file, 'rb') as f:
content = f.read()
file_hash = hashlib.sha256(content).digest()
rel_path = file.relative_to(model_path).as_posix()
file_hashes[rel_path] = file_hash.hex()
# Metadata
manifest = {
'version': version,
'device_id': self.device_id,
'files': file_hashes,
'timestamp': int(__import__('time').time()),
'min_boot_version': '2.1.0'
}
# Create archive
with tempfile.NamedTemporaryFile(suffix='.tar', delete=False) as tmp:
archive_path = tmp.name
with tarfile.open(archive_path, 'w') as tar:
tar.add(model_path, arcname='model')
manifest_bytes = json.dumps(manifest).encode('utf-8')
import io
manifest_info = tarfile.TarInfo(name='MANIFEST.json')
manifest_info.size = len(manifest_bytes)
tar.addfile(manifest_info, io.BytesIO(manifest_bytes))
# Compress
compressed_path = output_path + '.gz'
with open(archive_path, 'rb') as f_in:
with gzip.open(compressed_path, 'wb') as f_out:
shutil.copyfileobj(f_in, f_out)
# Sign
package_hash = self._hash_file(compressed_path)
signing = SigningKey(signing_key, encoder=RawEncoder)
signature = signing.sign(package_hash).signature
# Final package: signature + compressed data
with open(output_path, 'wb') as f:
f.write(signature)
with open(compressed_path, 'rb') as cf:
f.write(cf.read())
os.unlink(archive_path)
os.unlink(compressed_path)
print(f"Created signed package: {output_path}")
print(f"Package hash: {package_hash.hex()[:32]}...")
return output_path
def verify_and_install(self, package_path: str, rollback_version: str):
"""ตรวจสอบ signature และ install update"""
with open(package_path, 'rb') as f:
signature = f.read(self.SIGNATURE_BYTES)
compressed_data = f.read()
# Verify signature
package_hash = hashlib.sha256(compressed_data).digest()
from nacl.signing import VerifyKey
from nacl.exceptions import BadSignatureError
verify = VerifyKey(self.verify_key, encoder=RawEncoder)
try:
verify.verify(package_hash, signature)
print("✓ Signature verified")
except BadSignatureError:
raise SecurityError("Package signature invalid!")
# Decompress
with tempfile.NamedTemporaryFile(suffix='.tar', delete=False) as tmp:
extract_path = tmp.name
with gzip.GzipFile(fileobj=__import__('io').BytesIO(compressed_data)) as gz:
with open(extract_path, 'wb') as f:
shutil.copyfileobj(gz, f)
# Extract และตรวจสอบ manifest
with tarfile.open(extract_path, 'r') as tar:
# อ่าน manifest ก่อน
manifest_file = tar.extractfile('MANIFEST.json')
manifest = json.loads(manifest_file.read())
# ตรวจสอบ version และ device_id
if manifest['device_id'] != self.device_id:
raise SecurityError("Wrong device target!")
if manifest['version'] <= rollback_version:
raise ValueError(f"Version {manifest['version']} <= current {rollback_version}")
# ตรวจสอบ file hashes
for member in tar.getmembers():
if member.name == 'MANIFEST.json':
continue
f = tar.extractfile(member)
content = f.read()
expected = manifest['files'].get(member.name.replace('model/', ''))
actual = hashlib.sha256(content).hexdigest()
if expected != actual:
raise SecurityError(f"File corrupted: {member.name}")
# Backup current model
model_dir = Path(f"/var/edgeai/models/{self.device_id}")
backup_dir = model_dir.parent / f"backup_{rollback_version}"
if model_dir.exists():
shutil.move(str(model_dir), str(backup_dir))
# Extract new model
tar.extractall(self.update_dir / 'staging')
os.unlink(extract_path)
print(f"✓ Update {manifest['version']} verified and staged")
return manifest['version']
Generate keys (ควรทำ offline และเก็บ private key อย่างปลอดภัย)
def generate_keys():
signing_key = SigningKey.generate()
verify_key = signing_key.verify_key
return signing_key.encode(RawEncoder), verify_key.encode(RawEncoder)
if __name__ == "__main__":
# Demo
sk, vk = generate_keys()
updater = SecureOTAUpdater(
device_id="factory-line-01",
verify_key=vk
)
# Create signed package (ใน production ทำบน secure server)
updater.create_signed_package(
model_dir="/tmp/new_model",
output_path="/var/edgeai/releases/v1.2.0.signed",
signing_key=sk,
version="1.2.0"
)การเชื่อมต่อ HolySheep AI API สำหรับ Edge Model Management
สำหรับการจัดการโมเดลจากระยะไกล สามารถใช้ HolySheep AI เพื่อ sync metadata และ monitor สถานะ Edge devices ได้อย่างปลอดภัย ด้วย latency ต่ำกว่า 50ms และราคาที่ประหยัดกว่า 85% เมื่อเทียบกับบริการอื่น
#!/usr/bin/env python3
"""
Edge Device Model Manager - HolySheep AI Integration
จัดการโมเดลและ sync สถานะกับ HolySheep API
"""
import requests
import hashlib
import json
from datetime import datetime, timedelta
from typing import Optional, Dict, List
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor, as_completed
import asyncio
@dataclass
class ModelMetadata:
model_id: str
version: str
checksum: str
size_bytes: int
deployed_at: Optional[str] = None
status: str = "pending"
class HolySheepEdgeManager:
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, device_id: str):
self.api_key = api_key
self.device_id = device_id
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
})
def register_device(self, hardware_info: Dict) -> Dict:
"""ลงทะเบียน Edge device กับ HolySheep"""
response = self.session.post(
f"{self.BASE_URL}/devices/register",
json={
'device_id': self.device_id,
'hardware': hardware_info,
'security_level': 'high',
'offline_capable': True
}
)
response.raise_for_status()
return response.json()
def get_model_updates(self, current_version: str) -> List[Dict]:
"""ตรวจสอบโมเดลใหม่จาก HolySheep"""
response = self.session.get(
f"{self.BASE_URL}/models/updates",
params={
'device_id': self.device_id,
'current_version': current_version,
'check_signature': True
}
)
response.raise_for_status()
return response.json().get('updates', [])
def report_model_status(self, model_id: str, status: str,
metrics: Optional[Dict] = None) -> Dict:
"""รายงานสถานะโมเดลไปยัง HolySheep"""
payload = {
'device_id': self.device_id,
'model_id': model_id,
'status': status,
'timestamp': datetime.utcnow().isoformat(),
'metrics': metrics or {}
}
response = self.session.post(
f"{self.BASE_URL}/models/status",
json=payload
)
response.raise_for_status()
return response.json()
def sync_offline_logs(self, logs: List[Dict]) -> Dict:
"""Sync logs ที่เก็บไว้ตอน offline"""
response = self.session.post(
f"{self.BASE_URL}/devices/sync",
json={
'device_id': self.device_id,
'logs': logs,
'sync_timestamp': datetime.utcnow().isoformat()
}
)
response.raise_for_status()
return response.json()
def download_model_with_retry(self, model_id: str,
max_retries: int = 3) -> bytes:
"""ดาวน์โหลดโมเดลพร้อม retry logic"""
for attempt in range(max_retries):
try:
response = self.session.get(
f"{self.BASE_URL}/models/download/{model_id}",
stream=True,
timeout=300
)
response.raise_for_status()
data = b''
for chunk in response.iter_content(chunk_size=8192):
data += chunk
# Verify