sqlmap可谓是sql注入探测的神器，但是利用sqlmap测试SQL注入的效率很低,每一个url都需要手动测试。sqlmap的开发者新加了sqlmapapi.py，可以直接通过接口调用来操作，简化了sqlmap命令执行方式。

sqlmap api分为服务端和客户端，sqlmap api有两种模式，一种是基于HTTP协议的接口模式，一种是基于命令行的接口模式。

sqlmap源码下载地址：[https://github.com/sqlmapproject/sqlmap/]

一、查看帮助

• python sqlmapapi.py -h

image.png

二、开启api服务端

无论是基于HTTP协议的接口模式还是基于命令行的接口模式，首先都是需要开启api服务端的。通过输入以下命令即可开启api服务端:
*python sqlmapapi.py -s
命令成功后，在命令行中会返回一些信息。以下命令大概的意思是api服务端在本地8775端口上运行，admin token为c6bbb0c1f86b7d7bc2ed6ce3e3bbdcb5等

image.png

但是通过上面的这种方式开启api服务端有一个缺点，当服务端和客户端不是一台主机会连接不上，因此如果要解决这个问题，可以通过输入以下命令来开启api服务端:
python sqlmapapi.py -s -H "0.0.0.0" -p 8775
命令成功后，远程客户端就可以通过指定远程主机IP和端口来连接到API服务端。

三、基于命令行的接口模式

3.1、开启客户端，发起注入命令

*python sqlmapapi.py -c
如果是客户端和服务端不是同一台计算机的话，输入以下命令：
*python sqlmapapi.py -c -H "192.168.1.101" -p 8775

3.2、help命令，获取所有命令

help          显示帮助信息
new ARGS      开启一个新的扫描任务
use TASKID    切换
taskid data   获取当前任务返回的数据
log           获取当前任务的扫描日志
status        获取当前任务的扫描状态
option OPTION 获取当前任务的选项
options       获取当前任务的所有配置信息
stop          停止当前任务 
kill          杀死当前任务
list          显示所有任务列表
flush         清空所有任务
exit          退出客户端

image.png

3.3、检测注入

3.3.1.new命令

• new -u "url"
• 实例：new -u "http://www.baidu.com"

虽然我们仅仅只指定了-u参数，但是从返回的信息中可以看出，输入new命令后，首先先请求了/task/new，来创建一个新的taskid，后又发起了一个请求去开始任务，因此可以发现该模式实质也是基于HTTP协议的。

image.png

3.3.2. status 命令

获取该任务的扫描状态，若返回内容中的status字段为terminated，说明扫描完成，若返回内容中的status字段为run，说明扫描还在进行中。下图是扫描完成的截图：

image.png

3.3.3. data 命令

获取扫描完成后注入出来的信息，若返回的内容中data字段不为空就说明存在注入。下图是存在SQL注入返回的内容，可以看到返回的内容有数据库类型、payload、注入的参数等等。

image.png

四、基于HTTP协议的接口模式

简单介绍下sqlmapapi.py的h基于http接口调用模式的主要函数，进入到lib/utils/api.py的server类,可以发现通过向server提交数据进行与服务的交互。 一共分为3种类型。

Users' methods 用户方法
Admin function 管理函数
sqlmap core interact functions 核心交互函数

可以提交数据的种类如下：

4.1、用户方法

*新建任务

@get("/task/new")
def task_new():
    """
    Create a new task
    """
    taskid = encodeHex(os.urandom(8), binary=False)
    remote_addr = request.remote_addr
 
    DataStore.tasks[taskid] = Task(taskid, remote_addr)
 
    logger.debug("Created new task: '%s'" % taskid)
    return jsonize({"success": True, "taskid": taskid})

*删除任务

@get("/task/<taskid>/delete")
def task_delete(taskid):
    """
    Delete an existing task
    """
    if taskid in DataStore.tasks:
        DataStore.tasks.pop(taskid)
 
        logger.debug("(%s) Deleted task" % taskid)
        return jsonize({"success": True})
    else:
        response.status = 404
        logger.warning("[%s] Non-existing task ID provided to task_delete()" % taskid)
        return jsonize({"success": False, "message": "Non-existing task ID"})

4.2、核心交互函数
@get("/option/list")
@post("/option/get")
@post("/option/set")

*@post("/option/<taskid>/set") 设置option

def option_set(taskid):
    """
    Set value of option(s) for a certain task ID
    """
 
    if taskid not in DataStore.tasks:
        logger.warning("[%s] Invalid task ID provided to option_set()" % taskid)
        return jsonize({"success": False, "message": "Invalid task ID"})
 
    if request.json is None:
        logger.warning("[%s] Invalid JSON options provided to option_set()" % taskid)
        return jsonize({"success": False, "message": "Invalid JSON options"})
 
    for option, value in request.json.items():
        DataStore.tasks[taskid].set_option(option, value)
 
    logger.debug("(%s) Requested to set options" % taskid)
    return jsonize({"success": True}

• @post("/scan/start") 启动扫描

@post("/scan/<taskid>/start")
def scan_start(taskid):
    """
    Launch a scan
    """
 
    if taskid not in DataStore.tasks:
        logger.warning("[%s] Invalid task ID provided to scan_start()" % taskid)
        return jsonize({"success": False, "message": "Invalid task ID"})
 
    if request.json is None:
        logger.warning("[%s] Invalid JSON options provided to scan_start()" % taskid)
        return jsonize({"success": False, "message": "Invalid JSON options"})
 
    # Initialize sqlmap engine's options with user's provided options, if any
    for option, value in request.json.items():
        DataStore.tasks[taskid].set_option(option, value)
 
    # Launch sqlmap engine in a separate process
    DataStore.tasks[taskid].engine_start()
 
    logger.debug("(%s) Started scan" % taskid)
    return jsonize({"success": True, "engineid": DataStore.tasks[taskid].engine_get_id()})

*@get("/scan/stop") 停止扫描

@get("/scan/<taskid>/stop")
def scan_stop(taskid):
    """
    Stop a scan
    """
 
    if (taskid not in DataStore.tasks or DataStore.tasks[taskid].engine_process() is None or DataStore.tasks[taskid].engine_has_terminated()):
        logger.warning("[%s] Invalid task ID provided to scan_stop()" % taskid)
        return jsonize({"success": False, "message": "Invalid task ID"})
 
    DataStore.tasks[taskid].engine_stop()
 
    logger.debug("(%s) Stopped scan" % taskid)
    return jsonize({"success": True})

*@get("/scan/kill") 杀死任务

@get("/scan/<taskid>/kill")
def scan_kill(taskid):
    """
    Kill a scan
    """
 
    if (taskid not in DataStore.tasks or DataStore.tasks[taskid].engine_process() is None or DataStore.tasks[taskid].engine_has_terminated()):
        logger.warning("[%s] Invalid task ID provided to scan_kill()" % taskid)
        return jsonize({"success": False, "message": "Invalid task ID"})
 
    DataStore.tasks[taskid].engine_kill()
 
    logger.debug("(%s) Killed scan" % taskid)
    return jsonize({"success": True})

*@get("/scan/status") 获取扫描status

@get("/scan/<taskid>/status")
def scan_status(taskid):
    """
    Returns status of a scan
    """
 
    if taskid not in DataStore.tasks:
        logger.warning("[%s] Invalid task ID provided to scan_status()" % taskid)
        return jsonize({"success": False, "message": "Invalid task ID"})
 
    if DataStore.tasks[taskid].engine_process() is None:
        status = "not running"
    else:
        status = "terminated" if DataStore.tasks[taskid].engine_has_terminated() is True else "running"
 
    logger.debug("(%s) Retrieved scan status" % taskid)
    return jsonize({
        "success": True,
        "status": status,
        "returncode": DataStore.tasks[taskid].engine_get_returncode()
    })

*@get("/scan/data") 获取扫描结果data

@get("/scan/<taskid>/data")
def scan_data(taskid):
    """
    Retrieve the data of a scan
    """
 
    json_data_message = list()
    json_errors_message = list()
 
    if taskid not in DataStore.tasks:
        logger.warning("[%s] Invalid task ID provided to scan_data()" % taskid)
        return jsonize({"success": False, "message": "Invalid task ID"})
 
    # Read all data from the IPC database for the taskid
    for status, content_type, value in DataStore.current_db.execute("SELECT status, content_type, value FROM data WHERE taskid = ? ORDER BY id ASC", (taskid,)):
        json_data_message.append({"status": status, "type": content_type, "value": dejsonize(value)})
 
    # Read all error messages from the IPC database
    for error in DataStore.current_db.execute("SELECT error FROM errors WHERE taskid = ? ORDER BY id ASC", (taskid,)):
        json_errors_message.append(error)
 
    logger.debug("(%s) Retrieved scan data and error messages" % taskid)
    return jsonize({"success": True, "data": json_data_message, "error": json_errors_message})

sqlmapapi还有其他一些接口，此处就不一一列举。
从sqlmapapi.py文件分析，提取调用关系。不难发现这些操作可以完全满足我们的测试需求,因此利用这些就可以批量了。