Notitie
Voor toegang tot deze pagina is autorisatie vereist. U kunt proberen u aan te melden of de directory te wijzigen.
Voor toegang tot deze pagina is autorisatie vereist. U kunt proberen de mappen te wijzigen.
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=yesin 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
.envbestanden 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.