当服务器暴露在公网后,端口扫描几乎是最早出现的探测行为。攻击者希望通过快速连接22、3306、6379等常用端口判断哪些服务开放,从而寻找下一步攻击入口。管理员如果只依赖事后日志分析,往往会错过最佳响应时间。Ruby拥有简洁的Socket标准库,配合系统防火墙可以构建一套从端口状态采集、扫描行为识别到自动封禁的轻量防御链路。

下文不引入重量级监控框架,直接用Ruby标准库和少量系统调用完成这一目标。
一、TCP连接尝试是端口监控的基础
判断一个TCP端口是否开放,本质上是向目标主机发起connect调用。如果目标端口处于监听状态,协议栈会完成三次握手,客户端收到SYN+ACK后返回成功;如果端口关闭,目标一般会回复RST,连接立刻被拒绝;如果主机不可达或防火墙静默丢弃包,则会表现为超时。Ruby的Socket.tcp方法把底层的connect、读写和关闭流程封装了起来,同时支持连接超时时间,非常适合做端口探活。
相比直接使用TCPSocket.new,Socket.tcp的关键优势是connect_timeout参数。默认情况下,TCP连接超时由操作系统内核控制,可能会长达数十秒甚至数分钟。如果不显式设置应用层超时,扫描脚本会在某个不可达主机上长时间卡住,进而影响所有后续端口的检测。下面的函数利用该方法完成一次带超时的连接测试:
require 'socket'
def port_open?(host, port, timeout = 2)
Socket.tcp(host, port, connect_timeout: timeout) do |s|
true
end
true
rescue Errno::ECONNREFUSED, Errno::ETIMEDOUT, Errno::EHOSTUNREACH, SocketError
false
end
这里的异常捕获很关键。Errno::ECONNREFUSED表示目标端口明确关闭,Errno::ETIMEDOUT是超时,Errno::EHOSTUNREACH对应主机不可达,SocketError则覆盖DNS解析失败等错误。捕获这些异常后返回布尔值,可以让上层调用者只关心端口开或关,而不需要处理各种底层网络错误。
二、多端口监控与状态变更记录
单次探活只能确认某一个时间点的状态,监控的意义在于持续采样。可以维护一个需要重点关注的端口列表,例如SSH、HTTP、数据库和缓存服务的默认端口。如果不确定哪些端口开放,可以用Ruby脚本遍历一段连续端口,但要控制并发和超时,否则扫描时间会线性增长。
每次探测完成后,除了输出开关状态,还应该写入带时间戳的结构化日志。这样当安全事件发生时,可以回放某个时间段内端口状态的变化。下面的示例将结果输出为JSON行,便于后续用grep或日志系统过滤:
require 'json'
ports = [22, 80, 443, 3306, 6379, 8080]
host = '127.0.0.1'
ports.each do |port|
state = port_open?(host, port) ? 'open' : 'closed'
log = { time: Time.now.iso8601, host: host, port: port, state: state }
puts JSON.generate(log)
end
这种日志格式的优点在于每行独立、字段清晰,后续无论用脚本统计还是接入ELK都方便。更重要的是,状态变化可以作为告警依据:如果一个原本关闭的端口突然开放,或者一个核心端口从开放变为关闭,都需要立刻通知。可以在此基础之上增加diff逻辑,与上一次结果比较后只推送变化部分。
当然,对公网IP的监控要注意频率。频繁连接目标端口可能被对方防火墙当作扫描行为,甚至被源IP反制。如果监控对象是自有服务器,建议把探测间隔控制在30秒以上,并尽量使用内网或带外管理网络。
三、扫描行为的滑动窗口识别
端口扫描最明显的特征不是连接失败,而是同一来源IP在短时间内尝试访问多个目标端口。正常客户端访问Web服务通常只连接80或443,而扫描器会连续连接22、21、3306、6379、8080等端口,并且无视业务上下文。识别这种模式可以用滑动窗口算法:统计每个来源IP在最近N秒内发起的连接目标端口数量,当去重后的端口数或连接总次数超过阈值时,就认为是扫描。
下面的ScanDetector类实现了最基础的窗口计数。它的record_attempt方法记录一次连接尝试,自动剔除窗口之外的旧记录。当窗口内记录数达到阈值时,触发一次回调,由调用者决定如何处置该IP:
require 'set'
class ScanDetector
def initialize(window: 10, threshold: 5)
@attempts = Hash.new { |h, k| h[k] = [] }
@window = window
@threshold = threshold
@blocked = Set.new
end
def record_attempt(ip, port)
return if @blocked.include?(ip)
now = Time.now.to_f
@attempts[ip] << [now, port]
@attempts[ip].reject! { |time, _| now - time > @window }
if @attempts[ip].size >= @threshold
@blocked.add(ip)
yield ip if block_given?
end
end
def blocked?(ip)
@blocked.include?(ip)
end
end
要减少误报,window和threshold需要根据业务调整。例如10秒内5次连接尝试对普通管理来源来说可能很正常,但如果是10秒内连接30个不同端口则毫无疑问是扫描。还可以进一步增加端口分散度判定,比如同时访问数据库端口和远程桌面端口,这类组合访问往往不是正常用户行为。阈值设置过低会误封,过高则漏报,实际部署时先观察一段时间的正常基线,再让防御自动生效。
四、自动封禁与防火墙联动
识别出扫描IP之后,下一步通常是阻断。Linux下最直接的方案是通过iptables把来源IP加入INPUT链的DROP规则。Ruby可以用system调用系统命令,但必须注意两点:一是IP必须经过格式校验,防止命令注入;二是封禁动作应带有有效期,不能永久丢失所有流量,否则一旦误判会影响真实用户。
下面是一个安全封禁函数,使用IPAddr将输入解析成规范的IP字符串,只接受合法IPv4或IPv6地址,然后再拼接到iptables命令中:
require 'ipaddr'
def ban_ip(ip)
parsed = IPAddr.new(ip)
cmd = "iptables -A INPUT -s #{parsed} -j DROP"
system(cmd)
end
将ScanDetector与ban_ip组合后,可以在回调中直接封禁。但要注意iptables规则重启后不会保留,需要配合iptables-save或自建解封调度器。另一个更隐蔽的思路是部署陷阱端口:开放几个根本不存在的服务端口,一旦有IP连接这些端口,立刻判定为扫描并拉黑。这种方式比统计多端口尝试更精确,误报率更低。
如果主机处于云环境,直接调用云厂商API修改安全组也是类似的思路。但无论哪种方式,都不建议在没有人工确认的情况下永久封禁,应该在脚本中记录封禁时间,并在24小时后自动删除对应iptables规则。防御自动化程度越高,越需要可解释的日志与回滚机制。
Ruby的标准库足够完成端口状态监控、扫描检测和防火墙联动这一整套任务。它不像Go或C那样需要编译,也不像shell脚本那样难以维护复杂状态,非常适合运维团队快速落地。把探活脚本放到cron或systemd timer中周期执行,再配合ScanDetector和iptables规则,就能在扫描器得手之前获得一层基础防御。