online_unzipper
import os
import uuid
from flask import Flask, request, redirect, url_for, send_file, render_template, session, send_from_directory, abort, Response
app = Flask(__name__)
app.secret_key = os.environ.get("FLASK_SECRET_KEY", "test_key")
UPLOAD_FOLDER = os.path.join(os.getcwd(), "uploads")
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
users = {}
@app.route("/")
def index():
if "username" not in session:
return redirect(url_for("login"))
return redirect(url_for("upload"))
@app.route("/register", methods=["GET", "POST"])
def register():
if request.method == "POST":
username = request.form["username"]
password = request.form["password"]
if username in users:
return "用户名已存在"
users[username] = {"password": password, "role": "user"}
return redirect(url_for("login"))
return render_template("register.html")
@app.route("/login", methods=["GET", "POST"])
def login():
if request.method == "POST":
username = request.form["username"]
password = request.form["password"]
if username in users and users[username]["password"] == password:
session["username"] = username
session["role"] = users[username]["role"]
return redirect(url_for("upload"))
else:
return "用户名或密码错误"
return render_template("login.html")
@app.route("/logout")
def logout():
session.clear()
return redirect(url_for("login"))
@app.route("/upload", methods=["GET", "POST"])
def upload():
if "username" not in session:
return redirect(url_for("login"))
if request.method == "POST":
file = request.files["file"]
if not file:
return "未选择文件"
role = session["role"]
if role == "admin":
dirname = request.form.get("dirname") or str(uuid.uuid4())
else:
dirname = str(uuid.uuid4())
target_dir = os.path.join(UPLOAD_FOLDER, dirname)
os.makedirs(target_dir, exist_ok=True)
zip_path = os.path.join(target_dir, "upload.zip")
file.save(zip_path)
try:
os.system(f"unzip -o {zip_path} -d {target_dir}")
except:
return "解压失败，请检查文件格式"
os.remove(zip_path)
return f"解压完成！<br>下载地址: <a href='{url_for('download', folder=dirname)}'>{request.host_url}download/{dirname}</a>"
return render_template("upload.html")
@app.route("/download/<folder>")
def download(folder):
target_dir = os.path.join(UPLOAD_FOLDER, folder)
if not os.path.exists(target_dir):
abort(404)
files = os.listdir(target_dir)
return render_template("download.html", folder=folder, files=files)
@app.route("/download/<folder>/<filename>")
def download_file(folder, filename):
file_path = os.path.join(UPLOAD_FOLDER, folder, filename)
try:
with open(file_path, 'r') as file:
content = file.read()
return Response(
content,
mimetype="application/octet-stream",
headers={
"Content-Disposition": f"attachment; filename={filename}"
}
)
except FileNotFoundError:
return "File not found", 404
except Exception as e:
return f"Error: {str(e)}", 500
if __name__ == "__main__":
app.run(host="0.0.0.0")
抓包解cookie，发现需要伪造session，需要secretkey(笑死我了lmtx)
注意到upload路由
try:
os.system(f"unzip -o {zip_path} -d {target_dir}")
except:
return "解压失败，请检查文件格式"
这不就是经典软链接吗
我直接读flag(屮，读不到)，估计不叫flag
那先伪造admin吧
需要secretkey，直接读环境变量
ln -s /proc/self/environ env
zip --symlinks a.zip env
我复现的时候的secretkey就是test
所以你直接flask-unsign就可以伪造出来了
然后重定向写入解压目录即可
这题比较简单
ping
import base64
import subprocess
import re
import ipaddress
import flask
def run_ping(ip_base64):
try:
decoded_ip = base64.b64decode(ip_base64).decode('utf-8')
if not re.match(r'^\d+\.\d+\.\d+\.\d+$', decoded_ip):
return False
if decoded_ip.count('.') != 3:
return False
if not all(0 <= int(part) < 256 for part in decoded_ip.split('.')):
return False
if not ipaddress.ip_address(decoded_ip):
return False
if len(decoded_ip) > 15:
return False
if not re.match(r'^[A-Za-z0-9+/=]+$', ip_base64):
return False
except Exception as e:
return False
command = f"""echo "ping -c 1 $(echo '{ip_base64}' | base64 -d)" | sh"""
try:
process = subprocess.run(
command,
shell=True,
check=True,
capture_output=True,
text=True
)
return process.stdout
except Exception as e:
return False
app = flask.Flask(__name__)
@app.route('/ping', methods=['POST'])
def ping():
data = flask.request.json
ip_base64 = data.get('ip_base64')
if not ip_base64:
return flask.jsonify({'error': 'no ip'}), 400
result = run_ping(ip_base64)
if result:
return flask.jsonify({'success': True, 'output': result}), 200
else:
return flask.jsonify({'success': False}), 400
@app.route('/')
def index():
return flask.render_template('index.html')
app.run(host='0.0.0.0', port=5000)
当时想着再蒸蒸就能出，但是还是去做志愿活动了(哈哈哈，也没办法)
try:
decoded_ip = base64.b64decode(ip_base64).decode('utf-8')
if not re.match(r'^\d+\.\d+\.\d+\.\d+$', decoded_ip):
return False
if decoded_ip.count('.') != 3:
return False
if not all(0 <= int(part) < 256 for part in decoded_ip.split('.')):
return False
if not ipaddress.ip_address(decoded_ip):
return False
if len(decoded_ip) > 15:
return False
if not re.match(r'^[A-Za-z0-9+/=]+$', ip_base64):
return False
except Exception as e:
return False
这里正则写的很死，想绕过基本上是不可能的
注意到前端
<script>
document
.getElementById("pingForm")
.addEventListener("submit", function (event) {
event.preventDefault();
const ip = document.getElementById("ip").value;
const ip_base64 = btoa(ip);
const resultElement = document.getElementById("result");
const submitButton = document.getElementById("submitBtn");
resultElement.textContent = "Pinging...";
resultElement.style.color = "var(--text-secondary-color)";
submitButton.disabled = true;
submitButton.textContent = "Pinging...";
fetch("/ping", {
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({ ip_base64: ip_base64 }),
})
.then((response) =>
response
.json()
.then((data) => ({ status: response.status, body: data }))
)
.then(({ status, body }) => {
if (status === 200 && body.success) {
resultElement.style.color = "var(--success-color)";
resultElement.textContent = body.output;
} else {
resultElement.style.color = "var(--error-color)";
resultElement.textContent =
"Ping failed: " +
(body.error || "Invalid IP");
}
})
.catch((error) => {
console.error("Error:", error);
resultElement.style.color = "var(--error-color)";
resultElement.textContent =
"An error occurred";
})
.finally(() => {
submitButton.disabled = false;
submitButton.textContent = "Ping";
});
});
</script>
这里是直接把我们输入的ip编码为base64然后发送过去
那就很可疑了，好端端的，偏偏要前端编码一次，后端解码一次，这就是猫腻所在，而且也因为前端的编码，把我们对后端的输入内容限制死了，只限制在点分十进制的ipv4地址上
所以这里用hackbar绕过前端
这样子我就可以直接输入base64编码的内容，来进行操作
接下来就是去查查会存在什么问题了
import base64
import subprocess
import re
import ipaddress
import flask
导入base64库，这一整个题都在围绕base64展开，估计是有什么不为人知的小trick
跟进去看源码
这里用的是base64.b64decode
def b64decode(s, altchars=None, validate=False):
"""Decode the Base64 encoded bytes-like object or ASCII string s.
Optional altchars must be a bytes-like object or ASCII string of length 2
which specifies the alternative alphabet used instead of the '+' and '/'
characters.
The result is returned as a bytes object. A binascii.Error is raised if
s is incorrectly padded.
If validate is False (the default), characters that are neither in the
normal base-64 alphabet nor the alternative alphabet are discarded prior
to the padding check. If validate is True, these non-alphabet characters
in the input result in a binascii.Error.
For more information about the strict base64 check, see:
https://docs.python.org/3.11/library/binascii.html#binascii.a2b_base64
"""
s = _bytes_from_decode_data(s)
if altchars is not None:
altchars = _bytes_from_decode_data(altchars)
assert len(altchars) == 2, repr(altchars)
s = s.translate(bytes.maketrans(altchars, b'+/'))
return binascii.a2b_base64(s, strict_mode=validate)
哈哈哈，还给了文档，去看看
没什么头绪
回过来重新阅读源码，发现猫腻
做正则过滤的时候，用的是python的base64过滤
而在ping的时候，是用shell的base64来进行解码，这其中肯定存在问题
command = f"""echo "ping -c 1 $(echo '{ip_base64}' | base64 -d)" | sh"""
说明有某种方法可以绕过python的base64解码后的检查，同时也可以被shell正确解码，实现rce
重新读一下源码的注释
The result is returned as a bytes object. A binascii.Error is raised if
s is incorrectly padded.
这里说s要是被错误填充的话，就会抛出一个binascii错误，去看看错误
挖草，一下子全部都清晰起来了
这里说的，如果 strict_mode 为真，则仅转换有效的 base64 数据。无效的 base64 数据将引发 binascii.Error
而我们看b64decode的源码
return binascii.a2b_base64(s, strict_mode=validate)
密码的
尴尬了兄弟，这里strict_mode一直都是false
看wp
这里有一个小trick(目前还不知道该如何去解释)
python的base64.b64decode不会对=之后的内容继续解码，从而通过只能是ip格式的校验，
而shell的base64 -d 会将编码从中间拆开分别解码再拼接，从而可以命令拼接执行，因此我们将两部分拆开即可0.0.0.0;cat /flag
MC4wLjAuMA==O2NhdCAvZmxhZw==
成功拿到flag
关于我发现的那些东西，文档里讲的不太清楚，也有可能扯到new trick，我和拉蒙特徐讨论了之后，给python发了issue
最后issue被接受了，他们更改了那段模糊不清的解释
Peek a Fork
套接字吗，有点东西啊
import socket
import os
import hashlib
import fcntl
import re
import mmap
with open('flag.txt', 'rb') as f:
flag = f.read()
mm = mmap.mmap(-1, len(flag))
mm.write(flag)
os.remove('flag.txt')
FORBIDDEN = [b'flag', b'proc', b'<', b'>', b'^', b"'", b'"', b'..', b'./']
PAGE = """<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Secure Gateway</title>
<style>
body { font-family: 'Courier New', monospace; background-color: #0c0c0c; color: #00ff00; text-align: center; margin-top: 10%; }
.container { border: 1px solid #00ff00; padding: 2rem; display: inline-block; }
h1 { font-size: 2.5rem; text-shadow: 0 0 5px #00ff00; }
p { font-size: 1.2rem; }
.status { color: #ffff00; }
</style>
</head>
<body>
<div class="container">
<h1>Firewall</h1>
<p class="status">STATUS: All systems operational.</p>
<p>Your connection has been inspected.</p>
</div>
</body>
</html>"""
def handle_connection(conn, addr, log, factor=1):
try:
conn.settimeout(10.0)
if log:
with open('log.txt', 'a') as f:
fcntl.flock(f, fcntl.LOCK_EX)
log_bytes = f"{addr[0]}:{str(addr[1])}:{str(conn)}".encode()
for _ in range(factor):
log_bytes = hashlib.sha3_256(log_bytes).digest()
log_entry = log_bytes.hex() + "\n"
f.write(log_entry)
request_data = conn.recv(256)
if not request_data.startswith(b"GET /"):
response = b"HTTP/1.1 400 Bad Request\r\n\r\nInvalid Request"
conn.sendall(response)
return
try:
path = request_data.split(b' ')[1]
pattern = rb'\?offset=(\d+)&length=(\d+)'
offset = 0
length = -1
match = re.search(pattern, path)
if match:
offset = int(match.group(1).decode())
length = int(match.group(2).decode())
clean_path = re.sub(pattern, b'', path)
filename = clean_path.strip(b'/').decode()
else:
filename = path.strip(b'/').decode()
except Exception:
response = b"HTTP/1.1 400 Bad Request\r\n\r\nInvalid Request"
conn.sendall(response)
return
if not filename:
response_body = PAGE
response_status = "200 OK"
else:
try:
with open(os.path.normpath(filename), 'rb') as f:
if offset > 0:
f.seek(offset)
data_bytes = f.read(length)
response_body = data_bytes.decode('utf-8', 'ignore')
response_status = "200 OK"
except Exception as e:
response_body = f"Invalid path"
response_status = "500 Internal Server Error"
response = f"HTTP/1.1 {response_status}\r\nContent-Length: {len(response_body)}\r\n\r\n{response_body}"
conn.sendall(response.encode())
except Exception:
pass
finally:
conn.close()
os._exit(0)
def main():
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(('0.0.0.0', 1337))
server.listen(50)
print(f"Server listening on port 1337...")
while True:
try:
pid, status = os.waitpid(-1, os.WNOHANG)
except ChildProcessError:
pass
conn, addr = server.accept()
initial_data = conn.recv(256, socket.MSG_PEEK)
if any(term in initial_data.lower() for term in FORBIDDEN):
conn.sendall(b"HTTP/1.1 403 Forbidden\r\n\r\nSuspicious request pattern detected.")
conn.close()
continue
if initial_data.startswith(b'GET /?log=1'):
try:
factor = 1
pattern = rb"&factor=(\d+)"
match = re.search(pattern, initial_data)
if match:
factor = int(match.group(1).decode())
pid = os.fork()
if pid == 0:
server.close()
handle_connection(conn, addr, True, factor)
except Exception as e:
print("[ERROR]: ", e)
finally:
conn.close()
continue
else:
pid = os.fork()
if pid == 0:
server.close()
handle_connection(conn, addr, False)
conn.close()
if __name__ == '__main__':
main()
知识盲区了属于是
稍微了解一下mmap
文档
从文档中，可以看出来这里直接把flag写入了内存，然后删除了flag.txt
AI给出的做法是
用“分片发送”的 trick 请求 proc/self/maps（绕过 256 字节 peek 的黑名单），拿到 maps；
解析 maps，找候选的可读 rw 区间，然后分别请求读取这些地址对应的 /proc/self/mem