系统管理员完成检查工作后，在流程内上传纸质签字单。

目前主要接入的以业务系统为主，后续会考虑接入其它类型的系统账号权限数据，尤其是一些高风险的系统，比如GitLab、邮件、堡垒机或其它集权系统（这些系统有些可能并没有数据库或未对用户开放数据库访问，对接到计算平台可能会麻烦一些）。

目前账号检查需求基本收尾，准备在平台上继续实现统计报表、趋势视图、问题跟踪等管理功能的需求。

另外，老A说考虑未来会接入一部分日志数据，这个需求我们还在讨论，个人感觉除了登录日志可能会利于发现“长期不使用”的账号，其它日志接到这个系统里有悖于这个系统设计的初衷。

准备工作：

因为是快速测试验证，所以直接在本地进行，只管老A要了一个系统的用户、角色表的数据。

工具：python、numpy、pandas、sklearn、matplotlib

数据：PXX.xlsx（关键信息已脱敏）

目标：通过算法检出权限异常账号

# 1.读取数据，预览数据# 分别读取数据表，检查是否存在空值import pandas as pdimport numpy as np
print("获取用户部门表：df_user_dept")df_user_dept = pd.read_excel("./data/PXX.xlsx", sheet_name="user_dept").sort_values(by="user_id", ignore_index=True)validate(df_user_dept)print(df_user_dept)

print("获取用户角色表：df_user_role")df_user_role = pd.read_excel("./data/PXX.xlsx", sheet_name="user_role").sort_values(by="user_id", ignore_index=True)validate(df_user_role)print(df_user_role)

print("获取角色名称表：df_role")df_role = pd.read_excel("./data/PXX.xlsx", sheet_name="role").sort_values(by="role_id", ignore_index=True)validate(df_role)print(df_role)

这张表实际计算过程中并不需要，只是为了了解一下表结构

def validate(df):    # 检查空值    if df.isna().values.any():        print("发现空值：", df[df.isna().values == True])    # 检查重复    if df.duplicated().values.any():        print("发现重复值：", df[df.duplicated().values == True])

需要检查一下缺失值、重复数据，实际上我这里导出的数据并不涉及。

# 2.组装数据，形成用户权限矩阵# 建立role_id列user_id_uni_size = df_user_dept["user_id"].unique().sizerole_id_uni_size = df_role["role_id"].unique().sizeprint("建立以role_id作为列名的数据， 行数量：%s 列数量：%s" % (user_id_uni_size, role_id_uni_size))pd_role_id_cols = pd.DataFrame(np.zeros((user_id_uni_size, role_id_uni_size)),                               columns=["role_id_" + str(_) for _ in df_role["role_id"].unique().tolist()])print(pd_role_id_cols)

这里需要拼接一下数据，把64种role_id类型，变成64列全为“0”的数据

# 组装数据，形成每行为一个用户的信息，以及是否有每个role_id的信息，形成用户权限矩阵print("建立用户权限矩阵")df_user_role_matrix = df_user_dept.filter(    items=["user_id", "user_name", "domain_account", "company_code", "department_name", "role_id"])df_user_role_matrix = pd.concat([df_user_role_matrix, pd_role_id_cols], axis=1)print(df_user_role_matrix.head(5).to_string())

拼接到用户表，这里只打印了5行

print("更新用户权限矩阵数据")for rowi in df_user_role_matrix.index:    user_id = df_user_role_matrix.loc[rowi, "user_id"]    for _ in df_user_role[df_user_role["user_id"] == user_id]["role_id"].values:        col_role_id = "role_id_" + str(_)        df_user_role_matrix.loc[rowi, col_role_id] = 1print("更新完成")print(df_user_role_matrix.head(5).to_string())

根据user_role表的信息，把用户存在的role所在的列更新为“1”，一个用户可能存在多个role

得到[309 rows x 64 columns]的权限矩阵

# 3.进行特征编码、标准化print("进行特征编码")# 使用独热编码from sklearn.preprocessing import OneHotEncoder, StandardScaler
def one_hot_encode(df, cols):    enc = OneHotEncoder(handle_unknown='ignore', sparse_output=False)    enc.fit(df[cols].values)    matrix = enc.transform(df[cols].values)    col_names = []    print("%s %s 类别为：" % (cols, enc.categories_))    for col in cols:        for val in df[col].unique():            col_names.append("%s_%s" % (col, val))return pd.DataFrame(data=matrix, columns=col_names, dtype=int)
pd_encode = one_hot_encode(df_user_role_matrix, ["company_code", "department_name"])print(pd_encode.head(5).to_string())

可以看到，company_code、department_name这两个字段代表公司名称、部门名称，内容都是类别型数据，必须转换成数值型才能进行计算。

所以这里使用了热独编码。比如company_code['TBQ', 'TCM', 'TFM', 'TGI', 'TWM']有5种唯一的值，独热编码会将其编码成为一个5维的稀疏矩阵，像下面这样。

从行的角度看，只要这5列中有1列为1，则另外4列都为0。

同理，department_name也会进行编码，只不过由于部门这个属性的类别比较多，所以编码后得到的列也会更多。

df_user_role_matrix = pd.concat([df_user_role_matrix, pd_encode], axis=1)print(df_user_role_matrix.head(5).to_string())print("最终得到的数据：%s行 * %s 列" % df_user_role_matrix.shape)

将编码后的稀疏矩阵拼接到原数据后，最终得到的数据：309行 * 175 列

# 标准化# 由于我这里均为0、1的数据，各列数据没有比例差异，不再需要进行特征缩放。

x_train = df_user_role_matrix.drop(["user_id", "user_name", "domain_account", "company_code", "department_name"], axis=1)print(x_train)print("x_train：%s行 * %s 列" % x_train.shape)

去掉user_id、user_name等列，最终将[309 rows x 170 columns]的x_train进行训练

from sklearn.neighbors import LocalOutlierFactor# 异常比例设置为auto，也可根据预估期望情况设置，比如0.1即10%lof = LocalOutlierFactor(n_neighbors=50, contamination="auto")y_predict = lof.fit_predict(x_train)y_scores = lof.negative_outlier_factor_print("y_predict：%s" % y_predict)print("y_scores：%s" % y_scores)

结果：y_predict值1为正常，-1为异常；y_scores值，与1的比例越大，代表越异常

print("将计算结果拼接到权限矩阵")df_user_role_matrix_target = pd.concat([df_user_role_matrix, pd.DataFrame({"target": y_predict, "score": y_scores})],axis=1)print(df_user_role_matrix_target)

# 根据计算结果匹配出异常用户的信息outlier_num = df_user_role_matrix_target[df_user_role_matrix_target["target"] == -1].shape[0]print("总数据：%s个，异常数据%s个" % (df_user_role_matrix_target.shape[0], outlier_num))print("异常账号为：%s" % df_user_role_matrix_target[df_user_role_matrix_target["target"] == -1]["user_name"].values)print("异常值为：%s" % df_user_role_matrix_target[df_user_role_matrix_target["target"] == -1]["score"].values)

最终检测出16个异常账号，打印用户账号、异常值

实际到上一步已经完成算法的使用了，但是可能比较抽象，这里用可视化进行一下展示，会更直观一点。

# 查看一下相关性print("皮尔逊相关系数多重共线性：")print(df_user_role_matrix_target.corr()["target"].sort_values(ascending=False).to_string())

可以看到不同特征与最终结果的相关性（貌似每一个都不是特别相关的...比例都不大）

# 将多维数据降至2维用于后续的可视化展示from sklearn.manifold import TSNE
tsne = TSNE(n_components=2, method="exact")x_train_pc = tsne.fit_transform(x_train)x_train_pc = StandardScaler().fit_transform(x_train_pc)

降至二维后的数据

# 使用matplotlib查看一下可视化效果# 绘制散点图import matplotlib.pyplot as plt
fig = plt.figure()ax = fig.add_subplot()ax.set_title("异常数据可视化")ax.set_xlabel("异常数量: %s 个" % outlier_num)ax.scatter(x_train_pc[:, 0], x_train_pc[:, 1], c=y_predict, s=3)plt.show()

局部放大

异常数据可视化效果，可以理解那些离群的点就是异常。

注：更多详情可点击“原文查看”！

查询和订阅最新安全事件，请关注”安全小飞侠“吧！