AidLux智慧交通安全实战

1

采用Aidlux实现对抗检测视屏演示_哔哩哔哩_bilibili

对抗防御思路

对抗生成样本

目前主流对抗攻击算法的总体分支与逻辑：

7a3bc821df0d0f6caa52606bb660727

白盒攻击

该场景需要攻击者知道算法模型参数和训练数据等信息被攻击者所掌握，但真实业务中，大多情况下，模型参数和训练数据是不会暴露的。

主要手段如下:

  • 基于梯度
    • FGSM c963f6d7c268b3e7d72155d09e2421c
    • PGM是FGSM算法的一个优化版本。其在FGSM基础上引入了迭代攻击的思想，将多次攻击后的数据作为最终的对抗样本。
  • 基于优化
    • 其中Deepfool攻击算法是基于超平面分类的攻击方法，其通过迭代计算的方法生成最小规范对抗扰动，将位于分类边界内的图像逐步推到边界外，直到出现错误预测。 9c375f72c9a3c38c5d5bacab08643d8
    • 基于GAN，AdvGAN将GAN思想引对抗攻击，通过生成器生成对抗扰动，添加到干净样本中。而判别器主要负责判别输入的是对抗样本还是干净样本。整个对抗训练过程转化成GAN经典博弈过程。 cbf73a6c15c83ae8ede89dea53569de

黑盒攻击

  • 基于迁移的攻击 基于迁移的攻击逻辑由白盒攻击延伸而来，一般会有一个白盒模型作为****替身模型 （surrogate）进行攻击，而生成的对抗样本一般也会对其他模型有一定的 迁移攻击性 。 主要有下面三种：
    • 基于梯度的攻击
    • 基于优化的攻击
    • 基于GAN的攻击
  • 基于查询的攻击（×） 0b75cd2580a091e372974975833e899

防御对抗

目前主流对抗防御的总体分支与逻辑：

2fe0c7b43564a637f639f9925b29757

  • 对抗训练是指在训练过程中加入对抗样本，通过不断的学习对抗样本的特征，从而提升模型的鲁棒性。
  • 监测识别对抗样本顾名思义，在项目关键节点处，设置一些能够识别对抗样本的特种模型，从而提前预警对抗攻击风险。
  • 模型鲁棒结构设计是指在模型中设计特定的滤波结构能够一定程度上增强模型鲁棒性，抵御对抗噪声。
  • 对抗扰动结构破坏主要在数据流处理的时候使用，通过一些滤波算法，噪声结构破坏算法，噪声覆盖算法等策略，减弱对抗噪声的影响，使其不能对模型造成攻击。
  • 梯度掩膜则是在白盒对抗防御中非常高效的一种算法，因为其能掩盖真实梯度，从而能够使得白盒攻击算法失效。
    • GCM模块通过改造输入，使模型在前向测试时并不影响原来的结果，但是在进行梯度反向传播时，能隐藏原来的梯度，使得白盒攻击无所适从 e398c333d4769c138a1649e4407407a

智慧交通安全特点

  • 在智慧交通场景中，常用的对抗防御方法有****与对抗攻击结合紧密的AI业务的鲁棒性与安全性检查工具 ； 研究对抗攻击，来增强对实际场景中受到的攻击情形的判断力 ； 使用对抗攻击对敏感信息进行隐藏 ； 使用对抗样本在模型训练过程中进行对抗训练 。
  • 由于****智慧交通场景的算法解决方案对耗时和内存占用有较高的要求，所以防御预处理与防御后处理作为常规防御的首要选择。
  • 同时，针对可能存在的安全风险，在开发阶段，设计鲁棒性的模型结构，提升算法解决方案的整体安全性。
  • 或者训练轻量级的对抗攻击检测模型，作为算法解决方案的安全模块，在受到攻击风险时，启动防御算法功能。

2af3f5717ca3b379820f5e7c2241c30

  • 具体来说，我们需要生成一些对抗样本，并将其和干净样本一同作为训练集，训练一个能够捕捉对抗扰动的二分类模型，并将其作为实际AI项目中的一个前置防御模块。

案例（作业）

对抗生成样本

替身模型生成对抗样本

使用Resnet18作为替身模型进行对抗生成图片

  • 攻击原图： /home/lesson5/Lesson5_code/adv_code/orig_images/vid_5_26780.jpg_0.jpg
  • 使用PGD进行对抗，并保存图片 image-20221207103628696
  • 可考虑其他对抗算法，也可对对抗参数优化参数
    • L1PGD
    • L2PGD
    • FGSM
    • MomentumIterative
    • 等

验证有效性

  • 在对抗监测分类模型中类别输出>0.5 image-20221207104718397

防御

对抗检测模型+喵提醒

若出现对抗样本，则不进行之后的级联任务，以免带来风险

image-20221207105154554

梯度掩膜

GCM，对于基于梯度的对抗攻击有效

``` ### 常规模型加载 class Model(nn.Module): def __init__(self, l=290): super(Model, self).__init__() ​ self.l = l # GCM self.gcm = GradientConcealment() #model = resnet18(pretrained=True) model = mobilenet_v2(pretrained=True) ​ # pth_path = "/Users/rocky/Desktop/训练营/model/mobilenet_v2-b0353104.pth" # print(f'Loading pth from {pth_path}') # state_dict = torch.load(pth_path, map_location='cpu') # is_strict = False # if 'model' in state_dict.keys(): # model.load_state_dict(state_dict['model'], strict=is_strict) # else: # model.load_state_dict(state_dict, strict=is_strict) ​ normalize = NormalizeByChannelMeanStd( mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]) self.model = nn.Sequential(normalize, model) ​ def load_params(self): pass ​ def forward(self, x): #x = self.gcm(x) #x = ResizedPaddingLayer(self.l)(x) out = self.model(x) return out ​ 
<h4>可考虑其他手段：</h4>
<ul>
<li><strong>对训练样本增加白噪点，提升模型鲁棒性</strong></li>
<li><strong>将对抗样本和干净样本放在一起训练，提升模型的抗对抗性</strong></li>
</ul>
<h3>串联目标检测+对抗防御+分类流程</h3>
<p>主程序代码：</p>
<p>主要改动俩点：</p>
<ol>
<li>遍历整个裁剪的图片，若监测&gt;0.7,喵提醒，否则进行分类</li>
<li>将监测一场的threshold放大到0.7</li>
</ol>

export OMP_NUM_THREADS=1

aidlux相关

from cvs import *
import aidlite_gpu
from yolov5_code.aidlux.utils import detect_postprocess, preprocess_img, draw_detect_res, extract_detect_res

import time
import cv2

print(“流程开始”)
print(“流程1：检测car, 并提取框”)

AidLite初始化：调用AidLite进行AI模型的加载与推理，需导入aidlite

aidlite = aidlite_gpu.aidlite()

Aidlite模型路径

model_path = ‘/home/lesson5/Lesson5_code/yolov5_code/models/yolov5_car_best-fp16.tflite’

定义输入输出shape

in_shape = [1 * 640 * 640 * 3 * 4]
out_shape = [1 * 25200 * 6 * 4]

加载Aidlite检测模型：支持tflite, tnn, mnn, ms, nb格式的模型加载

aidlite.ANNModel(model_path, in_shape, out_shape, 4, 0)

读取图片进行推理

设置测试集路径

source = “/home/lesson5/Lesson5_code/yolov5_code/data/images/tests”
images_list = os.listdir(source)
print(“提取开始”)
print(images_list)
frame_id = 0

读取数据集

for image_name in images_list:
frame_id += 1
print(“frame_id:”, frame_id)
image_path = os.path.join(source, image_name)
frame = cvs.imread(image_path)

# 预处理
img = preprocess_img(frame, target_shape=(640, 640), div_num=255, means=None, stds=None)
# 数据转换：因为setTensor_Fp32()需要的是float32类型的数据，所以送入的input的数据需为float32,大多数的开发者都会忘记将图像的数据类型转换为float32
aidlite.setInput_Float32(img, 640, 640)
# 模型推理API
aidlite.invoke()
# 读取返回的结果
pred = aidlite.getOutput_Float32(0)
# 数据维度转换
pred = pred.reshape(1, 25200, 6)[0]
# 模型推理后处理
pred = detect_postprocess(pred, frame.shape, [640, 640, 3], conf_thres=0.25, iou_thres=0.45)
# 绘制推理结果
res_img = draw_detect_res(frame, pred)
# cvs.imshow(res_img)

# 测试结果展示停顿
#time.sleep(5)

# 图片裁剪，提取车辆目标区域
extract_detect_res(frame, pred, image_name)

# cvs.imshow(cut_img)
# cap.release()
# cv2.destroyAllWindows()

print(“提取结束”)
print(“流程2：在提取的图片中增加对抗样本”)

已提前增加

print(“已增加对抗样本，样本路径为”, “/home/lesson5/Lesson5_code/extract/adv_image_1.png”)
print(“已增加对抗样本，样本路径为”, “/home/lesson5/Lesson5_code/extract/adv_image_2.png”)

print(“流程3：进行分类，对于对抗样本喵提醒，对于正常样本执行正常分类任务”)

import os
import torch
import requests
import time
import torch.nn as nn
from torchvision.models import mobilenet_v2,resnet18
from adv_code.advertorch.utils import predict_from_logits
from adv_code.advertorch.utils import NormalizeByChannelMeanStd
from adv_code.robust_layer import GradientConcealment, ResizedPaddingLayer
from adv_code.timm.models import create_model

from adv_code.advertorch.attacks import LinfPGDAttack
from adv_code.advertorch_examples.utils import ImageNetClassNameLookup
from adv_code.advertorch_examples.utils import bhwc2bchw
from adv_code.advertorch_examples.utils import bchw2bhwc

device = “cuda” if torch.cuda.is_available() else “cpu”

读取图片

def get_image(image_name):
img_path = os.path.join(“/home/lesson5/Lesson5_code/extract/”, image_name)
img_url = “https://farm1.static.flickr.com/230/524562325_fb0a11d1e1.jpg”

def _load_image():
    from skimage.io import imread
    return imread(img_path) / 255.

if os.path.exists(img_path):
    return _load_image()
else:
    import urllib
    urllib.request.urlretrieve(img_url, img_path)
    return _load_image()

def tensor2npimg(tensor):
return bchw2bhwc(tensor[0].cpu().numpy())

normalize = NormalizeByChannelMeanStd(
mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])

常规模型加载

class Model(nn.Module):
def init(self, l=290):
super(Model, self).init()

    self.l = l
    self.gcm = GradientConcealment()
    #model = resnet18(pretrained=True)
    model = mobilenet_v2(pretrained=True)

    # pth_path = "/Users/rocky/Desktop/训练营/model/mobilenet_v2-b0353104.pth"
    # print(f'Loading pth from {pth_path}')
    # state_dict = torch.load(pth_path, map_location='cpu')
    # is_strict = False
    # if 'model' in state_dict.keys():
    #    model.load_state_dict(state_dict['model'], strict=is_strict)
    # else:
    #    model.load_state_dict(state_dict, strict=is_strict)

    normalize = NormalizeByChannelMeanStd(
        mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
    self.model = nn.Sequential(normalize, model)

def load_params(self):
    pass

def forward(self, x):
    #x = self.gcm(x)
    #x = ResizedPaddingLayer(self.l)(x)
    out = self.model(x)
    return out

对抗攻击监测模型

class Detect_Model(nn.Module):
def init(self, num_classes=2):
super(Detect_Model, self).init()
self.num_classes = num_classes
#model = create_model(‘mobilenetv3_large_075’, pretrained=False, num_classes=num_classes)
model = create_model(‘resnet50’, pretrained=False, num_classes=num_classes)

    # self.multi_PreProcess = multi_PreProcess()
    pth_path = os.path.join("/home/lesson5/Lesson5_code/model", 'track2_resnet50_ANT_best_albation1_64_checkpoint.pth')
    #pth_path = os.path.join("/Users/rocky/Desktop/训练营/Lesson5_code/model/", "track2_tf_mobilenetv3_large_075_64_checkpoint.pth")
    state_dict = torch.load(pth_path, map_location='cpu')
    is_strict = False
    if 'model' in state_dict.keys():
        model.load_state_dict(state_dict['model'], strict=is_strict)
    else:
        model.load_state_dict(state_dict, strict=is_strict)
    normalize = NormalizeByChannelMeanStd(
        mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
    # self.model = nn.Sequential(normalize, self.multi_PreProcess, model)
    self.model = nn.Sequential(normalize, model)

def load_params(self):
    pass

def forward(self, x):
    # x = x[:,:,32:193,32:193]
    # x = F.interpolate(x, size=(224,224), mode="bilinear", align_corners=True)
    # x = self.multi_PreProcess.forward(x)
    out = self.model(x)
    if self.num_classes == 2:
        out = out.softmax(1)
        #return out[:,1:]
        return out[:,1:]

source = “/home/lesson5/Lesson5_code/extract/”
images_list = os.listdir(source)
for image_name in images_list:
print(“开始分类”, image_name)
np_img = get_image(image_name)
img = torch.tensor(bhwc2bchw(np_img))[None, :, :, :].float().to(device)
imagenet_label2classname = ImageNetClassNameLookup()

model = Model().eval().to(device)

detect_model = Detect_Model().eval().to(device)


### 对抗攻击监测
detect_pred = detect_model(img)
print("detect_pred: ", detect_pred)

if detect_pred > 0.8:
    id = 'tO4SO8K'
    # 填写喵提醒中，发送的消息，这里放上前面提到的图片外链
    text = "【注意啦！】 "+ image_name + " 出现对抗攻击风险！！"
    ts = str(time.time())  # 时间戳
    type = 'json'  # 返回内容格式
    request_url = "http://miaotixing.com/trigger?"

    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.67 Safari/537.36 Edg/87.0.664.47'}

    result = requests.post(request_url + "id=" + id + "&text=" + text + "&ts=" + ts + "&type=" + type,
                        headers=headers)
    print("警告，为对抗样本！！！")
else:
    pred = imagenet_label2classname(predict_from_logits(model(img)))
    print(pred)
print("分类结束")

print(“流程结束”)

<p><strong>效果</strong></p>
<p><img src="https://aidlux.oss-cn-beijing.aliyuncs.com/imgs/image75363504192274019031670394920656990931.png" alt="" /></p>
2

大佬呀