Beveiligingsbest practices voor mssql-python-applicaties

Beveilig je mssql-python-applicaties door deze best practices te volgen voor authenticatie, geparametriseerde queries en gegevensbescherming.

Begin met wachtwoordloze authenticatie wanneer je kunt. Behandel lokale .env bestanden en SQL-wachtwoorden als tijdelijke ontwikkelingshulpmiddelen en verplaats geheimen naar beheerde identiteiten of een geheime opslag voordat code een gedeelde omgeving bereikt.

Beveiliging van verificatie

Gebruik Microsoft Entra-authenticatie boven SQL-authenticatie

Microsoft Entra-authenticatie elimineert opgeslagen wachtwoorden en ondersteunt beheerde identiteiten. Geef hieraan in alle omgevingen de voorkeur boven SQL-authenticatie.

Voor workloads die in Azure worden gehost, gebruik een beheerde identiteit met ActiveDirectoryMSI. Het heeft geen opgeslagen geheimen nodig en verbindt zonder een credential-keten te lopen:

import mssql_python

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

Voor lokale ontwikkeling gebruik ActiveDirectoryDefaultje , dat automatisch je Azure CLI of andere ontwikkelaarsgegevens oppikt. Vermijd dit in productie, want DefaultAzureCredential probeert tijdens de eerste verbinding elke provider voor aanmeldingsgegevens op volgorde uit, wat extra latentie toevoegt die workloads in productie niet nodig hebben:

import mssql_python

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

Vermijd: SQL-authenticatie slaat inloggegevens op in code/configuratie en is kwetsbaar voor lekken.


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

Hardcode inloggegevens nooit

Gebruik omgevingsvariabelen alleen voor lokale ontwikkeling. In gedeelde omgevingen geef je de voorkeur aan wachtwoordloze authenticatie. Wanneer een legacy SQL-authenticatieflow onvermijdelijk is, haal dan het geheim tijdens runtime op uit een geheime opslag in plaats van een volledige verbindingsreeks in de broncode te checken.

De volgende aanpak hardcodeert inloggegevens en mag nooit worden gebruikt:

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

Voor lokale ontwikkeling lees je referenties van omgevingsvariabelen:

import os

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

Voor gedeelde omgevingen die nog steeds een secret nodig hebben, haal deze tijdens runtime op uit 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

Preventie van SQL-injectie

Gebruik altijd geparametriseerde queries

Geparametriseerde queries voorkomen SQL-injectie door gebruikersinvoer te scheiden van de querystructuur. Parametriseer altijd gebruikersinvoer.

De volgende string-geformatteerde query is kwetsbaar voor SQL-injectie. Bouw nooit querys op deze manier:

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

De volgende geparametriseerde query is veilig, omdat de driver de waarde apart van de querytekst verzendt:

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

Parametriseer alle querycomponenten

Je kunt niet direct de namen van tabellen en kolommen parametriseren. Door ze te interpoleren uit gebruikersinvoer wordt je app kwetsbaar voor SQL-injectie:

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

Valideer in plaats daarvan dynamische identificaties aan de hand van een toelaatlijst van toegestane waarden en parametriseer de overige waarden.

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

Valideer en sanitiseer input

Wanneer je dynamische SQL met identifiers bouwt, valideer dan elke waarde aan een strikt patroon voordat je het gebruikt:

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

Gebruik opgeslagen procedures voor complexe bewerkingen

Opgeslagen procedures verkleinen het SQL-oppervlak dat aan applicatiecode wordt blootgesteld:

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

Verbindingsbeveiliging

Vereis encryptie

Versleutel altijd verbindingen. Azure SQL dwingt standaard encryptie af. Voor on-premises SQL Server, stel expliciet inEncrypt=yes:

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

Gebruik TDS 8.0 strict mode voor de hoogste beveiliging

TDS 8.0 biedt:

  • TLS 1.3 vanaf het begin van de verbinding
  • Validatie van certificaat vereist
  • Geen terugval op oudere protocollen
conn = mssql_python.connect(
    "Server=tcp:<server>.database.windows.net,1433;"
    "Database=<database>;"
    "Encrypt=strict"
)

Servercertificaten controleren

Valideer altijd het servercertificaat in productie om adversary-in-the-middle-aanvallen te voorkomen. Stel TrustServerCertificate=yes niet in, want dat omzeilt de validatie. Stel TrustServerCertificate=no in plaats daarvan in om te valideren tegen de CA-certificaten en stel HostNameInCertificate in om de hostnaam te verifiëren:

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

Gegevensbeveiliging

Bescherm gevoelige gegevens op serverniveau

Always Encrypted is momenteel niet configureerbaar via mssql-python verbindingsreeks keywords. Als je Always Encrypted nodig hebt, gebruik dan pyodbc met de ODBC-driver voor SQL Server, die het ondersteunt. Hoewel ze niet dezelfde mate van bescherming bieden, kun je SQL Server-functies zoals dynamische datamaskering en beveiliging op rijniveau gebruiken om gevoelige kolommen te beschermen.

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

row = cursor.fetchone()

Gegevens tijdens de overdracht beveiligen

  • Gebruik Encrypt=yes in verbindingsreeksen.
  • Gebruik VPN of privé-endpoints voor on-premises verbindingen.
  • Gebruik Azure Private Link voor Azure SQL.

Principe van de minste bevoegdheden

Gebruik minimale databaserechten

Applicatieaccounts zouden minimale rechten moeten hebben. Gebruik sa of db_owner niet voor applicatieverbindingen.

  • Alleen-lezen rapportage:GRANT SELECT ON SCHEMA::dbo TO ReportingApp;
  • Toegang tot specifieke tabellen: GRANT SELECT, INSERT, UPDATE ON Orders TO OrderProcessor;
  • Uitsluitend opgeslagen procedure: GRANT EXECUTE ON ProcessOrder TO OrderProcessor;

Gebruik verschillende accounts voor verschillende bewerkingen

Zet elk privilegeniveau terug met een eigen identiteit zodat een leespad geen schrijfacties kan uitvoeren. Dit voorbeeld gebruikt twee door gebruikers toegewezen beheerde identiteiten, één met alleen-lezen toegang en één met schrijftoegang, geselecteerd op basis van 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;"
    )

Houd geheimen van toepassingsrollen buiten broncodebeheer

Wachtwoorden voor applicatierollen blijven geheim. Sla ze op in een vault of een geheim-geïnjecteerde omgevingsvariabele en roteer ze met dezelfde zorg als elke andere credential.

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

Controle en logboekregistratie

Beveiligingsgebeurtenissen registreren

Log geen gevoelige gegevens, maar registreer wel beveiligingsrelevante gebeurtenissen zoals mislukte verbindingen, fouten in toestemmingen en verdachte zoekopdrachten.

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

Gevoelige bewerkingen controleren

Configureer databaseauditing voor gevoelige tabellen en bewerkingen. Je kunt ook applicatie-niveau auditing implementeren voor kritieke acties.

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

Log nooit gevoelige gegevens

Neem geen parameterwaarden toe in logberichten. Log de operatie, niet de data.

Vermijd dit - hierdoor lekt de waarde naar het log:

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

Aanbevolen - registreert intentie zonder waarden bloot te stellen:

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

Foutafhandeling

Geef geen interne details bloot

Log gedetailleerde fouten intern, maar geef generieke berichten terug aan gebruikers om lekken in de databasestructuur te voorkomen:

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

Foutmeldingen reinigen

Koppel databaseuitzonderingen aan gebruiksvriendelijke berichten die geen implementatiedetails onthullen:

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

Controlelijst voor beveiliging

Checklist voor verbindingsbeveiliging

  • [ ] Gebruik Microsoft Entra-authenticatie waar mogelijk.
  • [ ] Schakel encryptie in (Encrypt=yes).
  • [ ] Valideer servercertificaten.
  • [ ] Bewaar inloggegevens in een beveiligde kluis.
  • [ ] Bewaar .env bestanden alleen lokaal en injecteer geheimen via het doelplatform in gedeelde omgevingen.
  • [ ] Gebruik TDS 8.0 strict mode voor Azure SQL.

Zoekopdrachtbeveiliging

  • [ ] Gebruik altijd geparametriseerde queries.
  • [ ] Valideer dynamische identificaties.
  • [ ] Gebruik opgeslagen procedures voor complexe logica.
  • [ ] Beperk de grootte van zoekresultaten.

Gegevensbeveiliging

  • [ ] Gebruik gegevensbescherming op serverniveau (maskering, beveiliging op rijniveau).
  • [ ] Gebruik beveiliging op rijniveau waar passend.
  • [ ] Masker gevoelige gegevens in logs.

Toegangsbeheer

  • [ ] Gebruik het principe van het minste privilege.
  • [ ] Afzonderlijke lees-/schrijfaccounts.
  • [ ] Controleer regelmatig de toestemmingen.
  • [ ] Voer verbindingstime-outs in.

Toezicht

  • [ ] Registreer beveiligingsgebeurtenissen.
  • [ ] Houd in de gaten op afwijkingen.
  • [ ] Zet waarschuwingen voor storingen in.
  • [ ] Voer regelmatige veiligheidscontroles uit.