该脚本位于/opt/zeek/share/zeek/policy/protocols/ssh/detect-bruteforcing.zeek。最近的更新在15年，是很久没有更新的一个脚本。

$ ls -lh
total 16K
-rw-r--r-- 1 root root 3.2K Jan 28  2015 detect-bruteforcing.zeek
-rw-r--r-- 1 root root 1.5K Jan 28  2015 geo-data.zeek
-rw-r--r-- 1 root root 1.6K Jan 28  2015 interesting-hostnames.zeek
-rw-r--r-- 1 root root 1008 Jan 28  2015 software.zeek

1. 相关参数定义

首先是一系列参数的定义：

export {
	redef enum Notice::Type += {
		## ...
		Password_Guessing,
		## ...
		Login_By_Password_Guesser,
	};

	redef enum Intel::Where += {
		## An indicator of the login for the intel framework.
		SSH::SUCCESSFUL_LOGIN,
	};

	## The number of failed SSH connections before a host is designated as
	## guessing passwords.
	const password_guesses_limit: double = 30 &redef;

	## The amount of time to remember presumed non-successful logins to
	## build a model of a password guesser.
	const guessing_timeout = 30 mins &redef;

	## ...
	const ignore_guessers: table[subnet] of subnet &redef;
}

定义了2个Notice的新type，因为该脚本的检测结果最终输出到notice.log。从命名上容易发现：Password_Guessing定义了暴力破解事件；Login_By_Password_Guesser定义暴力破解成功登录事件。
目前业界检测暴破的一种方法是检测基于时间窗口内的尝试次数是否超过预设阈值。该方法简单粗暴，可能漏报跨时间窗口的尝试；但是实现简单，效率也高。该脚本也基于同样的逻辑。password_guesses_limit定义尝试频次阈值，默认30次；guessing_timeout定义了相应时间窗口，默认30分钟。
ignore_guessers定义IP白名单。

2. 使用SumStats Framework实现检测逻辑

Zeek提供了一个统计框架SumStats，主要用于统计各种网络数据的信息，也适用于实现一系列基于统计频次方法的检测脚本。SumStats框架¹的使用主要分为3个步骤：

Observation：观察，我理解是收集数据，譬如该脚本收集每一次SSH尝试登录信息
Reducer：对收集的数据进行处理，譬如统计求和等
Sumstat：对数据处理结果进行响应，我理解相当于action，譬如这里统计SSH尝试登录频次超过阈值，输出到notice日志等

注意Zeek架构是分布式的，所以推测SumStats的框架也借鉴了Map Reduce的方法逻辑。

接下来看该检测脚本怎么使用以上3个步骤实现检测逻辑。

2.1 观察ssh_auth_failed事件

ssh_auth_failed(c: connection)**是zeek提供的一个事件，当zeek检测到有SSH登录失败时触发：

event ssh_auth_failed(c: connection)
	{
	local id = c$id;

	# Add data to the FAILED_LOGIN metric unless this connection should
	# be ignored.
	if ( ! (id$orig_h in ignore_guessers &&
	        id$resp_h in ignore_guessers[id$orig_h]) )
		SumStats::observe("ssh.login.failure", [$host=id$orig_h], [$str=cat(id$resp_h)]);
	}

当该事件触发时，对于不在白名单内的IP使用observe方法进行观察。observe方法有3个参数：

Id：string类型，对观察到的数据进行标识。注意到observe可能会在多个脚本中被调用，用于观察不同的数据信息，因此需要对不同的数据流进行唯一标识，譬如这里使用"ssh.login.failure"标识SSH登录失败事件。
Key：SumStats::Key类型，实质是一个record，其中可以包含str和host2种字段。这里就使用了$host，ssh登录失败的源IP作为key。
Obs：SumStats::Observation类型，也是一个record，可以包含num、dbl和str3种字段，这里使用了$str类型，保存ssh登录失败目的IP。

2.2 reduce和sumstat

对观察数据进行reduce和sumstat是在事件**zeek_init()**中处理的。

2.2.1 Reducer

首先定义了SumStats::Reducer：

event zeek_init()
	{
	local r1: SumStats::Reducer = [$stream="ssh.login.failure", $apply=set(SumStats::SUM, SumStats::SAMPLE), $num_samples=5];
	...
	}

Reducer是record类型，其中包含

stream：数据流的标识，即对哪个数据流进行reduce处理，需要和对应的observe方法的Id保持一致
apply：一个包含SumStats::Calculation的set类型，定义对观察到的数据流进行什么样的处理。这里使用了SUM和SAMPLE 2种方法：
- SUM
Calculate the sum of the values. For string values, this will be the number of strings.
因为在观察阶段把SSH尝试登录的目的IP保存到string类型的字符串中，所以这里的SUM是string的统计数量，也就是SSH失败尝试的数量。
- SAMPLE
Get uniquely distributed random samples from the observation stream.
对观察到的SSH尝试登录的目的IP进行抽样，抽样的数量基于num_samples。

2.2.2 sumstat

接下来是经过reduce之后的action处理，使用sumstat方法。

event zeek_init()
	{
	...
		SumStats::create([$name="detect-ssh-bruteforcing",
	                  $epoch=guessing_timeout,
	                  $reducers=set(r1),
	                  $threshold_val(key: SumStats::Key, result: SumStats::Result) =
	                  	{
	                  	return result["ssh.login.failure"]$sum;
	                  	},
	                  $threshold=password_guesses_limit,
	                  $threshold_crossed(key: SumStats::Key, result: SumStats::Result) =
	                  	{
	                  	local r = result["ssh.login.failure"];
	                  	local sub_msg = fmt("Sampled servers: ");
	                  	local samples = r$samples;
	                  	for ( i in samples )
	                  		{
	                  		if ( samples[i]?$str )
	                  			sub_msg = fmt("%s%s %s", sub_msg, i==0 ? "":",", samples[i]$str);
	                  		}
	                  	# Generate the notice.
	                  	NOTICE([$note=Password_Guessing,
	                  	        $msg=fmt("%s appears to be guessing SSH passwords (seen in %d connections).", key$host, r$num),
	                  	        $sub=sub_msg,
	                  	        $src=key$host,
	                  	        $identifier=cat(key$host)]);
	                  	}]);
	}

sumstat::create包括：

name：该统计分析结果的名字，这里是ssh暴破检测
epoch：时间周期，在参数中定义的检测时间窗口
threashold_val：因为需要和预设阈值比较，那么需要统计在时间窗口内观察到的SSH登录失败次数。由于在reducer使用了SUM方法，这里就是SUM方法的结果
threashold：参数中定义的预设阈值
threashold_crossed：zeek提供的一个方法，即zeek检测到统计结果超过阈值时:
- 统计结果保存result["ssh.login.failure"]
- 由于reducer同样使用SAMPLE方法，保存抽样的SAMPLE在sub_msg
- 输出相应的信息到notice.log

3. 脚本验证

使用该脚本对一包含SSH暴力破解流量的pcap文件进行检测，成功生成告警信息到notice.log：

$ cat notice.log | jq
{
  "ts": 1617095241.708715,
  "note": "SSH::Password_Guessing",
  "msg": "192.168.153.6 appears to be guessing SSH passwords (seen in 30 connections).",
  "sub": "Sampled servers:  192.168.153.18, 192.168.153.18, 192.168.153.18, 192.168.153.18, 192.168.153.18",
  "src": "192.168.153.6",
  "actions": [
    "Notice::ACTION_LOG"
  ],
  "suppress_for": 3600
}

Footnotes

Summary Statistics ↩