命令审计ell 轻使用 Sh松搞定 L-江苏天煌照明集团有限公司 - 智慧路灯-智能路灯-江苏天煌照明集团有限公司

首先，当谈到 Linux 的操作审计需求时，大多数我们希望的是还原线上服务器被人为（误）操作时执行的命令行，以及它关联的上下文。这个需求场景其实跟通用的业务日志采集一致，简单一点可以直接通过 history 将内容发给 syslog，复杂一点的采用 auditd 或 ebpf 在内核层面上捕获行为。

不过本文不打算对上述的方案做原理解释，仅仅站在一个运维小白的角度来完成日常 80%（80%的数据来源？我也不知道，大概是二八原则）的操作审计。既然文章标题是用 Shell 来完成，由此可见今天的主题跟 Bash 脱不了关系了。

一句话概括今天的主题：利用定制 Bash 源增加日志审计功能，并将用户操作发给 rsyslog 聚合，最后在 elasticsearch 做日志存储和查询。

Linux 部分

1. 准备一些必要的工具
  - rsyslog: 一个Linux上自带并兼容 syslog 语法的日志处理服务
  - jq: 一个在 shell 下处理 json 数据的小工具
  - logger: 一个可以往 syslog 输入日志的工具

这些小工具除 jq 外，大多操作系统发行版都自带，如果没有的话也可以直接用操作系统内置的包管理工具安装。

1. ash.audit.sh，并将其拷贝到 /etc/profile.d/ 目录下

if [ "${ SHELL##*/}" != "bash" ]; then  returnfiif [ "${ AUDIT_READY}" = "yes" ]; then    returnfideclare -rx HISTFILE="$HOME/.bash_historydeclare -rx HISTSIZE=500000declare -rx HISTFILESIZE=500000declare -rx HISTCONTROL=""declare -rx HISTIGNORE=""declare -rx HISTCMDdeclare -rx AUDIT_READY="yes"shopt -s histappendshopt -s cmdhistshopt -s histverifyif shopt -q login_shell && [ -t 0 ]; then  stty -ixonfiif groups | grep -q root; then  declare -x TMOUT=86400  # chattr +a "$HISTFILE"fideclare -a LOGIN_INFO=( $(who -mu | awk '{ print $1,$2,$6}') )declare -rx AUDIT_LOGINUSER="${ LOGIN_INFO[0]}"declare -rx AUDIT_LOGINPID="${ LOGIN_INFO[2]}"declare -rx AUDIT_USER="$USER"declare -rx AUDIT_PID="$$"declare -rx AUDIT_TTY="${ LOGIN_INFO[1]}"declare -rx AUDIT_SSH="$([ -n "$SSH_CONNECTION" ] && echo "$SSH_CONNECTION" | awk '{ print $1":"$2"->"$3":"$4}')"declare -rx AUDIT_STR="$AUDIT_LOGINUSER  $AUDIT_LOGINPID  $AUDIT_TTY  $AUDIT_SSH"declare -rx AUDIT_TAG=$(echo -n $AUDIT_STR | sha1sum |cut -c1-12)declare -x AUDIT_LASTHISTLINE=""set +o functraceshopt -s extglobfunction AUDIT_DEBUG() {   if [ -z "$AUDIT_LASTHISTLINE" ]; then    local AUDIT_CMD="$(fc -l -1 -1)"    AUDIT_LASTHISTLINE="${ AUDIT_CMD%%+([^ 0-9])*}"  else    AUDIT_LASTHISTLINE="$AUDIT_HISTLINE"  fi  local AUDIT_CMD="$(history 1)"  AUDIT_HISTLINE="${ AUDIT_CMD%%+([^ 0-9])*}"  if [ "${ AUDIT_HISTLINE:-0}" -ne "${ AUDIT_LASTHISTLINE:-0}" ] || [ "${ AUDIT_HISTLINE:-0}" -eq "1" ]; then    MESSAGE=$(jq -c -n \     --arg pwd "$PWD" \     --arg cmd "${ AUDIT_CMD##*( )?(+([0-9])?(\*)+( ))}" \     --arg user "$AUDIT_LOGINUSER" \     --arg become "$AUDIT_USER" \     --arg pid "$AUDIT_PID" \     --arg info "${ AUDIT_STR}" \     '{ cmd: $cmd, user: $user, become: $become, pid: $pid, pwd: $pwd, info: $info}')    logger -p local6.info -t "$AUDIT_TAG" "@cee: $MESSAGE"  fi}function AUDIT_EXIT() {   local AUDIT_STATUS="$?"  if [ -n "$AUDIT_TTY" ]; then    MESSAGE_CLOSED=$(jq -c -n \        --arg action "session closed" \        --arg user "$AUDIT_LOGINUSER" \        --arg become "$AUDIT_USER" \        --arg pid "$AUDIT_PID" \        --arg info "${ AUDIT_STR}" \        '{ user: $user, become: $become, pid: $pid, action: $action, info: $info}')    logger -p local6.info -t "$AUDIT_TAG" "@cee: $MESSAGE_CLOSED"  fi  exit "$AUDIT_STATUS"}declare -frx +t AUDIT_DEBUGdeclare -frx +t AUDIT_EXITif [ -n "$AUDIT_TTY" ]; then  MESSAGE_OPENED=$(jq -c -n \      --arg action "session opened" \      --arg user "$AUDIT_LOGINUSER" \      --arg become "$AUDIT_USER" \      --arg pid "$AUDIT_PID" \      --arg info "${ AUDIT_STR}" \      '{ user: $user, become: $become, pid: $pid, action: $action, info: $info}')  logger -p local6.info -t "$AUDIT_TAG" "@cee: $MESSAGE_OPENED"fideclare -rx PROMPT_COMMAND="[ -n "$AUDIT_DONE" ] && echo ''; AUDIT_DONE=; trap 'AUDIT_DEBUG && AUDIT_DONE=1; trap DEBUG' DEBUGdeclare -rx BASH_COMMANDdeclare -rx SHELLOPTtrap AUDIT_EXIT EXIT

简单说明下这个脚本，大致就是定义了 shell 的历史条目、登录超时时间、以及审计日志的格式和发送。

1. 配置rsyslog 客户端，本地创建一个 /etc/rsyslog.d/40-audit.conf 文件，用于将本地local6级别的系统日志发送远端的rsyslog服务集中处理

$RepeatedMsgReduction offlocal6.info @<>:514& stop

配置完成后，别忘了重启下 rsyslog 服务！

数据部分

数据部分顾名思义，用于接收并处理客户端发来的操作系统日志。这里我们用到了 rsyslog 和 elasticsearch 两个服务了。

准备rsyslog-elasticsearch

要让 rsyslog 将日志发送给 elastichsearch，我们就必须安装它的 es 模块

# Ubuntu  apt-get install -y rsyslog-elasticsearch rsyslog-mmjsonparse  #CentOSyum install rsyslog-elasticsearch rsyslog-mmjsonparse

准备 ElasticSearch 服务

为了简单部署，本文直接用 docker 快速拉起一个 ES 服务

docker run -d --name elasticsearch  -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" elasticsearch:7.3.1

配置 rsyslog 服务端，创建一个文件 /etc/rsyslog.d/40-audit-server.conf，用于定义日志的写入策略。

$RepeatedMsgReduction off$ModLoad imudp$UDPServerRun 514module(load="mmjsonparse")          # for parsing CEE-enhanced syslog messagesmodule(load="omelasticsearch")      # for outputting to Elasticsearch#try to parse a structured log# this is for index names to be like: rsyslog-YYYY.MM.DDtemplate(name="rsyslog-index" type="string" string="bashaudit-%$YEAR%.%$MONTH%.%$DAY%")# this is for formatting our syslog in JSON with @timestamptemplate(name="json-syslog" type="list") {     constant(value="{ ")      constant(value="\"@timestamp\":\"")     property(name="timegenerated" dateFormat="rfc3339" date.inUTC="on")      constant(value="\",\"host\":\"")        property(name="fromhost-ip")      constant(value="\",\"severity\":\"")    property(name="syslogseverity-text")      constant(value="\",\"facility\":\"")    property(name="syslogfacility-text")      constant(value="\",\"program\":\"")     property(name="programname")      constant(value="\",\"tag\":\"")         property(name="syslogtag" format="json")      constant(value="\",")                   property(name="$!all-json" position.from="2")    # closing brace is in all-json}if ($syslogfacility-text == 'local6' and $syslogseverity-text == 'info') then {  action(type="mmjsonparse") action(type="omelasticsearch" template="json-syslog" searchIndex="rsyslog-index" dynSearchIndex="on" server="" serverport="")        # action(type="omfile" file="/var/log/bashaudit.log")        stop}

这里采用了 rsyslog 的两个 module 来处理收集的日志

1. mmjsonparse用于 json 格式化日志
2. omelasticsearch用于配置 ElastichSearch
3. 配置完成重启 rsyslog 服务

查询部分

审计日志的查询我们可以使用 Kibana 或者自己根据 ElasticSearch API 进行二次开发。这里我们以 Kibana 举例。

cat << EOF   > ./kibana.ymlserver.port: 15601elasticsearch.hosts: ["http://:"]i18n.locale: "zh-CN"EOFdocker run -d --ulimit nofile=1000000:1000000 --net host --name elasticsearch-audit -v ./kibana.yml:/usr/share/kibana/config/kibana.yml  --restart always docker.elastic.co/kibana/kibana-oss:7.3.1

本地访问http://localhost:15601进入 kibana 配置创建一个名为 bashaudit 的索引模式

使用 Shell 轻松搞定 Linux 命令审计