业务逻辑-高并发漏洞

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

我是使用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，演示视频

转载请注明来自网盾网络安全培训，本文标题：《业务逻辑-高并发漏洞》

标签：漏洞分析业务逻辑漏洞高并发