遵循以下最佳實務,保護您的 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 Key Vault 取回:
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 連接字串 關鍵字來設定。 如果你需要 Always Encrypted,可以用 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 Private Link 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檔案保留在本機,並在共用環境中透過目標平台注入機密資訊。 - [ ] 使用 TDS 8.0 嚴格模式來處理 Azure SQL。
查詢安全性
- [ ] 務必使用參數化查詢。
- [ ] 驗證動態識別碼。
- [ ] 使用儲存程序來處理複雜邏輯。
- [ ] 限制查詢結果大小。
數據安全性
- [ ] 使用伺服器層級的資料保護(遮蔽、列級安全)。
- [ ] 適當時使用行級安全措施。
- [ ] 將敏感資料隱藏在日誌中。
存取控制
- [ ] 使用最小特權原則。
- [ ] 讀寫帳號分離。
- [ ] 定期稽核權限。
- [ ] 實作連線逾時機制。
Monitoring
- [ ] 記錄安全事件。
- [ ] 監控異常。
- [ ] 設置故障警報。
- [ ] 定期進行安全審查。