域名监控

#!/usr/bin/env python3
# # -*- coding: utf-8 -*-
#
# # @Time : 2020/10/30 11:21
# # @Author : admin
# # @Software: PyCharm
# # @Desc:  域名检测配置文件

# 需要检测的域名列表
# 格式说明： 域名 角色[main|backup] 所属类别[QP|DY]
DOMAIN_LIST = [
    'game.test01.com main QP',
    'game.test02.com main QP',
    'game.test03.com main QP',
]

# PROD环境
game.test.com
gateway.test.com
openapi.test.com

## 域名有效期与域名备案状态检测

<br>

### 1. 项目说明：
+ 域名有效期检测：  
a. 使用 whois 命令对 .com 域名有效期及提供商信息进行查询（whois 只支持 .com 域名或*域名如：cn,us 等）  
b. 非 .com 域名使用 whoissoft.com 网站来解析查询  
c. 使用 domain_validity_check.sh 脚本对 domain.list 中的域名进行检测  
d. 在 domain_validity_check.sh 脚本中调用 send_msg_to_mango.py 发送信息到mango群组  
e. 域名有效期均大于30天时会发送一条域名有效期正常的信息到mango群组(可以变向检测脚本是否正常执行，mango接口是否正常)  
f. 域名有效期有小于30天的会发送一条告警信息到mango群组(带有域名有效期信息与提供商信息)  
g. 如有新增域名请将新域名写入 domain.list 文件  

+ 域名备案状态检测  
a. 通过模拟请求 icp.chinaz.com 站点，并解析返回数据,检索返回数据中的备案状态相关的关键字,判定域名备案是否正常  
b. 备案异常则发送告警到mango群  
c. 所有域名备案都正常则在指定时间段内发送备案正常的通知到mango群
d. 需要检测的域名写入到 config.py 配置中

<br>

### 2. 环境依赖：
    python3.8
    requests
    whois
    curl

<br>

### 3. 部署与使用
a. 安装依赖  
`pip3.8 install -r requirements.txt`  
`yum install whois curl -y`

b. 运行脚本  
+ 域名有效期:  
`bash domain_validity_check.sh domain.list`  

+ 域名备案状态:  
`python domain_icp_check.py` 

<br>

### 4. 设置定时任务  
`crontab -l`  

    # 检查域名有效期(每天10点10分运行)  
    10 10 * * * cd /scritps/path/to && bash domain_validity_check.sh domain.list &>>cron.log  
    
    # 检查域名备案状态(每5分钟运行一次)  
    */5 * * * * cd /scritps/path/to && python icp_check.py &>>cron.log

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# @Time : 2020/10/18 2:35
# @Author : admin
# @Software: PyCharm
# @Desc: 发送信息到mango群组


import requests
import json
import datetime
import sys


def sendmsg2mango(group, msg):
    """
    使用mango api发送信息到相关群组
    :param group: 芒果帐号所在群组(需要在芒果后台查看)
    :param msg: 要发送的信息
    """
    # 群组ID
    room_id_map = {
        "test": "codm_1080008429",  # TEST-Alert
        "prod": "codm_1080008138",  # PROD-Alert
        "uat": "codm_1080008139",   # UAT-Alert
        "net": "codm_1080008430",   # Network-Alert
        "hd": "codm_1080008431",    # Hardware-Alert
        "jenkins": "codm_1080008432"    # JenkinsCICD
    }

    # 机器人帐号ID
    USER_ID = "cod_3444444"
    # token (需要找开发公司申请)
    MANGO_TOKEN = "444444444444333"

    # mango发送消息api
    mango_api_url = 'https://imapi.rezffb.com/plugins/xhcodrestapi/v1/apiservice/{}:{}/v2/sendmessage'.format(USER_ID, MANGO_TOKEN)

    # headers
    headers = {
        "Authorization": "e43432fdvfbzd",
        "Content-Type": "application/json; charset=UTF-8",
    }

    # body
    post_data = {
        "roomname": room_id_map.get(group),  # 群组ID type: string
        "text": msg  # 要发送的消息体 type: string
    }

    try:
        response = requests.post(mango_api_url, headers=headers, data=json.dumps(post_data))
        response_data = json.loads(response.text)
    except Exception as e:
        response_data = { "success": False, "data": e }
    finally:
        return response_data


if __name__ == "__main__":
    if len(sys.argv) == 3:
        group = sys.argv[1]
        msg = sys.argv[2]
        result_data = sendmsg2mango(group, msg)
        #print(result_data)
        if result_data.get('success'):
            print('\n>>> Send to mango [test] group ok\n')
            for k,v in result_data.get('data').items():
                print("{}: {}".format(k,v))
        else:
            print(result_data.get('data'))
    else:
        print('Usage: {} 群组[prod,uat,net,hd,test] MSG'.format(sys.argv[0]))

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# @Time : 2020/11/20 11:25
# @Author : 111
# @Software: PyCharm
# @Desc:  检测域名备案信息


import requests
import re
import time
import random
import logging
import datetime
from functools import reduce
from logging.handlers import TimedRotatingFileHandler
import config
import send_msg_to_mango


def check_domain_icp(_domain):
    """
    获取域名备案状态信息页，截状态关键字，判断备案是否正常
    :param _domain: 要检测的域名 www.baidu.com
    :return: result
    """
    global check_result
    check_domain_url = "{}/{}".format(ICP_CHECK_API, _domain.split('.', 1)[1])  # 检测域名只需要*域名，要把二级域名截掉
    flag_str1 = "未备案或备案取消"   # 有此字串，表示没有备案
    flag_str2 = "网站备案/许可证号"  # 有些字串，并没有flag_str1表示有备案
    custom_ua = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36 Edg/86.0.622.51"
    headers = {'User-Agent': custom_ua}
    print('>>> start check {} ...'.format(_domain))
    try:
        r = requests.get(check_domain_url, headers=headers, timeout=10)
        if r.status_code == 200:
            check_data = r.text
            find_str1 = re.findall(flag_str1, check_data)
            find_str2 = re.findall(flag_str2, check_data)
            if not find_str1 and find_str2:
                check_result[_domain] = 0  # 请求成功，并有备案设置值为0
            else:
                check_result[_domain] = 1  # 请求成功，但没备案值设为1
    except Exception as e:
        check_result[_domain] = 2  # 请求失败设置值为2
        print(e)


def is_show_time(start_time, end_time):
    """
    检测当前时间是否在某个时间段内
    所有域名备案正常的信息只在设定时间段内发送
    """
    # 开始时间
    start_time = datetime.datetime.strptime(str(datetime.datetime.now().date()) + start_time, '%Y-%m-%d%H:%M')
    # 结束时间
    end_time = datetime.datetime.strptime(str(datetime.datetime.now().date()) + end_time, '%Y-%m-%d%H:%M')
    # 当前时间
    now_time = datetime.datetime.now()
    if start_time < now_time < end_time:
        return True
    else:
        return False


def main():
    send_ret = "No needs send msg."
    # 检测所有域名
    for dm_info in config.DOMAIN_LIST:
        domain, domain_role, domain_type = dm_info.split(' ')
        check_domain_icp(domain)
        if check_result.get(domain) == 1:
            msg = "						





						原文地址：https://www.cnblogs.com/zhuhaofeng/p/14163324.html