通过遵循这些认证、参数化查询和数据保护的最佳实践，保护您的 mssql-python 应用安全。
尽量先用无密码认证。 将本地 .env 文件和SQL密码视为临时开发辅助工具，并在代码进入共享环境之前，将秘密转移到托管身份或秘密存储中。
身份验证安全性
使用Microsoft Entra认证而非SQL认证
Microsoft Entra认证消除了存储的密码，支持托管身份。 在所有环境下，我都更倾向于用它而不是SQL认证。
对于在 Azure 中托管的工作负载，请使用带有 ActiveDirectoryMSI 的托管标识。 它不需要存储的秘密，并且无需经过凭证链即可连接：
import mssql_python
def connect_with_managed_identity():
return mssql_python.connect(
"Server=<server>.database.windows.net;"
"Database=<database>;"
"Authentication=ActiveDirectoryMSI;"
"Encrypt=yes;"
)
本地开发时使用ActiveDirectoryDefault，它会自动获取你的 Azure CLI 或其他开发者凭证。 避免在生产环境中使用它，因为 DefaultAzureCredential 每个凭证提供者在第一次连接时会按顺序尝试，这会增加生产工作负载不需要的延迟：
import mssql_python
def connect_with_default_credential():
return mssql_python.connect(
"Server=<server>.database.windows.net;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"Encrypt=yes;"
)
避免： SQL 认证将凭据存储在代码/配置中，且容易受到泄露。
conn = mssql_python.connect("Server=...;UID=user;PWD=password")
千万别硬编码凭证
仅在本地开发中使用环境变量。 在共享环境中，更倾向于无密码认证。 当无法避免使用旧版 SQL 身份验证流程时，应在运行时从机密存储中获取机密，而不要将完整的连接字符串签入源代码管理。
以下方法是硬编码凭证，且绝不应使用：
conn_str = "Server=<server>;UID=<login>;PWD=<password>"
在本地开发时，从环境变量中读取凭证：
import os
conn_str = (
f"Server={os.environ['DB_SERVER']};"
f"Database={os.environ['DB_NAME']};"
)
对于仍需秘密的共享环境，可在运行时从 Azure 密钥保管库 获取：
from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential
def get_connection_string():
credential = DefaultAzureCredential()
secret_client = SecretClient(
vault_url="https://myvault.vault.azure.net/",
credential=credential
)
return secret_client.get_secret("db-connection-string").value
SQL 注入预防
始终使用参数化查询
参数化查询通过将用户输入与查询结构分离，防止SQL注入。 始终对用户输入进行参数化。
以下字符串格式查询易受SQL注入影响。 切勿这样构建查询：
user_input = "'; DROP TABLE Person.Person; --"
cursor.execute(f"SELECT * FROM Person.Person WHERE FirstName = '{user_input}'")
以下参数化查询是安全的，因为驱动程序将值与查询文本分开发送：
user_input = "'; DROP TABLE Person.Person; --"
cursor.execute(
"SELECT * FROM Person.Person WHERE FirstName = %(name)s",
{"name": user_input}
)
参数化所有查询组件
你不能直接参数化表和列名。 将它们直接拼接到用户输入中会使你的应用容易遭受 SQL 注入攻击：
table = user_input
cursor.execute(f"SELECT * FROM {table}")
相反，应根据允许值列表验证动态标识符，并对剩余值进行参数化。
ALLOWED_TABLES = {"Person.Person", "Production.Product", "Sales.SalesOrderHeader"}
def query_table(cursor, table_name: str, conditions: dict):
"""Query with validated table name."""
if table_name not in ALLOWED_TABLES:
raise ValueError(f"Invalid table: {table_name}")
# Table name is safe, parameters are parameterized
where_clauses = [f"{k} = %({k})s" for k in conditions.keys()]
query = f"SELECT * FROM {table_name} WHERE {' AND '.join(where_clauses)}"
cursor.execute(query, conditions)
return cursor.fetchall()
验证并净化输入
当你构建带有标识符的动态SQL时，使用前请根据严格的模式验证每个值：
import re
def validate_identifier(value: str) -> bool:
"""Validate SQL identifier (table/column name)."""
# Only allow alphanumeric and underscore
return bool(re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', value))
def safe_order_by(cursor, table: str, order_column: str, direction: str):
"""Order by with validation."""
if not validate_identifier(order_column):
raise ValueError(f"Invalid column name: {order_column}")
if direction.upper() not in ("ASC", "DESC"):
raise ValueError(f"Invalid direction: {direction}")
cursor.execute(f"""
SELECT * FROM {table}
ORDER BY {order_column} {direction.upper()}
""")
对复杂操作使用存储过程
存储过程减少了应用代码暴露的SQL表面积：
employee_id = 5
cursor.execute("""
EXECUTE dbo.uspGetEmployeeManagers @BusinessEntityID = %(id)s
""", {"id": employee_id})
连接安全性
要求加密
一定要加密连接。 Azure SQL 默认强制加密。 对于本地部署的 SQL Server，显式设置为Encrypt=yes：
conn = mssql_python.connect(
"Server=<server>.database.windows.net;"
"Database=<database>;"
"Encrypt=yes;"
"TrustServerCertificate=no"
)
使用TDS 8.0严格模式以获得最高安全性
TDS 8.0 提供：
- 从连接开始即使用 TLS 1.3
- 需要证书验证
- 不回退到旧版协议
conn = mssql_python.connect(
"Server=tcp:<server>.database.windows.net,1433;"
"Database=<database>;"
"Encrypt=strict"
)
验证服务器证书
在生产环境中务必验证服务器证书，以防止中间攻击。 不要设置 TrustServerCertificate=yes，因为那会绕过验证。 相反，应将 TrustServerCertificate=no 设置为根据 CA 证书进行验证，并将 HostNameInCertificate 设置为验证主机名：
conn = mssql_python.connect(
"Server=<server>;"
"Database=<database>;"
"Encrypt=yes;"
"TrustServerCertificate=no;"
"HostNameInCertificate=<server>.domain.com"
)
数据保护
保护服务器级别的敏感数据
Always Encrypted 目前无法通过 mssql-python 连接字符串 关键字配置。 如果你需要始终加密，可以用 pyodbc 配合支持 SQL Server 的 ODBC 驱动。 虽然它们没有同样的保护层级，但你可以利用SQL Server的功能，如动态数据掩蔽和行级安全，来保护敏感列。
employee_id = 1
cursor.execute("""
SELECT NationalIDNumber, LoginID
FROM HumanResources.Employee
WHERE BusinessEntityID = %(id)s
""", {"id": employee_id})
row = cursor.fetchone()
保护传输中的数据
- 在连接字符串中使用
Encrypt=yes。 - 使用VPN或私有端点进行本地连接。
- Use Azure 专用链接 for Azure SQL.
最低权限原则
使用最小的数据库权限
应用账户权限应为最少。 不要使用 sa 或 db_owner 用于应用连接。
-
只读报告功能：
GRANT SELECT ON SCHEMA::dbo TO ReportingApp; -
特定表访问：
GRANT SELECT, INSERT, UPDATE ON Orders TO OrderProcessor; -
仅限存储过程：
GRANT EXECUTE ON ProcessOrder TO OrderProcessor;
针对不同操作使用不同的账户
为每个权限级别绑定各自独立的身份，以确保读路径无法执行写操作。 本示例使用两个用户指定的托管身份，一个为只读权限，一个为写权限，由客户端ID选择：
readonly_client_id = os.environ["READONLY_IDENTITY_CLIENT_ID"]
readwrite_client_id = os.environ["READWRITE_IDENTITY_CLIENT_ID"]
def get_readonly_connection():
"""Connection for read-only operations."""
return mssql_python.connect(
f"Server={server};Database={db};"
f"Authentication=ActiveDirectoryMSI;UID={readonly_client_id};"
f"Encrypt=yes;ApplicationIntent=ReadOnly;"
)
def get_readwrite_connection():
"""Connection for write operations."""
return mssql_python.connect(
f"Server={server};Database={db};"
f"Authentication=ActiveDirectoryMSI;UID={readwrite_client_id};"
f"Encrypt=yes;"
)
将应用角色秘密排除在源代码控制之外
应用角色密码仍然是秘密。 将它们存储在保管库或注入机密信息的环境变量中，并以对待其他任何凭证同样的谨慎程度定期轮换它们。
def execute_with_role(cursor, role: str, query: str, params: dict):
"""Execute query with specific application role."""
# Activate application role
cursor.execute(
"EXECUTE sp_setapprole @rolename = %(role)s, @password = %(pwd)s",
{"role": role, "pwd": os.environ[f"ROLE_{role.upper()}_PWD"]}
)
try:
cursor.execute(query, params)
return cursor.fetchall()
finally:
# Reset to original context
cursor.execute("EXECUTE sp_unsetapprole")
审核和日志记录
记录安全事件
不要记录敏感数据，但要记录与安全相关的事件，比如连接失败、权限错误和可疑查询。
import logging
logger = logging.getLogger("db_security")
def secure_connect(connection_string: str):
"""Connect with security logging."""
logger.info("Attempting database connection")
try:
conn = mssql_python.connect(connection_string)
logger.info("Database connection established")
return conn
except mssql_python.OperationalError as e:
logger.warning(f"Database connection failed: {type(e).__name__}")
raise
审计敏感操作
为敏感表和操作配置数据库审计。 你也可以对关键操作实施应用级审计。
def audit_data_access(cursor, user_id: str, action: str, resource: str):
"""Log data access for audit trail."""
cursor.execute("""
INSERT INTO AuditLog (UserID, Action, Resource, Timestamp, IPAddress)
VALUES (%(user)s, %(action)s, %(resource)s, GETUTCDATE(), %(ip)s)
""", {
"user": user_id,
"action": action,
"resource": resource,
"ip": get_client_ip()
})
切勿记录敏感数据
日志消息中不要包含参数值。 记录操作本身，而不是数据。
避免 - 会把数值泄漏到日志中：
nid = "295847284"
logger.debug(f"Query: SELECT * FROM HumanResources.Employee WHERE NationalIDNumber = '{nid}'")
推荐——记录意图而不暴露值：
logger.debug("Executing employee lookup query")
nid = "295847284"
cursor.execute("SELECT * FROM HumanResources.Employee WHERE NationalIDNumber = %(nid)s", {"nid": nid})
错误处理
不要泄露内部细节
在内部记录详细错误信息，但向用户返回通用错误消息，以避免泄露数据库结构：
def safe_query(cursor, query: str, params: dict):
"""Execute query with safe error handling."""
try:
cursor.execute(query, params)
return cursor.fetchall()
except mssql_python.ProgrammingError as e:
# Log full error internally
logging.error(f"Query error: {e}")
# Return generic error to user
raise UserFacingError("An error occurred processing your request")
except mssql_python.IntegrityError:
raise UserFacingError("Invalid data provided")
清理错误信息
将数据库异常映射为不暴露实现细节、用户易懂的提示信息：
class UserFacingError(Exception):
"""Exception safe to show to users."""
pass
def handle_database_error(error: Exception) -> str:
"""Convert database errors to safe user messages."""
if isinstance(error, mssql_python.IntegrityError):
if "UNIQUE" in str(error):
return "A record with this value already exists"
if "FOREIGN KEY" in str(error):
return "Referenced item not found"
return "An error occurred. Please try again later."
安全清单
连接安全检查清单
- [ ] 尽可能使用Microsoft Entra认证。
- [ ] 启用加密（
Encrypt=yes）。 - [ ] 验证服务器证书。
- [ ] 把凭证存进一个安全的保险库。
- [ ] 仅在本地保留
.env文件，并在共享环境中通过目标平台注入密钥。 - [ ] Azure SQL 使用 TDS 8.0 严格模式。
查询安全
- [ ] 始终使用参数化查询。
- [ ] 验证动态标识符。
- [ ] 对于复杂逻辑，使用存储过程。
- [ ] 限制查询结果大小。
数据安全
- [ ] 使用服务器级数据保护（掩蔽、行级安全）。
- [ ] 在适当情况下使用行级安保。
- [ ] 在日志中隐藏敏感数据。
访问控制
- [ ] 使用最小特权原则。
- [ ] 读写账号分离。
- [ ] 定期审查权限。
- [ ] 实现连接超时。
Monitoring
- [ ] 记录安全事件。
- [ ] 监测异常。
- [ ] 设置故障警报。
- [ ] 定期进行安全审查。