sqlmapapi を使用してスキャンを開始する方法

sqlmap は SQL インジェクション検出の成果物であると言えますが、sqlmap を使用して SQL インジェクションをテストする効率は非常に低く、各 URL を手動でテストする必要があります。 sqlmap の開発者は、インターフェイス呼び出しを通じて直接操作できる sqlmapapi.py を追加し、sqlmap コマンドの実行を簡素化しました。

sqlmap API はサーバーとクライアントに分かれており、HTTP プロトコルに基づくインターフェイス モードとコマンド ラインに基づくインターフェイス モードの 2 つのモードがあります。

sqlmap ソース コードのダウンロード アドレス: https://github.com/sqlmapproject/sqlmap/

1. ヘルプを表示

python sqlmapapi.py -h

2. API サーバーを起動します

API サービスを使用する前に、HTTP プロトコルまたはコマンド ライン ベースのインターフェイス モードに基づいて API サーバーを起動する必要があります。 API サーバーを起動するには、次のコマンドを実行するだけです: python sqlmapapi.py -s

コマンドが成功すると、コマンド ラインにいくつかの情報が返されます。次のコマンドは、API サーバーがローカル ポート 8775 で実行され、管理トークンが c6bbb0c1f86b7d7bc2ed6ce3e3bbdcb5 などであることを大まかに意味します。

ただし、上記の方法で API サーバーを開くには欠点があります。サーバーとクライアントがホストでない場合、クライアントは接続されないため、この問題を解決したい場合は、次のコマンドを入力して API サーバーを開くことができます: python sqlmapapi.py -s -H " 0.0.0.0" -p 8775

コマンドが成功すると、リモート クライアントはリモート ホストの IP とポートを指定して API サーバーに接続できます。

3. コマンドラインベースのインターフェイスモード

3.1. クライアントを開いてインジェクションコマンドを開始します

python sqlmapapi.py -c

クライアントとサーバーが同じコンピューター上にない場合は、次のコマンドを入力します:

#python sqlmapapi.py -c -H "192.168.1.101" -p 8775

3.2.ヘルプコマンド、すべてのコマンドを取得

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

3.3.検出インジェクション

3.3.1.新しいコマンド

new -u "url"

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

指定するだけですが-u パラメーターを使用していますが、返された情報から、新しいコマンドを入力した後、最初に /task/new に新しいタスク ID を作成するよう要求し、次にタスクを開始する要求を開始したことがわかります。このモードの本質も HTTP プロトコルに基づいています。

3.3.2. status コマンド

タスクのスキャン ステータスを取得します。返されたコンテンツのステータス フィールドが終了しました。スキャンが完了したことを意味します。返されたコンテンツのステータス フィールドが実行されている場合は、スキャンがまだ進行中であることを意味します。次の図は、完了したスキャンのスクリーンショットです:

3.3.3. data コマンド

返されたデータのデータ フィールドが空でない場合、注射が成功したことを確認できます。この例は、SQL インジェクションを含む戻りコンテンツを示しています。これには、データベース タイプ、ペイロード、インジェクション パラメーターなどの情報が含まれています。

4. HTTP プロトコルに基づくインターフェイス モード

sqlmapapi.py の http インターフェイス呼び出しモードに基づく h の主な機能を簡単に紹介します。lib/ と入力します。 utils/api.py のサーバー クラスは、サーバーにデータを送信することでサービスと対話することがわかります。全部で3種類あります。

ユーザーのメソッド ユーザー メソッド

管理関数 管理関数

sqlmap コア対話関数コア対話関数

はい送信されるデータの種類は次のとおりです:

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 ファイルの呼び出し関係。これらの操作はテストのニーズを完全に満たすことができるため、バッチ操作に使用できます。

5. sqlmapapi を使用してスキャンを開始します

sqlmapapi.py は便利な http リクエスト エントリを提供しますが、使用すると、注入するかどうかの最終結果しか取得できません。インターフェイスは注入スキャンを実行しますか? このようなリクエストは、リクエストがいくら多くても取得するのが難しい場合があります。sqlmap ソース コードを調べた後に記録されたフローチャートを共有しましょう。この図から、ペイロード レベルのリクエストの特定の場所を特定できます。開始されたリクエストの種類とリクエストの数を知りたい場合は、ここにカスタム コードを追加するだけで済みます。

#6. SQL インジェクション自動化実装プロセス

#

以上がsqlmapapi を使用してスキャンを開始する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。