系统安全之SSH入侵的检测与响应

Jan 14 07:37:08 victim sshd[54689]: pam_unix(sshd:auth): authentication failure; logname= uid=0 euid=0 tty=ssh ruser= rhost=192.168.171.130 user=victim

Jan 14 07:37:10 victim sshd[54689]: Failed password for victim from 192.168.171.130 port 48548 ssh2

3. 用户名正确且输入正确密码的日志如下所示：

Jan 14 07:37:30 victim sshd[54689]: Accepted password for victim from 192.168.171.130 port 48548 ssh2

Jan 14 07:37:30 victim sshd[54689]: pam_unix(sshd:session): session opened for user victim by (uid=0)

4. 客户端主动退出ssh连接的日志如下所示

Jan 14 07:37:38 victim sshd[54694]: Received disconnect from 192.168.171.130 port 48548:11: disconnected by user

Jan 14 07:37:38 victim sshd[54694]: Disconnected from 192.168.171.130 port 48548

Jan 14 07:37:38 victim sshd[54689]: pam_unix(sshd:session): session closed for user victim

5. 客户端强制关闭ssh连接的日志如下所示

Jan 14 07:42:25 victim sshd[54718]: Connection closed by 192.168.171.130 port 48552 [preauth]

由上所述，我们需要做的检测策略如下：

1. 能够检测到尝试登陆行为

1）这里我们设定10秒钟内发现5条存在 invalid user 或者password check failed语句的记录则表示存在尝试登陆行为（规则可自定义：包括时间和记录数）

2. 能够检测到登陆成功行为

1）检索所有日志发现存在 accepted password for 语句的记录则判定存在登陆成功情况

3. 能够检测到登陆成功账户

1）提取日志中匹配到accepted password for关键词语句后面的一个字段

4. 收集用户字典

1）提取日志中匹配到Invalid user关键词语句后面的一个字段

5. 记录登录失败的用户名/次数、登录失败用户正确的次数、登录成功的用户名/次数、登陆成功的攻击源IP地址/尝试次数、登录失败的攻击源IP地址/尝试次数

1）通过各种计算方法来统计如上信息

检查脚本如下：

#!/usr/bin/python

#coding:utf-8

from collections import Counter

#定义关键词信息

username_error='Invalid user'

username_correct='password check failed'

username_password_correct='Accepted password for'

user_quit='Received disconnect from'

user_forcequit='Connection closed'

#打开日志文件

f=open('secure','r')

#定义检测方法

def attack_detect():

#用户名错误的请求次数

failed_account=0

#用户名正确且密码错误的请求次数

correct_user_account=0

#用户名正确且密码正确的请求次数

correct_pass_account=0

#记录错误的用户名

failed_user=[]

#记录用户名正确且密码错误的用户名

correct_user=[]

#记录用户名正确且密码正确的用户名

correct_pass=[]

#记录登录失败的ip地址

failed_ipaddr=[]

#记录登录失败&用户名正确的ip地址

correct_ipaddr=[]

#记录登录成功的ip地址

correct_pass_ipaddr=[]

#标志位

alert=True

for i in f:

if username_error in i:

failed_account += 1

failed_user.append(i.split(': ')[1].split()[2])

failed_ipaddr.append(i.split(': ')[1].split()[4])

if username_correct in i:

correct_user_account += 1

correct_user.append(i.split(': ')[1].split('(')[1].strip(')'))

if username_password_correct in i:

correct_pass_account += 1

correct_pass.append(i.split(': ')[1].split()[3])

correct_pass_ipaddr.append(i.split(': ')[1].split()[5])

if failed_account > 30 and alert:

print 'exists ssh enumrate'

alert=False

#记录登陆失败攻击源IP地址和尝试次数

failed_ipaddr_count=Counter(failed_ipaddr)

failed_ipaddr_dict=dict(failed_ipaddr_count)

#记录登陆成功攻击源IP地址和尝试次数

correct_pass_ipaddr_count=Counter(correct_pass_ipaddr)

correct_pass_ipaddr_dict=dict(correct_pass_ipaddr_count)

#记录登陆失败用户名和次数

failed_user_count=Counter(failed_user)

failed_user_dict=dict(failed_user_count)

#记录登陆失败用户名正确和次数

correct_user_count=Counter(correct_user)

correct_user_dict=dict(correct_user_count)

#记录登陆成功用户名和次数

correct_pass_count=Counter(correct_pass)

correct_pass_dict=dict(correct_pass_count)

#记录所有尝试次数

all_account=failed_account+correct_user_account+correct_pass_account

return all_account,failed_account,correct_user_account,correct_pass_account,failed_user_dict,correct_user_dict,correct_pass_dict,failed_ipaddr_dict,correct_pass_ipaddr_dict

然后执行该脚本可以得出如下结果：

[/url]从结果中可以看出该脚本能够满足我们的需求，但是我们需要考虑如何把这段脚本加入到spark streaming中。

八、技术实现

1. victim主机上的flume配置

[url=http://image.3001.net/images/20190130/1548819473_5c511c1143ae5.png]注意：这里在配置channels选项的时候，增加了下面两个条目：

a1.channels.c1.keep-alive = 60

a1.channels.c1.capacity = 1000000

这里表示增加channels中queue的大小，默认为100，如果使用默认值，当处理大量的日志时就会发生如下报错情况。

[/url]

2. observer主机上的flume配置

[url=http://image.3001.net/images/20190127/1548585155_5c4d88c3a79a8.png]3. observer上的kafka、zookeeper配置

kafka和zookeeper的配置保持不变即可

4. spark streaming配置

#!/usr/bin/python

#coding:utf-8

from pyspark import SparkContext

from pyspark.streaming import StreamingContext

from pyspark.streaming.kafka import KafkaUtils

from collections import Counter

import datetime

import json

#定义保存的文件名

destname="result.json"

#定义读取文件方法

def readjson(file_name):

with open (file_name,"r") as file_obj:

numbers = json.load(file_obj)

print "读取json文件：",numbers

return numbers

#定义写入文件方法

def writejson(file_name,nums):

with open (file_name,"w") as file_obj:

json.dump(nums,file_obj)

print "写入json文件：",nums

#第一步：创建一个本地的StreamingContext，并设置批处理周期为1s

sc=SparkContext("local[2]","flumeWordCount")

ssc=StreamingContext(sc,1)

#第二步：创建一个kafka连接

topic="streamingtopic"

brokers="hadoop0:9092"

#参数分别表示ssc连接名，列表形式显示的topic名称，brokers列表

directkafkaStream = KafkaUtils.createDirectStream(ssc,[topic],{"metadata.broker.list":brokers})

#第三步：数据处理&转换

#0，提取kafka数据的第一个字段为消息正文字段

lines = directkafkaStream.map(lambda x: x[1])

#1，定义检测方法

def attack_detect(i):

username_error='Invalid user'

username_correct='password check failed'

username_password_correct='Accepted password for'

user_quit='Received disconnect from'

user_forcequit='Connection closed'

failed_account=0

correct_user_account=0

correct_pass_account=0

failed_user=[]

correct_user=[]

correct_pass=[]

failed_ipaddr=[]

correct_ipaddr=[]

correct_pass_ipaddr=[]

alert=True

if username_error in i:

print username_error

failed_account += 1

failed_user.append(i.split(': ')[1].split()[2])

failed_ipaddr.append(i.split(': ')[1].split()[4])

#print failed_account

if username_correct in i:

#print username_correct

correct_user_account += 1

correct_user.append(i.split(': ')[1].split('(')[1].strip(')').strip(')\n'))

if username_password_correct in i:

#print username_password_correct

correct_pass_account += 1

correct_pass.append(i.split(': ')[1].split()[3])

correct_pass_ipaddr.append(i.split(': ')[1].split()[5])

if failed_account > 5 and alert:

#print 'exists ssh enumrate'

alert=False

total_count=failed_account+correct_user_account+correct_pass_account

#记录登陆失败攻击源IP地址和尝试次数

failed_ipaddr_count=Counter(failed_ipaddr)

failed_ipaddr_dict=dict(failed_ipaddr_count)

#记录登陆成功攻击源IP地址和尝试次数

correct_pass_ipaddr_count=Counter(correct_pass_ipaddr)

correct_pass_ipaddr_dict=dict(correct_pass_ipaddr_count)

#记录登陆失败用户名和次数

failed_user_count=Counter(failed_user)

failed_user_dict=dict(failed_user_count)

#记录登陆失败用户名正确和次数

correct_user_count=Counter(correct_user)

correct_user_dict=dict(correct_user_count)

#记录登陆成功用户名和次数

correct_pass_count=Counter(correct_pass)

correct_pass_dict=dict(correct_pass_count)

#这里将处理的数据以字典的形式保存到json文件中，每次从json文件中读取变量值并跟处理的结果进行累加

nums=[{"total_count":total_count,"failed_account":failed_account,"correct_user_account":correct_user_account,"correct_pass_account":correct_pass_account,"failed_ipaddr_count":failed_ipaddr_count,"failed_ipaddr_dict":failed_ipaddr_dict,"correct_pass_ipadd_count":correct_pass_ipaddr_count,"correct_pass_dict":correct_pass_dict,"correct_pass_ipaddr_dict":correct_pass_ipaddr_dict,"failed_user_count":failed_user_count,"failed_user_dict":failed_user_dict,"correct_user_count":correct_user_count,"correct_user_dict":correct_user_dict}]

rs_old = readjson(destname)

for i in nums:

for j in rs_old:

for k,v in j.items():

if isinstance(v,dict):

for m,n in v.items():

try:

i[k][m] = i[k][m] + j[k][m]

except KeyError as e:

i[k][m] = j[k][m]

else:

i[k] = i[k] + j[k]

writejson(destname,nums)

#第四步：数据输出，这里选择打印出去

pairs=lines.map(lambda x:attack_detect(x))

pairs.pprint()

#第五步：开始sparkstreaming

ssc.start()

ssc.awaitTermination()

5. 启动各个组件

1）先启动zookeeper和kafka

./zkServer.sh start

./bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties

2）然后启动flume-server端（先监听，后发送）

./bin/flume-ng agent --name a1 --conf $FLUME_HOME/conf --conf-file $FLUME_HOME/conf/collect-ssh-kafka.conf -Dflume.root.logger=INFO,console

3）启动flume-client端（监听文件，发送数据到server端）

sudo ./flume-ng agent --name a1 --conf $FLUME_HOME/conf --conf-file $FLUME_HOME/conf/send-ssh.conf -Dflume.root.logger=INFO,console

4）启动spark-streaming

./bin/spark-submit --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 examples/src/main/python/streaming/sshdirect.py

6. 模拟攻击者向victim发起ssh爆破行为

hydra -L user -P pass ssh://192.168.171.121:3389

[/url]然后我们查看result.json文件中的内容不断实时更新，攻击完成后的最终结果如下：

[url=http://image.3001.net/images/20190130/1548820070_5c511e6660d3d.png]

由上所述我们能够看到文件记录了ssh总攻击次数、用户名错误尝试次数、用户名正确登陆次数以及登陆源ip地址等信息。

注：

1. 这里我使用的是以文件形式进行存储的，没有使用mysql数据库，理论上方法是一样的，这里我就不演示了，有兴趣的小伙伴可以自己搞

2. 这里的result.json文件内容是固定的，必须要跟spark streaming脚本中的变量格式一致，否则可能会出现问题，后面附录中我会把格式文件贴上来

九、图形页面展示

这里我就以一个例子来展示吧，以条形图的形式展示登陆失败的用户情况。

1，django环境安装

注：已安装django环境的自动忽略

1）使用virtualenv环境

virtualenv -p /usr/bin/python3 myechart

source myechart/bin/activate

pip install django==1.11.4

pip install pyecharts

2）新建django项目和app程序

django-admin startproject myechart

cd sshechart

python manage.py startapp sshechart

编辑myechart/settings.py文件，注册应用程序：

[/url]编辑sshechart的urls文件，该文件默认不存在，需要创建。内容如下：

from django.conf.urls import url

from . import views

urlpatterns = [

url(r'^$', views.index, name='index'),

]

编辑myechart的urls文件，更新内容如下：

[url=http://image.3001.net/images/20190131/1548902817_5c5261a131f6d.png]2. 数据处理

1）编辑views.py文件内容如下：

from __future__ import unicode_literals

import math

from django.http import HttpResponse

from django.template import loader

from pyecharts import Line

from django.shortcuts import render

import json

# Create your views here.

#加载echart.js文件

REMOTE_HOST = "https://pyecharts.github.io/assets/js"

def index(request):

#指定模板文件

template = loader.get_template('sshechart/index.html')

attack_line=line1()

context = dict(

myechart=attack_line.render_embed(),

host=REMOTE_HOST,

script_list=attack_line.get_js_dependencies()

)

return HttpResponse(template.render(context, request))

def line1():

#Line图需要指定x轴和y轴数据

x_raw=[]

y_raw=[]

with open('result.json') as f:

numbers = json.load(f)

for i in numbers:

dic=i["failed_user_dict"]

print (dic)

for k,v in dic.items():

x_raw.append(k)

y_raw.append(v)

line_attack=Line("登陆失败用户",title_pos="45%")

line_attack.add("",x_raw,y_raw,mark_point=["max","min"],is_datazoom_show=True,datazoom_type="inside")

return line_attack

2）创建模板index.html文件

在sshechart目录下创建templates/sshechart目录。

[/url]

并编辑index.html文件，内容如下：

<!DOCTYPE html>

<html>

<head>

<title>Proudly presented by PycCharts</title>

{% for jsfile_name in script_list %}

{% endfor %}

</head>

<body>

</body>

</html>

3. 运行程序

运行程序后，在浏览器中打开目标网站，如下所示：

[url=http://image.3001.net/images/20190131/1548903126_5c5262d68b13b.png]同样的，还可以多添加几个echart图，增加分析维度。

也可以将这个里面的源代码拷贝下来集成到其他web程序中。

至此，这个课题就此结束，期间也是碰到各种问题，拖延了不少时间，好在最终是搞定了，希望大家能够在这里有所收获，也希望大家能够指出文章的不足之处，共勉之。

附录：

1. result.json文件格式

[{"failed_account": 0, "correct_user_dict": {}, "correct_pass_ipaddr_count": {}, "correct_pass_account": 0, "failed_ipaddr_dict": {}, "failed_user_dict": {}, "correct_pass_ipaddr_dict": {}, "correct_pass_dict":{},"total_count": 0, "correct_pass_ipadd_count": {}, "correct_user_count": {}, "correct_user_account": 0, "failed_user_count": {}, "failed_ipaddr_count": {}}]

系统安全之SSH入侵的检测与响应

本帖子中包含更多资源