数据库安全与合规：保护你的数据资产-编程阁

数据库安全与合规：保护你的数据资产

引言

数据库是企业的核心数据资产，数据库安全不仅关系到业务的正常运行，更关系到用户隐私和企业声誉。本文将从访问控制、数据加密、审计日志、备份恢复等多个维度，全面探讨数据库安全与合规的最佳实践。

一、访问控制

1.1 最小权限原则

-- 创建应用角色 CREATE ROLE app_readonly; GRANT SELECT ON ALL TABLES IN SCHEMA public TO app_readonly; -- 创建应用读写角色 CREATE ROLE app_readwrite; GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO app_readwrite; -- 创建管理员角色 CREATE ROLE app_admin; GRANT ALL PRIVILEGES ON DATABASE myapp TO app_admin; -- 创建用户并分配角色 CREATE USER app_service WITH PASSWORD 'strong_password'; GRANT app_readwrite TO app_service; -- 行级安全策略 (PostgreSQL) CREATE POLICY user_data_policy ON users FOR ALL USING (created_by = current_user); -- 列级权限 GRANT UPDATE (email, phone) ON users TO app_readwrite;

1.2 Go语言数据库用户管理

package dbsecurity import ( "context" "database/sql" "fmt" ) type AccessControl struct { db *sql.DB } func NewAccessControl(db *sql.DB) *AccessControl { return &AccessControl{db: db} } type Role struct { Name string Permissions []string } type User struct { Username string Roles []string IsActive bool } func (ac *AccessControl) CreateRole(ctx context.Context, role *Role) error { query := `CREATE ROLE $1 WITH NOLOGIN` _, err := ac.db.ExecContext(ctx, query, role.Name) if err != nil { return fmt.Errorf("failed to create role: %w", err) } for _, perm := range role.Permissions { grantQuery := fmt.Sprintf("GRANT %s TO %s", perm, role.Name) if _, err := ac.db.ExecContext(ctx, grantQuery); err != nil { return fmt.Errorf("failed to grant permission: %w", err) } } return nil } func (ac *AccessControl) GrantTableAccess(ctx context.Context, roleName, tableName, privilege string) error { query := fmt.Sprintf("GRANT %s ON TABLE %s TO %s", privilege, tableName, roleName) _, err := ac.db.ExecContext(ctx, query) return err } func (ac *AccessControl) RevokeTableAccess(ctx context.Context, roleName, tableName, privilege string) error { query := fmt.Sprintf("REVOKE %s ON TABLE %s FROM %s", privilege, tableName, roleName) _, err := ac.db.ExecContext(ctx, query) return err } func (ac *AccessControl) CreateUserWithRoles(ctx context.Context, username, password string, roles []string) error { query := `CREATE USER $1 WITH PASSWORD $2` _, err := ac.db.ExecContext(ctx, query, username, password) if err != nil { return fmt.Errorf("failed to create user: %w", err) } for _, role := range roles { grantQuery := fmt.Sprintf("GRANT %s TO %s", role, username) if _, err := ac.db.ExecContext(ctx, grantQuery); err != nil { return fmt.Errorf("failed to grant role: %w", err) } } return nil } func (ac *AccessControl) DeactivateUser(ctx context.Context, username string) error { query := `ALTER USER $1 WITH NOLOGIN` _, err := ac.db.ExecContext(ctx, query, username) return err } func (ac *AccessControl) ListUsers(ctx context.Context) ([]User, error) { query := `SELECT rolname, rolsuper, rolinherit, rolcreaterole, rolcreatedb, rolcanlogin FROM pg_roles WHERE rolname NOT LIKE 'pg_%'` rows, err := ac.db.QueryContext(ctx, query) if err != nil { return nil, fmt.Errorf("failed to list users: %w", err) } defer rows.Close() var users []User for rows.Next() { var u User var isSuper, inherits, canLogin bool if err := rows.Scan(&u.Username, &isSuper, &inherits, &u.IsActive, &u.IsActive, &canLogin); err != nil { continue } u.IsActive = canLogin users = append(users, u) } return users, nil }

二、数据加密

2.1 透明数据加密（TDE）

-- PostgreSQL TDE (使用pgcrypto) CREATE EXTENSION IF NOT EXISTS pgcrypto; -- 列级加密 CREATE TABLE users ( id SERIAL PRIMARY KEY, name TEXT, ssn_encrypted BYTEA ); -- 加密插入 INSERT INTO users (name, ssn_encrypted) VALUES ('John Doe', pgp_sym_encrypt('123-45-6789', 'encryption_key')); -- 解密查询 SELECT name, pgp_sym_decrypt(ssn_encrypted, 'encryption_key') as ssn FROM users;

2.2 Go语言数据加密

package dbsecurity import ( "crypto/aes" "crypto/cipher" "crypto/rand" "crypto/sha256" "database/sql" "encoding/base64" "fmt" "io" ) type DataEncryption struct { key []byte } func NewDataEncryption(masterKey string) (*DataEncryption, error) { hash := sha256.Sum256([]byte(masterKey)) return &DataEncryption{key: hash[:]}, nil } func (de *DataEncryption) Encrypt(plaintext string) (string, error) { block, err := aes.NewCipher(de.key) if err != nil { return "", fmt.Errorf("failed to create cipher: %w", err) } gcm, err := cipher.NewGCM(block) if err != nil { return "", fmt.Errorf("failed to create GCM: %w", err) } nonce := make([]byte, gcm.NonceSize()) if _, err := io.ReadFull(rand.Reader, nonce); err != nil { return "", fmt.Errorf("failed to generate nonce: %w", err) } ciphertext := gcm.Seal(nonce, nonce, []byte(plaintext), nil) return base64.StdEncoding.EncodeToString(ciphertext), nil } func (de *DataEncryption) Decrypt(ciphertext string) (string, error) { data, err := base64.StdEncoding.DecodeString(ciphertext) if err != nil { return "", fmt.Errorf("failed to decode: %w", err) } block, err := aes.NewCipher(de.key) if err != nil { return "", fmt.Errorf("failed to create cipher: %w", err) } gcm, err := cipher.NewGCM(block) if err != nil { return "", fmt.Errorf("failed to create GCM: %w", err) } nonceSize := gcm.NonceSize() if len(data) < nonceSize { return "", fmt.Errorf("ciphertext too short") } nonce, ciphertextBytes := data[:nonceSize], data[nonceSize:] plaintext, err := gcm.Open(nil, nonce, ciphertextBytes, nil) if err != nil { return "", fmt.Errorf("failed to decrypt: %w", err) } return string(plaintext), nil } type EncryptedDB struct { db *sql.DB encryption *DataEncryption } func NewEncryptedDB(db *sql.DB, encryption *DataEncryption) *EncryptedDB { return &EncryptedDB{ db: db, encryption: encryption, } } func (edb *EncryptedDB) SaveUserSSN(ctx context.Context, userID, ssn string) error { encryptedSSN, err := edb.encryption.Encrypt(ssn) if err != nil { return err } query := `UPDATE users SET ssn_encrypted = $1 WHERE id = $2` _, err = edb.db.ExecContext(ctx, query, encryptedSSN, userID) return err } func (edb *EncryptedDB) GetUserSSN(ctx context.Context, userID string) (string, error) { query := `SELECT ssn_encrypted FROM users WHERE id = $1` var encryptedSSN string if err := edb.db.QueryRowContext(ctx, query, userID).Scan(&encryptedSSN); err != nil { return "", err } return edb.encryption.Decrypt(encryptedSSN) }

2.3 TLS连接加密

package dbsecurity import ( "context" "crypto/tls" "crypto/x509" "database/sql" "fmt" "io/ioutil" ) type TLSConfig struct { CertFile string KeyFile string CAFile string ServerName string InsecureSkip bool } func LoadTLSConfig(cfg *TLSConfig) (*tls.Config, error) { tlsCfg := &tls.Config{ ServerName: cfg.ServerName, InsecureSkipVerify: cfg.InsecureSkip, } if cfg.CAFile != "" { caCert, err := ioutil.ReadFile(cfg.CAFile) if err != nil { return nil, fmt.Errorf("failed to read CA file: %w", err) } caCertPool := x509.NewCertPool() caCertPool.AppendCertsFromPEM(caCert) tlsCfg.RootCAs = caCertPool } if cfg.CertFile != "" && cfg.KeyFile != "" { cert, err := tls.LoadX509KeyPair(cfg.CertFile, cfg.KeyFile) if err != nil { return nil, fmt.Errorf("failed to load cert: %w", err) } tlsCfg.Certificates = []tls.Certificate{cert} } return tlsCfg, nil } func ConnectWithTLS(ctx context.Context, dsn string, tlsCfg *tls.Config) (*sql.DB, error) { db, err := sql.Open("mysql", dsn) if err != nil { return nil, fmt.Errorf("failed to open database: %w", err) } if tlsCfg != nil { // MySQL driver specific TLS configuration // db.SetTLS(tlsCfg) } if err := db.PingContext(ctx); err != nil { return nil, fmt.Errorf("failed to ping database: %w", err) } return db, nil }

三、审计日志

3.1 审计日志表设计

CREATE TABLE audit_log ( id BIGSERIAL PRIMARY KEY, event_time TIMESTAMP WITH TIME ZONE DEFAULT NOW(), user_name VARCHAR(100), session_id VARCHAR(100), table_name VARCHAR(100), operation VARCHAR(20), old_data JSONB, new_data JSONB, application_name VARCHAR(100), client_addr INET, query_text TEXT ); -- 创建索引 CREATE INDEX idx_audit_log_time ON audit_log (event_time); CREATE INDEX idx_audit_log_user ON audit_log (user_name); CREATE INDEX idx_audit_log_table ON audit_log (table_name);

3.2 Go语言审计实现

package dbsecurity import ( "context" "database/sql" "encoding/json" "fmt" "net" "strings" "time" ) type AuditLogger struct { db *sql.DB } func NewAuditLogger(db *sql.DB) *AuditLogger { return &AuditLogger{db: db} } type AuditEvent struct { UserName string SessionID string TableName string Operation string OldData map[string]interface{} NewData map[string]interface{} ApplicationName string ClientAddr string QueryText string } func (al *AuditLogger) Log(ctx context.Context, event *AuditEvent) error { oldJSON, _ := json.Marshal(event.OldData) newJSON, _ := json.Marshal(event.NewData) query := ` INSERT INTO audit_log ( user_name, session_id, table_name, operation, old_data, new_data, application_name, client_addr, query_text ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) ` _, err := al.db.ExecContext(ctx, query, event.UserName, event.SessionID, event.TableName, event.Operation, oldJSON, newJSON, event.ApplicationName, event.ClientAddr, event.QueryText, ) return err } type AuditQueryInterceptor struct { logger *AuditLogger user string session string } func NewAuditQueryInterceptor(logger *AuditLogger, user, session string) *AuditQueryInterceptor { return &AuditQueryInterceptor{ logger: logger, user: user, session: session, } } func (aqi *AuditQueryInterceptor) WrapQuery(ctx context.Context, query string, oldData, newData map[string]interface{}) error { tableName := extractTableName(query) event := &AuditEvent{ UserName: aqi.user, SessionID: aqi.session, TableName: tableName, Operation: extractOperation(query), OldData: oldData, NewData: newData, ApplicationName: "app", ClientAddr: getClientIP(ctx), QueryText: query, } return aqi.logger.Log(ctx, event) } func extractTableName(query string) string { words := strings.Fields(query) if len(words) < 2 { return "" } switch strings.ToUpper(words[0]) { case "INSERT": if len(words) > 2 { return strings.TrimSuffix(words[2], "(") } case "UPDATE": return words[1] case "DELETE": if len(words) > 2 { return words[2] } case "SELECT": for i, w := range words { if strings.ToUpper(w) == "FROM" && i+1 < len(words) { return words[i+1] } } } return "" } func extractOperation(query string) string { words := strings.Fields(query) if len(words) > 0 { return strings.ToUpper(words[0]) } return "" } func getClientIP(ctx context.Context) string { if addr, ok := ctx.Value("client_addr").(net.Addr); ok { return addr.String() } return "" }

四、备份与恢复

4.1 自动备份策略

package dbsecurity import ( "context" "fmt" "os" "os/exec" "path/filepath" "time" ) type BackupManager struct { dbHost string dbPort string dbUser string dbPassword string dbName string backupDir string retention int } func NewBackupManager(host, port, user, password, dbName, backupDir string, retention int) *BackupManager { return &BackupManager{ dbHost: host, dbPort: port, dbUser: user, dbPassword: password, dbName: dbName, backupDir: backupDir, retention: retention, } } func (bm *BackupManager) CreateFullBackup(ctx context.Context) (string, error) { timestamp := time.Now().Format("20060102_150405") filename := fmt.Sprintf("%s_full_%s.sql", bm.dbName, timestamp) filepath := filepath.Join(bm.backupDir, filename) cmd := exec.CommandContext(ctx, "pg_dump", "-h", bm.dbHost, "-p", bm.dbPort, "-U", bm.dbUser, "-F", "c", "-f", filepath, bm.dbName, ) cmd.Env = append(os.Environ(), fmt.Sprintf("PGPASSWORD=%s", bm.dbPassword)) if err := cmd.Run(); err != nil { return "", fmt.Errorf("backup failed: %w", err) } go bm.cleanupOldBackups() return filepath, nil } func (bm *BackupManager) CreateIncrementalBackup(ctx context.Context) (string, error) { timestamp := time.Now().Format("20060102_150405") walDir := filepath.Join(bm.backupDir, "wal") os.MkdirAll(walDir, 0755) filename := fmt.Sprintf("%s_incr_%s.tar", bm.dbName, timestamp) filepath := filepath.Join(walDir, filename) cmd := exec.CommandContext(ctx, "pg_basebackup", "-h", bm.dbHost, "-p", bm.dbPort, "-U", bm.dbUser, "-D", filepath, "-Ft", "-P", "-z", ) cmd.Env = append(os.Environ(), fmt.Sprintf("PGPASSWORD=%s", bm.dbPassword)) if err := cmd.Run(); err != nil { return "", fmt.Errorf("incremental backup failed: %w", err) } return filepath, nil } func (bm *BackupManager) Restore(ctx context.Context, backupFile string) error { cmd := exec.CommandContext(ctx, "pg_restore", "-h", bm.dbHost, "-p", bm.dbPort, "-U", bm.dbUser, "-d", bm.dbName, "-c", backupFile, ) cmd.Env = append(os.Environ(), fmt.Sprintf("PGPASSWORD=%s", bm.dbPassword)) if err := cmd.Run(); err != nil { return fmt.Errorf("restore failed: %w", err) } return nil } func (bm *BackupManager) cleanupOldBackups() { entries, err := os.ReadDir(bm.backupDir) if err != nil { return } type backupFile struct { path string modTime time.Time } var backups []backupFile for _, entry := range entries { if entry.IsDir() { continue } info, err := entry.Info() if err != nil { continue } backups = append(backups, backupFile{ path: filepath.Join(bm.backupDir, entry.Name()), modTime: info.ModTime(), }) } if len(backups) <= bm.retention { return } for i := 0; i < len(backups)-bm.retention; i++ { os.Remove(backups[i].path) } }

五、合规检查

5.1 GDPR合规检查

package compliance import ( "context" "database/sql" "fmt" ) type GDPRCompliance struct { db *sql.DB } func NewGDPRCompliance(db *sql.DB) *GDPRCompliance { return &GDPRCompliance{db: db} } type DataSubject struct { UserID string Email string HasData bool DataTypes []string } func (gc *GDPRCompliance) GetDataSubject(ctx context.Context, email string) (*DataSubject, error) { ds := &DataSubject{ Email: email, HasData: false, DataTypes: make([]string, 0), } tables := []string{"users", "orders", "activities", "communications"} for _, table := range tables { var count int query := fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE email = $1", table) if err := gc.db.QueryRowContext(ctx, query, email).Scan(&count); err != nil { continue } if count > 0 { ds.HasData = true ds.DataTypes = append(ds.DataTypes, table) } } return ds, nil } func (gc *GDPRCompliance) ExportData(ctx context.Context, userID string) (map[string]interface{}, error) { export := make(map[string]interface{}) tables := map[string]string{ "users": "user_id", "orders": "user_id", "activities": "user_id", "preferences": "user_id", } for table, column := range tables { query := fmt.Sprintf("SELECT * FROM %s WHERE %s = $1", table, column) rows, err := gc.db.QueryContext(ctx, query, userID) if err != nil { continue } columns, _ := rows.Columns() var records []map[string]interface{} for rows.Next() { values := make([]interface{}, len(columns)) ptrs := make([]interface{}, len(columns)) for i := range values { ptrs[i] = &values[i] } if err := rows.Scan(ptrs...); err != nil { continue } record := make(map[string]interface{}) for i, col := range columns { record[col] = values[i] } records = append(records, record) } rows.Close() if len(records) > 0 { export[table] = records } } return export, nil } func (gc *GDPRCompliance) DeleteUserData(ctx context.Context, userID string) error { tables := []string{"activities", "preferences", "sessions", "orders", "users"} for _, table := range tables { query := fmt.Sprintf("DELETE FROM %s WHERE user_id = $1", table) if _, err := gc.db.ExecContext(ctx, query, userID); err != nil { return fmt.Errorf("failed to delete from %s: %w", table, err) } } return nil }