Bewährte Sicherheitsmethoden für mssql-python-Anwendungen

Sichern Sie Ihre mssql-python-Anwendungen, indem Sie diese Best Practices für Authentifizierung, parametrisierte Abfragen und Datenschutz befolgen.

Beginne mit passwortloser Authentifizierung, wann immer du kannst. Behandle lokale .env Dateien und SQL-Passwörter als temporäre Entwicklungshilfen und übertrage Geheimnisse in verwaltete Identitäten oder einen geheimen Speicher, bevor der Code eine gemeinsame Umgebung erreicht.

Authentifizierung und Sicherheit

Verwenden Sie Microsoft Entra-Authentifizierung statt SQL-Authentifizierung

Microsoft Entra-Authentifizierung eliminiert gespeicherte Passwörter und unterstützt verwaltete Identitäten. Ich bevorzuge es gegenüber SQL-Authentifizierung in allen Umgebungen.

Für Workloads, die in Azure gehostet werden, verwenden Sie eine verwaltete Identität mit ActiveDirectoryMSI. Es benötigt keine gespeicherten Geheimnisse und verbindet sich, ohne eine Zugangsdatenkette zu durchlaufen:

import mssql_python

def connect_with_managed_identity():
    return mssql_python.connect(
        "Server=<server>.database.windows.net;"
        "Database=<database>;"
        "Authentication=ActiveDirectoryMSI;"
        "Encrypt=yes;"
    )

Verwenden Sie für die lokale Entwicklung ActiveDirectoryDefault, das automatisch Ihre Azure CLI oder andere Anmeldeinformationen für Entwickler verwendet. Vermeiden Sie dies in der Produktion, da DefaultAzureCredential bei der ersten Verbindung jeden Anmeldeinformationsanbieter der Reihe nach ausprobiert, was zu zusätzlicher Latenz führt, die in Produktionsworkloads nicht benötigt wird:

import mssql_python

def connect_with_default_credential():
    return mssql_python.connect(
        "Server=<server>.database.windows.net;"
        "Database=<database>;"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes;"
    )

Vermeiden Sie: Die SQL-Authentifizierung speichert Zugangsdaten im Code/in der Konfiguration und ist anfällig für Lecks.


conn = mssql_python.connect("Server=...;UID=user;PWD=password")

Niemals festkodieren Sie Zugangsdaten

Verwenden Sie Umweltvariablen nur für lokale Entwicklung. In gemeinsamen Umgebungen bevorzugen Sie die passwortlose Authentifizierung. Wenn ein Legacy-SQL-Authentifizierungsablauf unvermeidbar ist, rufen Sie den geheimen Wert zur Laufzeit aus einem Geheimnisspeicher ab, anstatt eine vollständige Verbindungszeichenfolge in die Quellcodeverwaltung einzuchecken.

Folgender Ansatz kodiert die Zugangsdaten fest und sollte niemals verwendet werden:

conn_str = "Server=<server>;UID=<login>;PWD=<password>"

Für die lokale Entwicklung lesen Sie die Zugangsdaten aus den Umgebungsvariablen:

import os

conn_str = (
    f"Server={os.environ['DB_SERVER']};"
    f"Database={os.environ['DB_NAME']};"
)

Für geteilte Umgebungen, die noch ein Geheimnis benötigen, rufen Sie es zur Laufzeit aus Azure Key Vault ab:

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

Verhinderung der Einschleusung von SQL-Befehlen

Verwenden Sie immer parametrisierte Abfragen

Parametrisierte Abfragen verhindern SQL-Injection, indem sie Benutzereingaben von der Abfragestruktur trennen. Parametriere immer die Benutzereingabe.

Die folgende stringformatierte Abfrage ist anfällig für SQL-Injektionen. Erstellen Sie Abfragen niemals auf diese Weise:

user_input = "'; DROP TABLE Person.Person; --"
cursor.execute(f"SELECT * FROM Person.Person WHERE FirstName = '{user_input}'")

Die folgende parametrisierte Abfrage ist sicher, da der Treiber den Wert separat vom Abfragetext sendet:

user_input = "'; DROP TABLE Person.Person; --"
cursor.execute(
    "SELECT * FROM Person.Person WHERE FirstName = %(name)s",
    {"name": user_input}
)

Parametrisiere alle Abfragekomponenten

Man kann Tabellen- und Spaltennamen nicht direkt parametrisieren. Das Interpolieren von Nutzereingaben macht Ihre App anfällig für SQL-Injektion:

table = user_input
cursor.execute(f"SELECT * FROM {table}")

Stattdessen sollten dynamische Identifikatoren anhand einer Erlaubnisliste erlaubter Werte überprüft werden und die übrigen Werte parametrisiert werden.

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()

Eingaben validieren und bereinigen

Wenn du dynamisches SQL mit Identinern baust, validiere jeden Wert vor der Verwendung anhand eines strikten Musters:

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()}
    """)

Verwenden Sie gespeicherte Verfahren für komplexe Operationen

Gespeicherte Prozeduren verringern die SQL-Oberfläche, die dem Anwendungscode zugänglich ist:

employee_id = 5
cursor.execute("""
    EXECUTE dbo.uspGetEmployeeManagers @BusinessEntityID = %(id)s
""", {"id": employee_id})

Verbindungssicherheit

Verschlüsselung erforderlich.

Verschlüssele immer Verbindungen. Azure SQL erzwingt standardmäßig die Verschlüsselung. Für lokale SQL Server-Instanzen legen Sie Encrypt=yes explizit fest:

conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Encrypt=yes;"
    "TrustServerCertificate=no"
)

Verwenden Sie TDS 8.0 im strengen Modus für höchste Sicherheit

TDS 8.0 bietet:

  • TLS 1.3 ab Verbindungsaufbau
  • Zertifikatsüberprüfung erforderlich
  • Kein Rückgriff auf ältere Protokolle
conn = mssql_python.connect(
    "Server=tcp:<server>.database.windows.net,1433;"
    "Database=<database>;"
    "Encrypt=strict"
)

Serverzertifikate validieren

Validiere immer das in der Produktion befindliche Serverzertifikat, um Angriffe von Adversary-in-the-Middle zu verhindern. Setze TrustServerCertificate=yes nicht, weil es die Validierung umgeht. Stattdessen setzen Sie TrustServerCertificate=no so, dass gegen die CA-Zertifikate validiert wird, und HostNameInCertificate so, dass der Hostname überprüft wird:

conn = mssql_python.connect(
    "Server=<server>;"
    "Database=<database>;"
    "Encrypt=yes;"
    "TrustServerCertificate=no;"
    "HostNameInCertificate=<server>.domain.com"
)

Datenschutz

Schützen Sie sensible Daten auf Serverebene

Always Encrypted ist derzeit nicht über mssql-python-Verbindungszeichenfolge-Keywords konfigurierbar. Wenn du Always Encrypted brauchst, nutze pyodbc mit dem ODBC-Treiber für SQL Server, der es unterstützt. Auch wenn sie nicht denselben Schutz bieten, können Sie SQL Server-Funktionen wie dynamische Datenmaskierung und Zeilensicherheit nutzen, um sensible Spalten zu schützen.

employee_id = 1
cursor.execute("""
    SELECT NationalIDNumber, LoginID
    FROM HumanResources.Employee
    WHERE BusinessEntityID = %(id)s
""", {"id": employee_id})

row = cursor.fetchone()

Schützen der Daten während der Übertragung

  • Verwenden Sie Encrypt=yes in Verbindungszeichenfolgen.
  • Verwenden Sie VPN oder private Endgeräte für lokale Verbindungen.
  • Verwenden Sie Azure Private Link für Azure SQL.

Prinzip der geringsten Rechte

Verwenden Sie minimale Datenbankberechtigungen

Anwendungskonten sollten minimale Berechtigungen haben. Verwenden Sie sa oder db_owner nicht für Anwendungsverbindungen.

  • Schreibgeschützte Berichterstellung: GRANT SELECT ON SCHEMA::dbo TO ReportingApp;
  • Spezifischer Tabellenzugang: GRANT SELECT, INSERT, UPDATE ON Orders TO OrderProcessor;
  • Nur gespeichertes Verfahren: GRANT EXECUTE ON ProcessOrder TO OrderProcessor;

Verwenden Sie verschiedene Konten für unterschiedliche Operationen

Jede Berechtigungsstufe sollte durch eine eigene Identität abgesichert werden, sodass ein Lesepfad keine Schreibvorgänge ausführen kann. Dieses Beispiel verwendet zwei benutzerzugewiesene verwaltete Identitäten, eine mit Lesezugriff und eine mit Schreibzugriff, ausgewählt anhand der Client-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;"
    )

Halten Sie Anwendungsrollengeheimnisse aus der Quellkontrolle heraus

Passwörter für Anwendungsrollen sind weiterhin Geheimnisse. Speichere sie in einem Tresor oder in einer geheim injizierten Umgebungsvariable und rotiere sie mit der gleichen Sorgfalt wie jede andere Zugangsberechtigung.

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")

Überwachung und Protokollierung

Sicherheitsereignisse protokollieren

Protokolliere keine sensiblen Daten, sondern dokumentiere sicherheitsrelevante Ereignisse wie fehlgeschlagene Verbindungen, Berechtigungsfehler und verdächtige Anfragen.

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

Auditrelevante Vorgänge

Konfigurieren Sie Datenbankprüfungen für sensible Tabellen und Operationen. Sie können auch Audits auf Anwendungsebene für kritische Aktionen implementieren.

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()
    })

Niemals sensible Daten protokollieren

Fügen Sie keine Parameterwerte in Log-Nachrichten hinzu. Protokolliere die Operation, nicht die Daten.

Vermeiden – der Wert gelangt ins Protokoll:

nid = "295847284"
logger.debug(f"Query: SELECT * FROM HumanResources.Employee WHERE NationalIDNumber = '{nid}'")

Empfohlen – Protokollierung der Absicht, ohne Werte freizugeben:

logger.debug("Executing employee lookup query")
nid = "295847284"
cursor.execute("SELECT * FROM HumanResources.Employee WHERE NationalIDNumber = %(nid)s", {"nid": nid})

Fehlerbehandlung

Gib keine internen Details preis.

Detaillierte Fehler intern protokollieren, aber generische Nachrichten an die Benutzer zurückgeben, um ein Leak der Datenbankstruktur zu vermeiden:

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")

Fehlermeldungen reinigen

Zuordnung von Datenbank-Ausnahmen auf benutzerfreundliche Nachrichten, die keine Implementierungsdetails offenlegen:

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."

Sicherheitscheckliste

Checkliste für Verbindungssicherheit

  • [ ] Verwenden Sie Microsoft Entra-Authentifizierung, wenn möglich.
  • [ ] Verschlüsselung aktivieren (Encrypt=yes).
  • [ ] Serverzertifikate validieren.
  • [ ] Bewahren Sie die Zugangsdaten in einem sicheren Tresor auf.
  • [ ] Speichern .env Sie Dateien nur lokal und injizieren Sie Geheimnisse über die Zielplattform in gemeinsamen Umgebungen.
  • [ ] Verwenden Sie TDS 8.0 im strengen Modus für Azure SQL.

Abfragesicherheit

  • [ ] Verwenden Sie immer parametrisierte Abfragen.
  • [ ] Validiere dynamische Kennungen.
  • [ ] Verwenden Sie gespeicherte Verfahren für komplexe Logik.
  • [ ] Begrenze die Größen von Abfrageergebnissen.

Datensicherheit

  • [ ] Verwenden Sie den Datenschutz auf Serverebene (Datenmaskierung, Sicherheit auf Zeilenebene).
  • [ ] Setze Sicherheitsmaßnahmen auf Reihenebene ein, wo es angebracht ist.
  • [ ] Maskiere sensible Daten in Protokollen.

Zugriffskontrolle

  • [ ] Nutze das Prinzip des geringsten Privilegs.
  • [ ] Getrennte Lese-/Schreibkonten.
  • [ ] Überprüfe regelmäßig Genehmigungen.
  • [ ] Implementiere Verbindungs-Timeouts.

Monitoring

  • [ ] Sicherheitsereignisse protokollieren.
  • [ ] Überwacht auf Anomalien.
  • [ ] Setzt Alarme für Ausfälle ein.
  • [ ] Führen Sie regelmäßige Sicherheitsüberprüfungen durch.