基于Aidlux智慧交通——AI安全领域

1 训练营课程

（1）本节课主要学习AI安全及车辆检测实战训练营的课程：

11月AI实战训练营正式开启，智慧交通专场，一起来参加吧！

该训练营为11月AI实战训练营，本节课基于Aidlux开发，AidLux是基于ARM架构的跨生态（Android/鸿蒙+Linux）一站式AIoT应用快速开发和部署平台APP。通过共享 Linux 内核实现Android 和 Linux 生态融合，为单一设备同时提供Android和Linux运行环境；有兴趣的同学可以关注公众号：AidLux

本次课程目录如下：

2 学习目的

本次训练营带大家了解YOLO训练一个车辆检测模型，同时将模型部署在aidlux上，当然还离不开本次训练营的主题，本次训练营的主题是AI安全，此次训练营带领大家讨论了在AI发展途中，常见的常用AI对抗攻击算法讲解，现有AI的漏洞已经如何去防御AI攻击，我们将会从攻击者的角度和防御者的角度来亲自体验AI和算法的魅力。

3 学习内容

AI安全存在的问题

对抗攻击（adversarial attack）是AI安全方向的重要分支，其核心逻辑是在数据中增加一些微小扰 动，在人类视觉系统无法察觉的情况下，使得算法模型对这些数据产生误判。其中被增加扰动的数据 也称为对抗样本。

然后就是AI攻击中经典的熊猫图

常用AI对抗攻击算法划分

白盒攻击

其中，当算法模型参数和训练数据等信息被攻击者所掌握，并且攻击者在此基础上进行的针对性攻击 称为白盒攻击。白盒攻击主要分为基于梯度的攻击，基于优化的攻击以及基于GAN的攻击。

黑盒攻击

而在对算 法模型的结构和参数一无所知，甚至相关训练数据也一无所知的情况下，进行攻击的过程称为黑盒攻 击。

黑盒攻击主要分为基于迁移的攻击和基于查询的攻击两大类。 基于迁移的攻击逻辑由白盒攻击延伸而来，一般会有一个白盒模型作为替身模型（surrogate）进行攻 击，而生成的对抗样本一般也会对其他模型有一定的迁移攻击性。 基于查询的攻击其主要是通过查询黑盒模型的输出信息，对黑盒模型进行直接的攻击，但完成整个攻 击流程往往需要大量的查询，容易被模型所有者检测到异常。 再者，基于此我们再介绍两个有趣的攻击策略： 有目标攻击（targeted attack）：通过攻击使得模型将特定数据错误输出到指定的label上。 无目标攻击：通过攻击使得模型将特定数据错误输出到除真实label外的其他任何label上。 类模型，针对有目标攻击，我们通过生成对抗样本，想让分类模型将猫的类别错判成狗。而无目标攻 击，并不限定特定类别，只要模型发生错判即可。

对抗防御算法

在智慧交通场景中，常用的对抗防御方法有与对抗攻击结合紧密的AI业务的鲁棒性与安全性检查工 具；研究对抗攻击，来增强对实际场景中受到的攻击情形的判断力；使用对抗攻击对敏感信息进行隐 藏；使用对抗样本在模型训练过程中进行对抗训练。

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

本次训练营使用到的对抗防御方法主要是基于梯度掩膜的GCM模块。

4 大作业

本次训练营作业：针对前面讲到的对抗样本监测模型与交通场景中存在的黑盒攻击风险，我们在此基础上优化对抗样本，并将车辆检测结果+对抗样本生成+对抗防御策略串联形成完善的AI安全功能。

程序流程：

第一步:车辆检测，获取到汽车的位置坐标

第二步:裁剪出车辆

第三步:对裁剪出的图片进行对抗攻击

第四步:检验对抗防御策略是否有效，喵提醒

代码部分

## 车辆检测代码
# aidlux相关
from cvs import *
import aidlite_gpu
from utils2 import detect_postprocess, preprocess_img, draw_detect_res, extract_detect_res

import time
import cv2

# AidLite初始化：调用AidLite进行AI模型的加载与推理，需导入aidlite
aidlite = aidlite_gpu.aidlite()
# Aidlite模型路径
model_path = '/home/secound_lesson/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)
image_path  = "/home/secound_lesson/Lesson5_code/yolov5_code/data/images/tests/vid_5_420.jpg"
frame = cvs.imread(image_path)
# print(frame)
# 预处理
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)
all_boxes = pred[0]
frame = frame.astype(np.uint8)
if len(all_boxes) > 0:
    for box in all_boxes:
        x, y, w, h = [int(t) for t in box[:4]]
        cut_img = frame[y:(y+h), x:(x + w)]

# cvs.imshow(cut_img)
# time.sleep(5)
# cap.release()
# cv2.destroyAllWindows()

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

from advertorch.attacks import LinfPGDAttack
from advertorch_examples.utils import ImageNetClassNameLookup
from advertorch_examples.utils import bhwc2bchw
from advertorch_examples.utils import bchw2bhwc
import time
import requests
device = "cuda" if torch.cuda.is_available() else "cpu"


### 读取图片
def get_image():
    img_path = os.path.join("./orig_images", "vid_5_31040.jpg_3.jpg")
    # img_url = "https://farm1.static.flickr.com/230/524562325_fb0a11d1e1.jpg"
    img_url  = "https://img.zcool.cn/community/[email protected]_1l_2o_100sh.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())

### 展示攻击结果
def show_images(model, img, advimg, enhance=127):
    np_advimg = tensor2npimg(advimg)
    np_perturb = tensor2npimg(advimg - img)

    pred = imagenet_label2classname(predict_from_logits(model(img)))
    advpred = imagenet_label2classname(predict_from_logits(model(advimg)))

    import matplotlib.pyplot as plt

    plt.figure(figsize=(10, 5))
    plt.subplot(1, 3, 1)
    plt.imshow(np_img)

    plt.axis("off")
    plt.title("original image\n prediction: {}".format(pred))
    plt.subplot(1, 3, 2)
    plt.imshow(np_perturb * enhance + 0.5)

    plt.axis("off")
    plt.title("the perturbation,\n enhanced {} times".format(enhance))
    plt.subplot(1, 3, 3)
    plt.imshow(np_advimg)
    plt.axis("off")
    plt.title("perturbed image\n prediction: {}".format(advpred))
    plt.show()


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


### 常规模型加载
model = mobilenet_v2(pretrained=True)
model.eval()
model = nn.Sequential(normalize, model)
model = model.to(device)


### 替身模型加载
model_su = resnet18(pretrained=True)
model_su.eval()
model_su = nn.Sequential(normalize, model_su)
model_su = model_su.to(device)


### 数据预处理
# np_img = get_image()
np_img = cut_img
img = torch.tensor(bhwc2bchw(np_img))[None, :, :, :].float().to(device)
imagenet_label2classname = ImageNetClassNameLookup()


### 测试模型输出结果
pred = imagenet_label2classname(predict_from_logits(model(img)))
print("test output:", pred)


### 输出原label
pred_label = predict_from_logits(model_su(img))


### 对抗攻击：PGD攻击算法
adversary = LinfPGDAttack(
   model_su, eps=8/255, eps_iter=2/255, nb_iter=80,
   rand_init=True, targeted=False)


### 完成攻击，输出对抗样本
advimg = adversary.perturb(img, pred_label)
print(advimg.shape)

# 将攻击样本送入
img = advimg
imagenet_label2classname = ImageNetClassNameLookup()


### 常规模型加载
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(r"/home/secound_lesson/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:]


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

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


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

if detect_pred > 0.5:
    id = 'xxxx' 这里填自己的喵码

    # 填写喵提醒中，发送的消息，这里放上前面提到的图片外链
    text = "出现对抗攻击风险！！"
    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)
else:
    pred = imagenet_label2classname(predict_from_logits(model(img)))
    print("没被攻击",pred)
           

本次训练营第一次让我学习到了什么是对抗攻击，AI方面也存在着一些巨大的问题，这是以前没有想过的