业务逻辑-高并发漏洞 – 作者:陌度

首先需要一个测试环境来进行跑高并发的问题。

我是使用python3+django 2.0+mysql去写这个测试环境

一开始，我是打算直接简单粗暴，直接往数据库生成数据

def ceshi(request):
    name = request.GET.get('name')
    ceshi.objects.create(name=name)
return HttpResponse("写入成功")

但是使用bp跑了很久才发现，没有少增多增数据。

于是就直接写了个复杂一点，按照商城的功能去写

views.py我写了三个功能，一个是购买的功能，一个返回订单的ID，一个是删除订单。

#models.py

from django.db import models

# Create your models here.
class order(models.Model):
    name = models.CharField(max_length=100)
    price = models.IntegerField()

class userinfo(models.Model):
    money = models.IntegerField()

class dingdan(models.Model):
    buyer = models.ForeignKey(userinfo,on_delete=models.CASCADE)
    lists = models.CharField(max_length=100)

#views.py

from django.shortcuts import render,HttpResponse
from django.http import JsonResponse
from .models import *
# Create your views here.
import requests
import uuid


def buy_order(request):
    name = request.GET.get('name')
    price = order.objects.get(name=name).price
    money = userinfo.objects.get(id=1).money
    userinfo.objects.filter(id=1).update(money=money-price)
    dingdan.objects.create(buyer_id=1,lists=uuid.uuid1())
    return HttpResponse("购买成功")


def show_order(request):
    t = []
    ids = dingdan.objects.all()
    for i in ids:
        t.append({'id':i.id})
    return  JsonResponse(t,safe=False)


def del_order(request):
    r = requests.get(url='http://127.0.0.1:8000/show_order/')
    for i in r.json():
        money = userinfo.objects.get(id=1).money

        userinfo.objects.filter(id=1).update(money=money+1)
        dingdan.objects.get(id=i['id']).delete()

    return  HttpResponse("删除成功")

#urls.py
from django.contrib import admin
from django.urls import path
from app.views import *

urlpatterns = [
    path('admin/', admin.site.urls),
    path('buy/',buy_order),
    path('show_order/',show_order),
    path('del_order/',del_order),
]

在这里面我没有设置用户，主要写的太复杂了，我在数据库设置一个商品的名字和价格，设置一个用户的金钱，方便用来购买和统计最后是否添加或减少金钱。

设置初始值10w

商品名称

此时，是没有订单的

我们使用bp来跑一下，这里需要开两个爆破器，一个是购买，一个是删除

emmm，跑出来了

并发测试脚本

import requests
from threading import Thread



def get1():
    r = requests.get(url="http://192.168.199.132:8000/buy/?name=test")
    print(r.text)


def get2():
    r = requests.get(url="http://192.168.199.132:8000/del_order/")
    print(r.text)

while True:

    ts1 = [Thread(target=get1) for i in range(1,10)]
    ts2 = [Thread(target=get2) for i in range(1,10)]
    for t in ts1:
        t.start()
    for t in ts1:
        t.join()
    for t in ts2:
        t.start()
    for t in ts2:
        t.join()

PS，演示视频

来源：freebuf.com 2019-03-07 19:50:43 by: 陌度