mssql-python 應用程式的安全最佳實務

遵循以下最佳實務,保護您的 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.

最低權限原則

使用最小的資料庫權限

應用程式帳號應該擁有最少的權限。 不要使用 sadb_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

  • [ ] 記錄安全事件。
  • [ ] 監控異常。
  • [ ] 設置故障警報。
  • [ ] 定期進行安全審查。