Лучшие практики безопасности для приложений 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 с драйвером ODBC для SQL Server, который его поддерживает. Хотя они не обеспечивают такого же уровня защиты, вы можете использовать функции 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;

Используйте разные аккаунты для разных операций

Назначьте каждому уровню привилегий отдельную учётную запись, чтобы путь чтения не мог выполнять операции записи. В этом примере используются две управляемые идентичности, назначаемые пользователем: одной предоставлен доступ только для чтения, другой — доступ на запись; выбор выполняется по идентификатору клиента:

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

  • [ ] Регистрировать события безопасности.
  • [ ] Следите за аномалиями.
  • [ ] Настройте оповещения о сбоях.
  • [ ] Проводите регулярные проверки безопасности.

Связанные материалы

  • Last updated on