Praktik terbaik keamanan untuk aplikasi mssql-python

Amankan aplikasi mssql-python Anda dengan mengikuti praktik terbaik ini untuk autentikasi, kueri berparameter, dan perlindungan data.

Mulailah dengan autentikasi tanpa kata sandi kapan pun Anda bisa. Perlakukan file lokal .env dan kata sandi SQL sebagai alat bantu pengembangan sementara, dan pindahkan rahasia ke identitas terkelola atau penyimpanan rahasia sebelum kode mencapai lingkungan bersama.

Autentikasi keamanan

Menggunakan autentikasi Microsoft Entra melalui autentikasi SQL

Autentikasi Microsoft Entra menghilangkan kata sandi yang disimpan dan mendukung identitas terkelola. Gunakan ini alih-alih autentikasi SQL di semua lingkungan.

Untuk beban kerja yang dihosting di Azure, gunakan identitas terkelola dengan ActiveDirectoryMSI. Ini tidak memerlukan rahasia yang disimpan dan terhubung tanpa menelusuri rantai kredensial:

import mssql_python

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

Untuk pengembangan lokal, gunakan ActiveDirectoryDefault, yang mengambil Azure CLI atau kredensial pengembang lainnya secara otomatis. Hindari penggunaan ini di lingkungan produksi, karena DefaultAzureCredential akan mencoba setiap penyedia kredensial secara berurutan pada koneksi pertama, yang menambah latensi yang tidak diperlukan oleh beban kerja produksi:

import mssql_python

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

Hindari: Autentikasi SQL menyimpan kredensial dalam kode/konfigurasi dan rentan terhadap kebocoran.


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

Jangan pernah menanamkan kredensial langsung ke dalam kode

Gunakan variabel lingkungan hanya untuk pengembangan lokal. Di lingkungan bersama, lebih suka autentikasi tanpa kata sandi. Ketika alur autentikasi SQL lawas tidak dapat dihindari, ambil rahasia pada saat runtime dari penyimpanan rahasia alih-alih menyimpan string koneksi lengkap ke dalam kontrol versi.

Pendekatan berikut menanamkan kredensial secara hardcode dan tidak boleh pernah digunakan:

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

Untuk pengembangan lokal, baca kredensial dari variabel lingkungan:

import os

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

Untuk lingkungan bersama yang masih memerlukan rahasia, ambil rahasia tersebut pada waktu proses dari 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

Pencegahan injeksi SQL

Selalu gunakan kueri berparameter

Kueri berparameter mencegah injeksi SQL dengan memisahkan input pengguna dari struktur kueri. Selalu parameterkan input pengguna.

Kueri berformat string berikut rentan terhadap injeksi SQL. Jangan pernah membuat kueri dengan cara ini:

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

Kueri berparameter berikut aman, karena driver mengirimkan nilai secara terpisah dari teks kueri:

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

Jadikan semua komponen kueri sebagai parameter

Anda tidak dapat secara langsung membuat parameterisasi nama tabel dan kolom. Menginterpolasinya dari input pengguna membuat aplikasi Anda rentan terhadap injeksi SQL:

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

Sebagai gantinya, validasi pengidentifikasi dinamis terhadap daftar nilai yang diizinkan, dan parameterkan nilai yang tersisa.

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

Validasikan dan bersihkan input

Saat Anda membangun SQL dinamis dengan pengidentifikasi, validasi setiap nilai terhadap pola ketat sebelum menggunakannya:

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

Gunakan prosedur tersimpan untuk operasi yang kompleks

Prosedur tersimpan mengurangi area permukaan SQL yang terekspos ke kode aplikasi:

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

Keamanan koneksi

Memerlukan enkripsi

Selalu enkripsi koneksi. Azure SQL menerapkan enkripsi secara default. Untuk SQL Server lokal, atur Encrypt=yes secara eksplisit:

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

Gunakan mode ketat TDS 8.0 untuk keamanan tertinggi

TDS 8.0 menyediakan:

TLS 1.3 sejak awal koneksi
Validasi sertifikat diperlukan
Tidak ada penggantian ke protokol lama

conn = mssql_python.connect(
    "Server=tcp:<server>.database.windows.net,1433;"
    "Database=<database>;"
    "Encrypt=strict"
)

Memverifikasi sertifikat server

Selalu validasi sertifikat server dalam produksi untuk mencegah serangan musuh di tengah. Jangan atur TrustServerCertificate=yes, karena melewati validasi. Sebagai gantinya, atur TrustServerCertificate=no untuk memvalidasi terhadap sertifikat CA, dan atur HostNameInCertificate untuk memverifikasi nama host:

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

Perlindungan data

Lindungi data sensitif di tingkat server

Selalu Terenkripsi saat ini tidak dapat dikonfigurasi melalui kata kunci string koneksi mssql-python. Jika Anda memerlukan Always Encrypted, gunakan pyodbc dengan Driver ODBC untuk SQL Server, yang mendukungnya. Meskipun tidak memberikan tingkat perlindungan yang sama, Anda dapat menggunakan fitur SQL Server seperti penyamaran data dinamis dan keamanan tingkat baris untuk melindungi kolom sensitif.

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

row = cursor.fetchone()

Memproteksi data dalam perjalanan

Gunakan Encrypt=yes dalam string koneksi.
Gunakan VPN atau titik akhir privat untuk koneksi lokal.
Gunakan Azure Private Link untuk Azure SQL.

Prinsip hak istimewa paling rendah

Gunakan izin database minimal

Akun aplikasi harus memiliki izin minimal. Jangan gunakan sa atau db_owner untuk koneksi aplikasi.

Pelaporan baca-saja: GRANT SELECT ON SCHEMA::dbo TO ReportingApp;
Akses tabel tertentu: GRANT SELECT, INSERT, UPDATE ON Orders TO OrderProcessor;
Hanya prosedur tersimpan: GRANT EXECUTE ON ProcessOrder TO OrderProcessor;

Gunakan akun yang berbeda untuk operasi yang berbeda

Dukung setiap tingkat hak istimewa dengan identitasnya sendiri sehingga jalur baca tidak dapat melakukan penulisan. Contoh ini menggunakan dua identitas terkelola yang ditetapkan pengguna, satu diberikan akses baca-saja dan satu akses tulis yang diberikan, dipilih oleh ID klien:

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

Jauhkan rahasia peran aplikasi dari kontrol sumber

Kata sandi peran aplikasi masih rahasia. Simpan semuanya di vault atau variabel lingkungan yang diisi melalui injeksi rahasia, dan lakukan rotasi dengan tingkat kehati-hatian yang sama seperti kredensial lainnya.

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

Audit dan pencatatan

Mencatat peristiwa keamanan

Jangan mencatat data sensitif, tetapi catat peristiwa yang relevan dengan keamanan seperti koneksi yang gagal, kesalahan izin, dan kueri yang mencurigakan.

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

Audit operasi yang sensitif

Konfigurasikan audit database untuk tabel dan operasi sensitif. Anda juga dapat menerapkan audit tingkat aplikasi untuk tindakan penting.

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

Jangan pernah mencatat data sensitif

Jangan sertakan nilai parameter dalam pesan log. Catat operasi, bukan data.

Hindari - membocorkan nilai ke log:

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

Direkomendasikan - mencatat intent tanpa mengekspos nilai:

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

Penanganan kesalahan

Jangan mengungkapkan detail internal

Catat kesalahan terperinci secara internal tetapi kembalikan pesan umum kepada pengguna untuk menghindari kebocoran struktur database:

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

Bersihkan pesan kesalahan

Petakan pengecualian database ke pesan yang mudah digunakan yang tidak mengungkapkan detail implementasi:

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

Daftar periksa keamanan

Daftar periksa keamanan koneksi

[ ] Gunakan autentikasi Microsoft Entra jika memungkinkan.
[ ] Aktifkan enkripsi (Encrypt=yes).
[ ] Validasi sertifikat server.
[ ] Simpan kredensial di brankas yang aman.
[ ] Simpan .env file lokal saja, dan suntikkan rahasia melalui platform target di lingkungan bersama.
[ ] Gunakan mode ketat TDS 8.0 untuk Azure SQL.

Keamanan kueri

[ ] Selalu gunakan kueri berparameter.
[ ] Validasi pengidentifikasi dinamis.
[ ] Gunakan prosedur tersimpan untuk logika yang kompleks.
[ ] Batasi ukuran hasil kueri.

Keamanan data

[ ] Gunakan perlindungan data tingkat server (masking, keamanan tingkat baris).
[ ] Gunakan keamanan tingkat baris jika perlu.
[ ] Tutupi data sensitif dalam log.

Kontrol akses

[ ] Gunakan prinsip hak istimewa paling kecil.
[ ] Pisahkan akun baca/tulis.
[ ] Audit izin akses secara teratur.
[ ] Terapkan batas waktu koneksi.

Monitoring

[ ] Catat peristiwa keamanan.
[ ] Pantau anomali.
[ ] Siapkan peringatan untuk kegagalan.
[ ] Lakukan tinjauan keamanan secara teratur.

Saran dan Komentar

Apakah halaman ini membantu?

Last updated on 2026-07-22