如何利用sqlmapapi發起掃描

sqlmap可謂是sql注入探測的神器，但是利用sqlmap測試SQL注入的效率很低,每一個url都需要手動測試。 sqlmap的開發者新加了sqlmapapi.py，可以直接透過介面呼叫來操作，簡化了sqlmap指令執行方式。

sqlmap api分為服務端和客戶端，sqlmap api有兩種模式，一種是基於HTTP協定的介面模式，一種是基於命令列的介面模式。

sqlmap原始碼下載位址：https://github.com/sqlmapproject/sqlmap/

一、查看幫助

python sqlmapapi.py -h

二、開啟api服務端

在使用API​​服務之前，需要先啟動API伺服器，無論是基於HTTP協定還是基於命令列的介面模式。要開啟API服務端，只需執行以下命令：python sqlmapapi.py -s

命令成功後，在命令列中會傳回一些資訊。以下指令大概的意思是api服務端在本地8775埠上運行，admin token為c6bbb0c1f86b7d7bc2ed6ce3e3bbdcb5等

但是透過上面的這種方式開啟api服務端有一個缺點，當服務端和客戶端不是一台主機機會連線不上，因此如果要解決這個問題，可以透過輸入以下指令開啟api服務端:python sqlmapapi.py -s -H "0.0.0.0" -p 87775

指令成功後，遠端客戶端就可以透過指定遠端主機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          退出客户端

#3.3、偵測注入

##3.3.1.new指令

new -u "url"

#實例：new -u "http://www.baidu.com"

#雖然我們只指定了-u參數，但是從傳回的資訊可以看出，輸入new指令後，首先請求了/task/new，來建立一個新的taskid，後又發起了一個請求去開始任務，因此可以發現該模式實質也是基於HTTP協定的。

3.3.2. status 指令

取得該任務的掃描狀態，若傳回內容中的status欄位為terminated，說明掃描完成，若返回內容中的status欄位為run，表示掃描仍在進行中。下圖是掃描完成的截圖：

3.3.3. data 指令

如果傳回的資料中data欄位非空，則可以得到注入已經成功。這個範例展示了一個含有 SQL 注入的回傳內容，其中包含資料庫類型、payload 和注入參數等資訊。

四、基於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")

@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/delete")

@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"})</taskid>

4.2、核心互動函數

#@get("/option/list")

@post("/ option/get")

@post("/option/set")

@post("/option/<taskid>/set")
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})</taskid>

@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()})</taskid>

@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})</taskid>

@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})</taskid>

@get("/scan/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()
    })</taskid>

@ get("/scan/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})</taskid>

@get("/scan/log")

@get("/download/")

4.3、管理函數

@get("/admin/list")

@get("/admin/list")
@get("/admin/<token>/list")
def task_list(token=None):
    """
    Pull task list
    """
    tasks = {}
 
    for key in DataStore.tasks:
        if is_admin(token) or DataStore.tasks[key].remote_addr == request.remote_addr:
            tasks[key] = dejsonize(scan_status(key))["status"]
 
    logger.debug("(%s) Listed task pool (%s)" % (token, "admin" if is_admin(token) else request.remote_addr))
    return jsonize({"success": True, "tasks": tasks, "tasks_num": len(tasks)})</token>

@get("/admin//flush")

@get("/admin/flush")
@get("/admin/<token>/flush")
def task_flush(token=None):
    """
    Flush task spool (delete all tasks)
    """
 
    for key in list(DataStore.tasks):
        if is_admin(token) or DataStore.tasks[key].remote_addr == request.remote_addr:
            DataStore.tasks[key].engine_kill()
            del DataStore.tasks[key]
 
    logger.debug("(%s) Flushed task pool (%s)" % (token, "admin" if is_admin(token) else request.remote_addr))
    return jsonize({"success": True})</token>

從sqlmapapi.py檔案分析，提取調用關係。這些操作可以充分滿足我們的測試需求，因此可以利用它們來進行大量操作。

五、利用sqlmapapi發起掃描

sqlmapapi.py很方便的提供了http請求入口，但是使用起來只能得到最終是否注入的結果，每個接口進行注入掃描時具體發起了什麼樣的請求，多少個請求就難以得到，下面分享一下個人梳理sqlmap源碼後記錄的流程圖。從圖中可以定位到具體發起payload等級的請求位置，從而想要取得發起了什麼樣的請求，多少個請求，只需要在此處增加自訂程式碼即可。

六、sql注入自動化實作流程

#

以上是如何利用sqlmapapi發起掃描的詳細內容。更多資訊請關注PHP中文網其他相關文章！