import sqlite3  # 輕量級(jí)數(shù)據(jù)庫(kù)
import mysql.connector  # MySQL數(shù)據(jù)庫(kù)
import psycopg2  # PostgreSQL數(shù)據(jù)庫(kù)

# SQLite連接示例
conn = sqlite3.connect('mydatabase.db')
cursor = conn.cursor()

# MySQL連接示例
conn = mysql.connector.connect(
    host="localhost",
    user="root",
    password="secret",
    database="mydb"
)

# 創(chuàng)建表
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    email TEXT UNIQUE
)
""")

# 插入數(shù)據(jù) - UNSAFE方式（存在注入風(fēng)險(xiǎn)）
cursor.execute(f"INSERT INTO users (name, email) VALUES ('{name}', '{email}')")

# 插入數(shù)據(jù) - SAFE參數(shù)化查詢
cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", (name, email))

# 查詢數(shù)據(jù)
cursor.execute("SELECT * FROM users WHERE email = ?", (target_email,))
results = cursor.fetchall()

# 更新數(shù)據(jù)
cursor.execute("UPDATE users SET name = ? WHERE id = ?", (new_name, user_id))

# 刪除數(shù)據(jù)
cursor.execute("DELETE FROM users WHERE id = ?", (user_id,))

conn.commit()  # 重要！提交事務(wù)

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

# 定義數(shù)據(jù)模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100), unique=True)

# 初始化數(shù)據(jù)庫(kù)連接
engine = create_engine('sqlite:///mydatabase.db')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# CRUD操作示例
# 創(chuàng)建
new_user = User(name="張三", email="zhangsan@example.com")
session.add(new_user)

# 查詢
user = session.query(User).filter_by(email="zhangsan@example.com").first()

# 更新
user.name = "李四"
session.commit()

# 刪除
session.delete(user)
session.commit()

# 使用sqlparse庫(kù)進(jìn)行SQL語(yǔ)法分析
import sqlparse

def validate_sql(query):
    parsed = sqlparse.parse(query)
    if not parsed:
        raise ValueError("空SQL語(yǔ)句")
    
    first_token = parsed[0].tokens[0].value.upper()
    if first_token not in ["SELECT", "INSERT", "UPDATE", "DELETE"]:
        raise ValueError(f"無(wú)效的SQL命令: {first_token}")
    
    # 檢查是否存在直接字符串拼接
    if "'" in query or '"' in query:
        print("警告：可能存在字符串拼接風(fēng)險(xiǎn)，建議使用參數(shù)化查詢")

# 用戶輸入： ' OR 1=1; --
cursor.execute(f"SELECT * FROM users WHERE email = '{input_email}'")
# 實(shí)際執(zhí)行： SELECT * FROM users WHERE email = '' OR 1=1; --'

# 安全
cursor.execute("SELECT * FROM users WHERE email = %s", (input_email,))

import re
if not re.match(r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$", email):
    raise ValueError("無(wú)效郵箱格式")

CREATE USER app_user WITH PASSWORD 'strong_pwd';
GRANT SELECT, INSERT ON users TO app_user; -- 僅授予必要權(quán)限

# 使用sqlmap進(jìn)行自動(dòng)化檢測(cè)（僅用于測(cè)試?。?
sqlmap -u "http://example.com/?id=1" --risk=3 --level=5

# 偽代碼示例：使用OpenAI API
response = openai.Completion.create(
  model="text-davinci-003",
  prompt=f"將自然語(yǔ)言轉(zhuǎn)換為SQL: 查詢姓張的用戶",
)
# 輸出： SELECT * FROM users WHERE name LIKE '張%'