原來一直用shell結合nmap做端口監控，最近剛好有時間改用python重寫。

監控效果：

mysql數據庫用于讀取IP地址，輸出IP詳細信息，記錄故障時間，send記錄是否發生變量。

#?-*-?coding:?utf-8?-*-

#?@Time?:?2020-4-10?22:13

#?@Author?:?yejunhai

#?@Site?:

#?@File?:?port_monitor.py

#?@Software:?PyCharm

import?pymysql

import?socket

import?sys

import?time

import?requests

import?json

def?msg(text)?:

#發送到企業微信機器人

headers?=?{'Content-Type':?'application/json;charset=utf-8'}

api_url?=?""??#?這個是企業微信機器人生成的webhook地址，修改為你的即可。

json_text?=?{

"msgtype"?:?"text",

"text"?:?{

"content"?:?text

},

}

requests.post(api_url,?json.dumps(json_text),?headers=headers).content

def?port_check(ip,port):

#檢查socket返回值

s?=?socket.socket(socket.AF_INET,socket.SOCK_STREAM)

s.settimeout(2)

result=s.connect_ex((ip,int(port)))

return?result

def?out_log(text):

#運行寫日志。。。。。。。

with?open(f'{sys.argv[0].split(".")[0]}.log','a')?as?f:

print(text,file=f)

def?down_time(ip):

#計算故障時間，mysql自帶也可以計算，不會-?-！

cursor.execute(f"SELECT?mzt.start_time,mzt.end_time?FROM?mzt?WHERE?mzt.ip?=?'{ip}'")

total_time?=?cursor.fetchone()

try:

start_time?=?total_time[0]

end_time?=total_time[1]

duration?=?end_time-start_time

return?f"\n本次故障開始時間?{start_time}\n本次故障結束時間?{end_time}\n本次故障持續時間?{duration}"

except:

return?"\n故障開始時間未記錄"

#時間格式

cur_time=time.strftime("%Y-%m-%d?%H:%M:%S",time.localtime())

#?打開數據庫連接

db?=?pymysql.connect("127.0.0.1",user?=?"root",passwd?=?"root",db?=?"zwy")

#?使用?cursor()?方法創建一個游標對象?cursor

cursor?=?db.cursor()

#?SQL?查詢語句

sql?=?"SELECT?mzt.ip,mzt.port1,mzt.port2,mzt.send,mzt.`部署業務`?FROM?mzt?"

#?try:

cursor.execute(sql)

results?=?cursor.fetchall()

#獲取到IP和端口開始搞事

for?row?in?results:

ip=row[0]

port1=row[1]

port2=row[2]

send=row[3]

description=row[4]

#檢查多個端口

for?port?in?port1,port2:

if?port?!=?''?and?port?!=?None:

port_status=port_check(ip,port)

if?port_status?!=?0?and?send?==?0:

cursor.execute(f"UPDATE?mzt?SET?send?=?'1'?WHERE?mzt.ip?=?'{ip}'")?#發生告警后更新防止一直告警

cursor.execute(f"UPDATE?mzt?SET?start_time?=?'{cur_time}'?WHERE?mzt.ip?=?'{ip}'")?#記錄故障時間

db.commit()

msg(f"{cur_time}?{description}?{ip}:{port}?端口關閉，請檢查!")?#發送告警

out_log(f"{cur_time}?{ip}:{port}?check?{port_status}?send:?{send}")?#寫入日志

elif?port_status?==?0?and?send?==?1:

cursor.execute(f"UPDATE?mzt?SET?send?=?'0'?WHERE?mzt.ip?=?'{ip}'")

cursor.execute(f"UPDATE?mzt?SET?end_time?=?'{cur_time}'?WHERE?mzt.ip?=?'{ip}'")

db.commit()

msg(f"{cur_time}?{description}?{ip}:{port}?端口恢復.{down_time(ip)}")

out_log(f"{cur_time}?{ip}:{port}?check?{port_status}?send:?{send}")

else:

out_log(f"{cur_time}?{ip}:{port}?check?{port_status}?send:?{send}")

#?except:

#?????print("Error:?unable?to?fetch?data")

#關閉數據庫

db.close()

#老傳統crontab定時的跑就完事了

*/1 * * * * /usr/bin/python3 /root/port_monitor.py

與50位技術專家面對面20年技術見證，附贈技術全景圖

總結

以上是生活随笔為你收集整理的python监控端口_python3 端口监控的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。