防火墙Deny记录检测

实现思路

部署在不同区域的网络日志采集服务器实时收集不同型号防火墙设备日志，通过syslog推送到日志采集服务器，日志采集服务器利用流式处理的方式，实时过滤其中deny的记录，本地入库记录，并推送邮件负责人。必要时驱动网络变更接口，自动提交防火墙变更申请，减少因防火墙不通引发的环境问题。

具体原则如下：

1、过滤日志中Deny关键字，并不同网络设备中的IP和端口信息

2、根据IP地址信息匹配CMDB的应用信息，当源IP和目标IP其中有一个在CMDB登记过，即入库记录

3、当源IP和目标IP同时在CMDB中存在记录时，触发邮件提醒，收件人是CMDB中登记角色的人员。

4、利用redis缓存做报警间隔的控制，目前重复报警一个小时推送一次

防火墙Deny的两种类型

参考提醒邮件中“网络设备日志”项，报警日志可以分为两种类型

1、因防火墙限制，直接被拒绝的请求，这类报警需要提交开通防火墙的申请。例如以下两种网络设备的日志：

Deny tcp src YEWU2:123.25.142.29/60452 dst JIERU1:152.186.218.90/445 by access-group “YEWU2_in_ALL” [0x0, 0x0]

NetScreen device_id=BS52FW0A-M5 [Root]system-notification-00257(traffic): start_time=”2018-06-12 08:33:28” duration=0 policy_id=1 service=syslog proto=17 src zone=CS dst zone=M5 action=Deny sent=0 rcvd=177 src=123.1.2.143 dst=123.7.35.2 src_port=1041 dst_port=514 session_id=0

2、防火墙上面开通策略的连接，session因超时被防火墙断开，但是后续应用发送的数据包没有sync包，所以连接建立不起来被丢弃的日志，这种应用一般要开通长连接。如果应用自身有keep-alive机制，则不需要网络上开通长连接。例如如下日志，存在 no connecttion 关键字：

Deny TCP (no connection) from 123.253.19.3/443 to 83.28.254.14/55312 flags RST on interface JIERU1

数据流图

邮件提醒

【标题】：提醒:<网络设备记录的报警时间> <源应用> 到 <目标应用> 访问被防火墙Deny

【示例】：提醒:2018-06-12 08:56:51 AAA 到 BBB 访问被防火墙Deny

【正文】：

具体信息如下:

时间点:<网络设备记录的报警时间>

原地址:<源IP>

原端口:<原端口>

原地址信息:<原IP地址在CMDB中登记的信息>

目标地址:<目标IP>

目标端口:<目标端口>

目标地址信息:<目标IP地址在CMDB中登记的信息>

网络设备日志:<网络设备日志>

代码

import re
import redis
from pykafka import KafkaClient
from sqlalchemy.orm import *
from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base
import datetime
import os
Base = declarative_base()
os.environ['NLS_LANG']='AMERICAN_AMERICA.AL32UTF8'
from ELK.NOTES import *#mysql中记录deny连接记录
class DENYLOG(Base):__tablename__ = "DENYLOG"logtime = Column("logtime", DateTime(), primary_key=True)srcip = Column("srcip", String(20))srcport = Column("srcport", String(10))srcdesc = Column("srcdesc", String(2000))tgtip = Column("tgtip", String(20))tgtport = Column("tgtport", String(10))tgtdesc = Column("tgtdesc", String(2000))mesg = Column("mesg", Text())regex_ipport0=r'((?<![0-9])(?:(?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2}))(?![0-9])*)/([1-9][0-9]*)'
regex_ipport1=r'src=(.*) dst=(.*) src_port=(.*) dst_port=([1-9][0-9]*)'
regex_ipport2=r'source-ip=(.*?), source-port=(.*?), destination-ip=(.*?), destination-port=(.*?), '
blacklist=["123.4.19.11","123.6.35.7","123.6.131.101"]
whitelist=["123","124","125"]#黑名单和白名单过滤
def IpFilter(ip):ipstr=ip.split(".")[0]#print("ipstr",ipstr)if ipstr in whitelist and ip not in blacklist:return 0else:print(ip,"命中过滤条件")return 1#按照应用过滤，过滤一些监控类或运维类相关的广播数据
def AppFilter(srcapp,tgtapp):#print("ipstr",ipstr)if srcapp=="F-AAA" and tgtapp=="F-AAA":print("命中APP过滤条件")return 1elif srcapp=="F-BBB" and tgtapp=="F-BBB":print("命中APP过滤条件")return 1elif srcapp in ["V-z1",'V-z2','V-z3','V-z4'] or tgtapp in ["V-z1",'V-z2','V-z3','V-z4']:print("命中APP过滤条件")return 1else:return 0#根据CMDB绑定应用节点信息
def getdetail(ip):engine = create_engine('oracle://test:test@127.0.0.1/test')DBsession = sessionmaker(bind=engine)session = DBsession()str = "SELECT APP,ENV,SRVZONE,NODE,FUNC TYPE FROM V_APPNODE WHERE IP='%s' ORDER BY CHECKTIME DESC"%ipe=session.execute(str)r=e.fetchone()session.close()if r:app=r[0]env=r[1]srvzone=r[2]node=r[3]func=r[4]return {"app": app, "env": env, "srvzone": srvzone, "node": node, "func": func}else:return {"app":"nodefine"}def setmysql(logtime,srcip,srcport,srcdesc,tgtip,tgtport,tgtdesc,mesg):engine = create_engine("mysql://test:test@123.28.206.32:3307/test?charset=utf8")DBsession = sessionmaker(bind=engine)session = DBsession()denylog=DENYLOG(logtime=logtime,srcip=srcip,srcport=srcport,srcdesc=srcdesc,tgtip=tgtip,tgtport=tgtport,tgtdesc=tgtdesc,mesg=mesg)session.add(denylog)session.commit()#利用redis做报警收敛
def setredis(key,value):r = redis.StrictRedis(host='123.28.206.32', port=6379)logtime=valuervalue=r.get(key)if rvalue:lasttime=str(rvalue,encoding='utf-8')lasttime=datetime.datetime.strptime(lasttime, "%Y-%m-%d %H:%M:%S")if lasttime+ datetime.timedelta(hours=1)<logtime:r.set(key, logtime)return 0else:print(lasttime,logtime,key,'距离上次报警没有超过1小时')return 1else:r.set(key,logtime)return 0def findcase(logtext):r = redis.StrictRedis(host='123.28.206.32', port=6379)try:# 提取tradeid,apitext,非贪婪模式s = logtext.split("###")# print('s', s)host = s[0]timestamp = s[1]message = s[2]logtime=datetime.datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S.%fZ") + datetime.timedelta(hours=8)# print("message",message)rule = re.compile(r'deny', re.S | re.I)# print("tradeid",tradeid)denymsg = rule.search(message)if denymsg:if re.compile(regex_ipport0, re.S | re.I).findall(message):ipport=re.compile(regex_ipport0, re.S | re.I).findall(message)srcip=ipport[0][0]srcport=ipport[0][1]tgtip=ipport[1][0]tgtport=ipport[1][1]elif re.compile(regex_ipport1, re.S | re.I).findall(message):ipport = re.compile(regex_ipport1, re.S | re.I).findall(message)srcip=ipport[0][0]srcport=ipport[0][2]tgtip=ipport[0][1]tgtport=ipport[0][3]else:ipport = re.compile(regex_ipport2, re.S | re.I).findall(message)srcip=ipport[0][0]srcport=ipport[0][1]tgtip=ipport[0][2]tgtport=ipport[0][3]key=srcip+"_"+tgtip+"_"+tgtportif IpFilter(srcip)==0:flag=setredis(key,logtime)#if flag=0 发邮件,开火墙srcdesc=getdetail(srcip)tgtdesc=getdetail(tgtip)#print(logtime, srcip, srcport, srcdesc, tgtip, tgtport, tgtdesc, message)setmysql(logtime, srcip, srcport, srcdesc, tgtip, tgtport, tgtdesc, message)if flag==0:if "nodefine"  not in [srcdesc["app"],tgtdesc["app"]]:notesid=getuser_by_ip(srcip)text="您的应用出现被防火墙Deny的记录,请了解<br/>具体信息如下:<br/>" \"时间点:%s<br/>" \"原地址:%s<br/>" \"原端口:%s<br/>" \"原地址信息:%s<br/>" \"目标地址:%s<br/>" \"目标端口:%s<br/>" \"目标地址信息:%s<br/>" \"网络设备日志:%s<br/>" \"报警规则请参考: <link>http://123.20.17.88/article/81/</link> <br/>"%(logtime,srcip,srcport,srcdesc,tgtip,tgtport,tgtdesc,message)subject="提醒:%s %s 到 %s 访问被防火墙Deny"%(logtime,srcdesc["app"],tgtdesc["app"])if AppFilter(srcdesc["app"],tgtdesc["app"])==0:sendmail(notesid,text,subject)print(notesid,text,subject)else:pass#print("notfound")except Exception as e:#print(message)print(e)passdef myjob(x):print(" JOB %s START ....."%x)# 启动kafka监控进程client = KafkaClient(hosts="123.28.206.32:9092")topic = client.topics[str.encode("netsyslog")]consumer = topic.get_simple_consumer(consumer_group=b"t1",auto_commit_enable=True,auto_commit_interval_ms=1,consumer_id=str.encode(str(topic)))for x in consumer:if x is not None:#seq = int(datetime.datetime.strftime(datetime.datetime.now(), "%Y%m%d%H%M%S%f"))# print(x.value.decode('utf-8'))findcase(x.value.decode('utf-8'))if __name__ == '__main__':myjob("1")

防火墙Deny记录实时检测相关推荐

【计算机网络学习笔记18】防火墙技术、入侵检测技术
[计算机网络学习笔记18]防火墙技术.入侵检测技术一.防火墙防火墙 (firewall) :一种访问控制技术,通过严格控制进出网络边界的分组,禁止任何不必要的通信,从而减少潜在入侵的发生,尽可能降 ...
MPB：亚热带生态所葛体达组-原位酶谱法高分辨率实时检测土壤微界面酶活分布...
为进一步提高<微生物组实验手册>稿件质量,本项目新增大众评审环节.文章在通过同行评审后,采用公众号推送方式分享全文,任何人均可在线提交修改意见.公众号格式显示略有问题,建议电脑端点击文末阅 ...
实时检测神经振荡可实现行为相关的神经反馈
目录本分享为脑机学习者Rose整理发表于公众号:脑机接口社区 QQ交流群:941473018 神经振荡-也称脑电波,是大脑中重要的信息载体,越来越多的研究者将它们解释为瞬态脉冲而不是持续的振荡.短脉 ...
如何使用yunyang tensorflow-yolov3训练自己的数据集并使用 Intel realsense D435 传输视频流应用于实时检测？
训练详细步骤:YunYang1994/tensorflow-yolov3 训练自己的数据集我这边使用自己的数据集已经训练好了权重文件,将该权重文件应用于evaluate.py效果不错,但用video ...
2021极术通讯-CSL-YOLO | 超越Tiny-YOLO V4，全新设计轻量化YOLO模型实现边缘实时检测
首发极术社区如对Arm相关技术感兴趣,欢迎私信aijishu20加入技术微信群. 导读:极术社区与E-learning平台联合推出极术通讯,引入行业媒体和技术社区.咨询机构优质内容,定期分享产业技术 ...
薄膜表面瑕疵在线检测系统实时检测出表面瑕疵的具体位置
薄膜表面瑕疵在线检测系统实时检测出表面瑕疵的具体位置导读:随着人们对产品质量要求越来越高,产品表面质量检测变得日益重要,过去很多表面缺质量测是依靠人工检测,这种检测方法效率低,受到人的情绪影响比较大 ...
TX2安装pytorch+TensorRT+yolov5实现实时检测
已完成的环境配置: TX2刷机后,完成了opencv4.5.1的编译:Ubuntu18.04安装opencv4.5.1+contrib 支持cuda加速(附带编译好的opencv4.5.1及缺失文件) ...
计算机实时检测系统组成,检测系统
在信号传输通道中,检测系统是指连接输入.输出并具有特定功能的部分.在工程测试实践中,大多数检测系统属于线性时不变系统.检测技术的目的是从测量对象中获取反映其变化规律的有用信息,为了实现此目的,一个广义 ...
MATLAB热障涂层成像,一种涡轮叶片热障涂层模拟试验过程中损伤实时检测方法与流程...
本发明涉及航空发动机防护涂层技术领域,特别涉及一种涡轮叶片热障涂层模拟试验过程中损伤实时检测方法. 背景技术: 现代航空发动机已有两千多个部件应用有先进涂层材料,包括涡轮叶片.密封件.燃烧室密封件动件 ...

防火墙Deny记录实时检测