利用python检测远程 IP和端口是否可连接并钉钉报警

利用python检测远程 IP和端口是否可连接 并dingtalk 报警 ==

code

cat  /home/evan/scr/chk-port.py 
import socket 
import yaml 
import requests 
from datetime import datetime 

DING_WEBHOOK ="https://oapi.dingtalk.com/robot/send?access_token=4578bcd63ad05f3a315ef4f971c7defe9990ef0adc47f804c3273371a7355555"

def load_from_yaml(file_path):
    with open(file_path,'r') as f:
        data = yaml.safe_load(f)
        return data.get("targets",[])

def check_port(host,port,timeout=4):
    try:
        with socket.create_connection((host,port),timeout=timeout):
            return True
    except:
        return False
    
def send_ding_alert(message):
    headers = {'Content-Type': 'application/json;charset=utf-8'}
    data = {
        "msgtype": "text",
        "text": {
            "content": message
        }
    }
    try:
        requests.post(DING_WEBHOOK, headers=headers, json=data,timeout=4)
    except Exception as e:
        print(f"Failed to send dingtalk alert: {str(e)}")

def main():
    ips = load_from_yaml("/home/evan/scr/ips.yaml")
    down_list = []

    for item in ips:
        host = item["host"]
        port = item["port"]
        if not check_port(host,port):
            down_list.append(f"{host}:{port}")

    if down_list:
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        msg = f"[{now}] Port detection anomaly端口检测异常:\n" + "\n".join(down_list)
        #print(msg)
        #print(down_list)
        send_ding_alert(msg)
    else:
        print("All ports are up.")
        send_ding_alert("All ports are up.")

if __name__== "__main__":
    main()
    #send_ding_alert("这是测试消息")

cat ips.yaml 
targets:
  - host: 45.192.18.733
    port: 7100
  - host: 45.192.18.733
    port: 7101
  - host: 45.192.18.733
    port: 7148
  - host: 45.192.18.733
    port: 7149

Usage

*/4  * * * * /usr/bin/python3 /home/evan/scr/chk-port.py >> /home/evan/port_checker.log 2>&1