第 10 章 · 安全与 Python 连接

import pymysql

username = input("用户名：")
password = input("密码：")

conn = pymysql.connect(host="127.0.0.1", port=3306, user="hiwen_app",
                       password="App@Pass2026", database="hiwen", charset="utf8mb4")
try:
    with conn.cursor() as cur:
        # ✗ 反面教材：把用户输入用 + 直接拼进 SQL
        sql = ("SELECT id, username, nickname FROM user "
               "WHERE username = '" + username + "' AND password = '" + password + "'")
        cur.execute(sql)
        row = cur.fetchone()
        print("登录成功：" + str(row) if row else "用户名或密码错误")
finally:
    conn.close()

-- 攻击者输入： 用户名 = admin' -- （注意 -- 后面有一个空格）
-- 拼出来的语句变成了这样：
SELECT id, username, nickname FROM user
WHERE username = 'admin' -- ' AND password = '随便填的';

-- 在 MySQL 里 -- 之后是注释，整段「密码校验」被注释掉了！
-- 真实执行的只有：
SELECT id, username, nickname FROM user WHERE username = 'admin';

-- ① 准备：把 SQL 模板交给服务器编译，? 是占位符
PREPARE stmt FROM 'SELECT id, title, author FROM article WHERE id = ?';

-- ② 给占位符赋值（用用户变量传值）
SET @aid = 1;

-- ③ 执行：把值填进模板
EXECUTE stmt USING @aid;

-- ④ 释放：不再使用就清掉，避免占用服务器资源
DEALLOCATE PREPARE stmt;

PREPARE stmt FROM 'SELECT id, username FROM user WHERE username = ?';

-- 把攻击者的输入当成「一个普通字符串」传进去
SET @name = 'admin\'-- ';

EXECUTE stmt USING @name;

DEALLOCATE PREPARE stmt;

-- ① 建一个只允许本机连接的应用账号
CREATE USER 'hiwen_app'@'localhost' IDENTIFIED BY 'App@Pass2026';

-- ② 只授「增删改查」，不给 CREATE / DROP / ALTER / GRANT
GRANT SELECT, INSERT, UPDATE, DELETE ON hiwen.* TO 'hiwen_app'@'localhost';

-- ③ 让权限立即生效
FLUSH PRIVILEGES;

-- ④ 检查一下到底给了什么
SHOW GRANTS FOR 'hiwen_app'@'localhost';

-- 假设一开始图省事给了全部权限
GRANT ALL PRIVILEGES ON hiwen.* TO 'hiwen_app'@'localhost'; -- ✗ 太宽了

-- 全部收回，再按需要最小化地授
REVOKE ALL PRIVILEGES ON hiwen.* FROM 'hiwen_app'@'localhost';
GRANT SELECT, INSERT, UPDATE, DELETE ON hiwen.* TO 'hiwen_app'@'localhost';
FLUSH PRIVILEGES;

-- 账号不用了就删掉
DROP USER 'hiwen_app'@'localhost';

# ① 导出整个 hiwen 库（结构 + 数据）到带日期的文件
mysqldump -uroot -p --databases hiwen > backup_hiwen_20261012.sql

# ② 只要结构不要数据（想拿一份干净的建表语句时用）
mysqldump -uroot -p --no-data hiwen > hiwen_schema_only.sql

# ③ 只导某几张表
mysqldump -uroot -p hiwen article category > backup_two_tables.sql

# ④ 确认文件生成成功
ls -lh *.sql

-- 先登录：  mysql -uroot -p
-- 再执行（source 后面建议写绝对路径，避免找不到文件）
source /Users/you/backup_hiwen_20261012.sql;

-- 验证数据是否回来了
SELECT COUNT(*) AS 作品数 FROM article;

# 建议装在虚拟环境里
python -m venv .venv
source .venv/bin/activate

pip install pymysql

import pymysql

# ① 建立连接
conn = pymysql.connect(
    host="127.0.0.1",       # 数据库地址
    port=3306,              # 端口
    user="hiwen_app",       # 应用专用账号
    password="App@Pass2026",
    database="hiwen",       # 默认库
    charset="utf8mb4",      # 一定要写，否则中文乱码
)

# ② 用游标执行 SQL 并取结果
with conn.cursor() as cur:
    cur.execute("SELECT id, title, author, dynasty FROM article ORDER BY views DESC LIMIT 3")
    rows = cur.fetchall()
    for row in rows:
        print(row)

# ③ 用完必须关连接
conn.close()

dynasty = "唐"

# ========== ✗ 错误：把用户输入拼进 SQL ==========
cur.execute("SELECT title, author FROM article WHERE dynasty = '" + dynasty + "'")
print("拼接写法：", cur.fetchall())

# ========== ✓ 正确：占位符 %s + 参数元组 ==========
cur.execute("SELECT title, author FROM article WHERE dynasty = %s", (dynasty,))
#                                                                    ↑ 注意这个逗号
print("参数化写法：", cur.fetchall())

# 多个参数就多放几个占位符，按位置一一对应
cur.execute(
    "SELECT title FROM article WHERE dynasty = %s AND views > %s ORDER BY views DESC",
    (dynasty, 200),
)
print("两个参数：", cur.fetchall())

import pymysql

conn = pymysql.connect(host="127.0.0.1", port=3306, user="hiwen_app",
                       password="App@Pass2026", database="hiwen", charset="utf8mb4")
try:
    with conn.cursor() as cur:
        # ① 正常筛选：唐朝作品，浏览量从高到低
        cur.execute(
            "SELECT title, author, views FROM article WHERE dynasty = %s ORDER BY views DESC",
            ("唐",),
        )
        for row in cur.fetchall():
            print(row)

        # ② 把注入 payload 当成朝代名传进去
        evil = "唐' OR '1'='1"
        cur.execute("SELECT COUNT(*) AS n FROM article WHERE dynasty = %s", (evil,))
        print("注入测试结果：", cur.fetchone())
finally:
    conn.close()

# 接口参数：order_by = latest / hot / likes（对应接口文档里的作品列表）
ORDER_MAP = {
    "latest": "id DESC",
    "hot":    "views DESC",
    "likes":  "likes DESC",
}

def build_order(order_by: str) -> str:
    # 不在白名单里的值，一律回落到默认排序，绝不拼进 SQL
    return ORDER_MAP.get(order_by, "id DESC")

cur.execute(f"SELECT id, title, views FROM v_article_list ORDER BY {build_order('hot')} LIMIT 3")
for row in cur.fetchall():
    print(row)

import pymysql

conn = pymysql.connect(host="127.0.0.1", port=3306, user="hiwen_app",
                       password="App@Pass2026", database="hiwen", charset="utf8mb4")
try:
    with conn.cursor() as cur:
        # ① 增：用户 1 收藏作品 5
        cur.execute("INSERT INTO favorite (user_id, article_id) VALUES (%s, %s)", (1, 5))
        print("插入行数：", cur.rowcount, "新记录 id：", cur.lastrowid)

        # ② 改：浏览量 +1
        cur.execute("UPDATE article SET views = views + 1 WHERE id = %s", (5,))
        print("更新行数：", cur.rowcount)

        # ③ 查
        cur.execute("SELECT title, views FROM article WHERE id = %s", (5,))
        print("查询结果：", cur.fetchone())

        # ④ 删：取消刚才那次收藏
        cur.execute("DELETE FROM favorite WHERE user_id = %s AND article_id = %s", (1, 5))
        print("删除行数：", cur.rowcount)

    conn.commit()                 # 四步全部成功，一起提交
    print("事务已提交")
except Exception as e:
    conn.rollback()               # 任何一步出错，全部撤销
    print("出错已回滚：", e)
finally:
    conn.close()

import pymysql

DB_CONF = dict(host="127.0.0.1", port=3306, user="hiwen_app",
               password="App@Pass2026", database="hiwen", charset="utf8mb4")

# 连接对象支持 with：正常退出自动 commit，异常自动 rollback，并自动关闭
with pymysql.connect(**DB_CONF) as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT COUNT(*) AS n FROM article")
        print(cur.fetchone())