認証、パラメータ付きクエリ、データ保護に関するベストプラクティスに従い、mssql-pythonアプリケーションを安全にしましょう。
可能な限りパスワードレス認証から始めましょう。 ローカル .env ファイルやSQLパスワードは一時的な開発補助として扱い、コードが共有環境に到達する前に秘密は管理されたアイデンティティや秘密ストアに移します。
認証セキュリティ
SQL認証ではなくMicrosoft Entra認証を使用してください
Microsoft Entra認証は、保存されたパスワードを排除し、管理されたIDをサポートします。 すべての環境でSQL認証よりも優先されます。
Azureでホストされるワークロードについては、ActiveDirectoryMSIを使ったマネージドIDを使いましょう。 保存された秘密は不要で、認証情報チェーンを経ずに接続できます:
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が必要な場合は、サポートしているODBCドライバーfor SQL Serverと組み合わせたpyodbcを使うと良いでしょう。 同じレベルの保護は提供しませんが、動的データマスキングや行レベルのセキュリティなどの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によって選択された2つのユーザー割り当て管理ID(1つは読み取り専用アクセス、もう1つは書き込みアクセス)を使用します。
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
- [ ] セキュリティイベントを記録しろ。
- [ ] 異常を監視しろ。
- [ ] 故障のアラートを設定しろ。
- [ ] 定期的なセキュリティレビューを実施。