如何利用sqlmapapi發起掃描

本篇文章為大家展示了如何利用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服務端

無論是基于HTTP協議的接口模式還是基于命令行的接口模式，首先都是需要開啟api服務端的。通過輸入以下命令即可開啟api服務端:python sqlmapapi.py -s

命令成功后，在命令行中會返回一些信息。以下命令大概的意思是api服務端在本地8775端口上運行，admin token為c6bbb0c1f86b7d7bc2ed6ce3e3bbdcb5等

但是通過上面的這種方式開啟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          退出客戶端

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"})

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})

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

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

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

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

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

從sqlmapapi.py文件分析，提取調用關系。不難發現這些操作可以完全滿足我們的測試需求,因此利用這些就可以批量了。

五、利用sqlmapapi發起掃描

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

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

上述內容就是如何利用sqlmapapi發起掃描，你們學到知識或技能了嗎？如果還想學到更多技能或者豐富自己的知識儲備，歡迎關注億速云行業資訊頻道。