当前位置：首页 > news >正文

AI 应用安全最佳实践：保护数据和系统安全

news 2026/5/31 19:07:29

AI 应用安全最佳实践：保护数据和系统安全

前言

AI 应用的安全是一个综合性的话题，涉及数据保护、模型安全、API 安全等多个方面。随着 AI 应用的普及，安全问题变得越来越重要。

我在项目中负责过多个 AI 系统的安全设计，对安全风险和防护措施有深入理解。今天分享一些安全最佳实践。

数据安全

数据加密

from cryptography.fernet import Fernet from cryptography.hazmat.primitives.asymmetric import rsa, padding from cryptography.hazmat.primitives import serialization, hashes class DataSecurity: """数据安全""" def __init__(self): self.symmetric_key = Fernet.generate_key() self.cipher = Fernet(self.symmetric_key) def encrypt(self, data: str) -> bytes: """对称加密""" return self.cipher.encrypt(data.encode()) def decrypt(self, encrypted_data: bytes) -> str: """对称解密""" return self.cipher.decrypt(encrypted_data).decode() def generate_rsa_key_pair(self) -> tuple: """生成 RSA 密钥对""" private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) public_key = private_key.public_key() return private_key, public_key def rsa_encrypt(self, public_key, data: bytes) -> bytes: """RSA 加密""" return public_key.encrypt( data, padding.OAEP( mgf=padding.MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA256(), label=None ) )

数据脱敏

import re class DataAnonymizer: """数据脱敏器""" def __init__(self): self.patterns = [ (r"\b[\w.-]+@[\w.-]+\.\w+\b", "[邮箱]"), (r"\b\d{11}\b", "[手机号]"), (r"\b\d{3}-\d{2}-\d{4}\b", "[SSN]"), (r"\b\d{16}\b", "[卡号]") ] def anonymize(self, text: str) -> str: """脱敏文本""" result = text for pattern, replacement in self.patterns: result = re.sub(pattern, replacement, result) return result

模型安全

输入验证

class InputValidator: """输入验证器""" def __init__(self): self.max_length = 1000 self.blocked_patterns = [ "忽略之前的指令", "DAN 模式", "越狱模式" ] def validate(self, text: str) -> tuple: """验证输入""" # 长度检查 if len(text) > self.max_length: return False, "输入过长" # 内容检查 for pattern in self.blocked_patterns: if pattern.lower() in text.lower(): return False, "包含恶意内容" return True, "验证通过"

输出过滤

class OutputFilter: """输出过滤器""" def __init__(self): self.sensitive_topics = [ "暴力", "色情", "仇恨", "自杀", "诈骗" ] def filter(self, text: str) -> str: """过滤敏感内容""" result = text for topic in self.sensitive_topics: result = result.replace(topic, "[内容已过滤]") return result def check_safety(self, text: str) -> bool: """检查安全性""" for topic in self.sensitive_topics: if topic in text: return False return True

API 安全

认证与授权

from fastapi import FastAPI, Depends, HTTPException from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from jose import JWTError, jwt from passlib.context import CryptContext class APISecurity: """API 安全""" def __init__(self, secret_key: str, algorithm: str = "HS256"): self.secret_key = secret_key self.algorithm = algorithm self.pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") self.oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") def verify_password(self, plain_password: str, hashed_password: str) -> bool: """验证密码""" return self.pwd_context.verify(plain_password, hashed_password) def create_access_token(self, data: dict) -> str: """创建访问令牌""" to_encode = data.copy() encoded_jwt = jwt.encode(to_encode, self.secret_key, algorithm=self.algorithm) return encoded_jwt async def get_current_user(self, token: str = Depends(oauth2_scheme)): """获取当前用户""" credentials_exception = HTTPException( status_code=401, detail="无法验证凭据" ) try: payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm]) username: str = payload.get("sub") if username is None: raise credentials_exception except JWTError: raise credentials_exception return username

请求限流

from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded class RateLimiterMiddleware: """请求限流""" def __init__(self, app): self.limiter = Limiter(key_func=get_remote_address) app.state.limiter = self.limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) def limit(self, limit_str: str): """限制装饰器""" return self.limiter.limit(limit_str)

安全监控

import logging from pythonjsonlogger import jsonlogger class SecurityLogger: """安全日志""" def __init__(self): self.logger = logging.getLogger("security") self.logger.setLevel(logging.INFO) handler = logging.StreamHandler() formatter = jsonlogger.JsonFormatter( '%(asctime)s %(levelname)s %(message)s %(request_id)s' ) handler.setFormatter(formatter) self.logger.addHandler(handler) def log_security_event(self, event_type: str, details: dict): """记录安全事件""" self.logger.info( f"Security event: {event_type}", extra={"details": details} )