運維中，我們有時候想要記錄服務器上誰登錄了、干了什么，前者可以使用之前的ssh登錄報警這篇文章，后者則可以參考本文。
此功能利用的是history命令、PROMPT_COMMAND環境變量、/etc/profile.d/擴展全局環境變量，Linux中的PROMPT_COMMAND會記錄下出現提示符前面的命令，利用這個特性可以實現記錄所有用戶的操作記錄。
環境：
centos6、7，要求安裝iproute(yum install iproute)
原理：
自定義history和PROMPT_COMMAND相關環境變量，放到/etc/profile.d/中令全局生效，記錄所有用戶的命令操作。
步驟：
1. 將以下代碼保存到/etc/profile.d/cmdhistLog.sh
# bash cmd audit
# root touch $HISTFILE && chmod a+w $HISTFILE, yum install iproute
readonly HISTSIZE=10000
readonly HISTFILESIZE=10000
readonly HISTFILE=/tmp/XingkaOpsCmdHist.log
readonly SERVER_IPS=$(/usr/sbin/ip a | grep inet | grep -v inet6 | grep -v 127 | sed 's/^[ \t]*//g' | cut -d ' ' -f2 | awk -F '/' '{print $1}' | xargs | sed 's/\s/,/g')
readonly PROMPT_COMMAND='{ \ if [ -z "$OLD_PWD" ];then export OLD_PWD=$PWD; fi; if [ ! -z "$LAST_CMD" ] && [ "$(history 1)" != "$LAST_CMD" ]; then msg=$(history 1 | { read x y; echo $y; });echo [$(date +"%Y-%m-%d %H:%M:%S")] [\($USER\) $(who am i |awk "{print \$1\" \"\$2\" \"\$NF}" | sed -e "s/[()]//g")] [$(hostname)@$SERVER_IPS] "$msg" >> $HISTFILE; fi; export LAST_CMD="$(history 1)"; export OLD_PWD=$PWD; }'
shopt -s histappend
export HISTSIZE HISTFILESIZE HISTFILE PROMPT_COMMAND
2. 上述保存后，重新登錄的終端即可生效，建議手動創建下日志文件，並需要授予所有人可寫權限
# $HISTFILE改為實際日志文件路徑
touch $HISTFILE && chmod a+w $HISTFILE
3. 使用python清洗日志，篩選出有效的命令數據並上報給運維平台、數據庫等
# PS: 腳本可能需要根據實際情況自行修改。
#!/usr/bin/python
# coding: utf8
# Author: taochengwei
# Email: taochengwei@starokay.com
# Date: 2018-12-11
# Docs: Command the audit log report
import os
import re
import sys
import stat
import json
import urllib
import urllib2
import traceback
reload(sys)
sys.setdefaultencoding('utf-8')
log = os.getenv('HISTFILE', '/tmp/XingkaOpsCmdHist.log')
pat = re.compile(r'^(\[.*\])\s(\[.*\])\s(\[.*\])\s(.*)$')
if not os.path.isfile(log):
with open(log, "w") as f:
f.write("")
os.chmod(log, stat.S_IWUSR+stat.S_IRUSR+stat.S_IRGRP+stat.S_IWGRP+stat.S_IWOTH)
def post(url, data, token=None):
""" POST請求 """
if isinstance(data, dict):
data = urllib.urlencode(data) # 將字典以url形式編碼
headers = {'AccessToken': token} if token else {}
headers.update({"User-Agent": "Mozilla/5.0 (X11; CentOS; Linux i686; rv:7.0.1406) Gecko/20100101 OpsRequestBot/0.1", "Content-Type": "application/json"})
request = urllib2.Request(url, data=data, headers=headers)
response = urllib2.urlopen(request)
response = response.read()
try:
data = json.loads(response)
except Exception, e:
traceback.print_exc()
data = response
return data
if __name__ == "__main__" and log and os.path.isfile(log):
access_token = ""
# read log
with open(log, 'rb') as f:
data = f.read()
os.chmod(log, stat.S_IWUSR+stat.S_IRUSR+stat.S_IRGRP+stat.S_IWGRP+stat.S_IWOTH)
# set post data pool
# pool 是篩選后的有效命令數據，是一個列表，可以上報到運維平台或寫入數據庫中
pool = []
for cmd in data.split("\n"):
if pat.match(cmd):
cmd = [i for i in pat.split(cmd) if i]
if cmd and isinstance(cmd, (list, tuple)) and len(cmd) == 4:
try:
# Time to execute the command
ctime = cmd[0].lstrip('[').rstrip(']')
# Switch user, real login user, terminal for pts or tty, client ip
suser, luser, terminal, clientIp = cmd[1].lstrip('[').rstrip(']').split()
suser = suser.replace('(', '').replace(')', '')
# the server hostname and all ip
hostname, serverIps = cmd[2].lstrip('[').rstrip(']').split('@')
# real bash command
command = cmd[3]
except Exception, e:
traceback.print_exc()
print cmd
else:
pool.append(dict(ctime=ctime, suser=suser, luser=luser, terminal=terminal, clientIp=clientIp, hostname=hostname, serverIps=serverIps, command=command))
# post data with pool
res = post("接收上報的運維平台接口地址", json.dumps(pool), access_token)
if res["code"] == 0:
with open(log, "w") as f:
f.write("")
print(res)
4. 后端處理
# 這是后端運維平台的接口路由函數，它接收命令審計數據，並用異步任務寫入數據庫
def serverCmdHist(self, data):
"""服務器歷史命令記錄"""
res = dict(code=1, msg=None)
if data and isinstance(data, (list, tuple)):
# 需要進一步篩選有效條目
pool = []
for log in data:
if log and isinstance(log, dict) and \
"ctime" in log and \
"serverIps" in log and \
"hostname" in log and \
"terminal" in log and \
"command" in log and \
"suser" in log and \
"luser" in log and \
"clientIp" in log:
# check log params
try:
if not user_pat.match(log["luser"]) or not ip_check(log["clientIp"]) or not log["hostname"] or not log["command"]:
raise
datetime_to_timestamp(log["ctime"])
except:
pass
else:
# 此處生成一條有效命令唯一標識，避免誤提交審計日志時重復寫入數據庫
log.update(oci=md5("%s-%s-%s-%s-%s" %(log["hostname"], log["ctime"], log["luser"], log["suser"], log["command"])))
pool.append(log)
# 此處pool中的命令記錄都應該是有效的，放到異步任務中寫入到mysql
self.asyncQueueHigh.enqueue_call(func=ServerCmdHist2MySQL, args=(pool,), timeout=3600)
res.update(code=0)
else:
res.update(msg="data error")
return res
5. 數據庫表結構
CREATE TABLE `servercmdhist` (
`id` int(10) unsigned NOT NULL AUTO_INCREMENT,
`oci` char(32) NOT NULL COMMENT '命令唯一標識',
`hostname` varchar(150) NOT NULL COMMENT '主機名',
`serverIps` varchar(255) DEFAULT NULL COMMENT '主機ip',
`ctime` char(19) NOT NULL COMMENT '命令執行的時間',
`luser` varchar(32) NOT NULL COMMENT '實際登錄的用戶',
`suser` varchar(32) DEFAULT NULL COMMENT '實際用戶切換后的用戶',
`terminal` varchar(50) DEFAULT NULL COMMENT '登錄終端類型',
`clientIp` varchar(15) NOT NULL COMMENT '登錄來源ip',
`command` varchar(10000) NOT NULL COMMENT '執行的命令',
PRIMARY KEY (`id`),
UNIQUE KEY `oci` (`oci`),
KEY `hostname` (`hostname`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;