1
import requests
2
import json
3

4

5
url = "https://Server_IP/api/v1/attack/ip?api_key=YOUR_API_KEY"
6

7
payload = json.dumps(
8
    {
9
        "start_time": 0,
10
        "end_time": 0,
11
        "intranet": 0,
12
        "threat_label": [
13
            "Scanner",
14
            "Botnet",
15
            "uspicious",
16
            "Mobile",
17
        ],
18
    }
19
)
20
headers = {"Content-Type": "application/json"}
21

22
response = requests.request("POST", url, headers=headers, data=payload, verify=False)
23
print(response.text)

1
import requests
2
import json
3

4

5
url = "https://Server_IP/api/v1/attack/ip?api_key=YOUR_API_KEY"
6

7
payload = json.dumps(
8
    {
9
        "start_time": 0,
10
        "end_time": 0,
11
        "intranet": 0,
12
        "threat_label": [
13
            "IDC",
14
            "Spam",
15
            "Zombie",
16
            "Exploit",
17
            "Edu",
18
            "BTtracker",
19
            "Backbone",
20
        ],
21
    }
22
)
23
headers = {"Content-Type": "application/json"}
24

25
response = requests.request("POST", url, headers=headers, data=payload, verify=False)
26
print(response.text)

1
import requests
2
import json
3
import re
4
import logging
5
import os
6
import warnings
7

8
# 忽略警告
9
warnings.filterwarnings("ignore")
10

11
# 获取当前脚本所在的目录
12
script_dir = os.path.dirname(os.path.realpath(__file__))
13

14
# 设置日志
15
logging.basicConfig(level=logging.INFO, format="%(message)s")
16
logger = logging.getLogger(__name__)
17

18
# 配置选项
19
RESULT_DIR = os.path.join(script_dir, "result")
20

21
# 检查并创建结果目录
22
if not os.path.exists(RESULT_DIR):
23
    os.makedirs(RESULT_DIR)
24

25
BASE_URL = "https://蜜罐的API链接"  # 此处为蜜罐的API
26
HEADERS = {"Content-Type": "application/json"}
27
THREAT_LABELS_ATTACK = [
28
    "IDC",
29
    "Spam",
30
    "Zombie",
31
    "Exploit",
32
    "Edu",
33
    "BTtracker",
34
    "Backbone",
35
]
36
THREAT_LABELS_SCANNER = ["Scanner", "Botnet", "uspicious", "Mobile"]
37
IP_REGEX_PATTERN = r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}"
38

39

40
def fetch_data(payload):
41
    try:
42
        response = requests.post(BASE_URL, headers=HEADERS, data=payload, verify=False)
43
        response.raise_for_status()
44
        return response.text
45
    except requests.exceptions.RequestException as e:
46
        logger.error(f"Error fetching data: {e}")
47
        return ""
48

49

50
def match_ips(data):
51
    return re.findall(IP_REGEX_PATTERN, data)
52

53

54
def find_new_ips(old_ips, new_ips):
55
    return list(set(new_ips) - set(old_ips))
56

57

58
def write_ips_to_file(file_path, ips):
59
    with open(file_path, "w") as file:
60
        file.write("\n".join(ips))
61

62

63
def send_notification(webhook_url, content):
64
    headers = {"Content-Type": "application/json", "Charset": "UTF-8"}
65
    message = {"msg_type": "text", "content": {"text": content}}
66
    response = requests.post(url=webhook_url, headers=headers, data=json.dumps(message))
67

68

69
def main():
70
    attack_webhook = (
71
        "https://open.feishu.cn/open-apis/bot/v2/hook/xxxx"  # 接收攻击告警机器人的 webhook
72
    )
73
    scan_webhook = (
74
        "https://open.feishu.cn/open-apis/bot/v2/hook/xxxxx"  # 接收扫描告警机器人的 webhook
75
    )
76

77
    payload_attacker = json.dumps(
78
        {
79
            "start_time": 0,
80
            "end_time": 0,
81
            "intranet": 0,
82
            "source": 0,
83
            "threat_label": THREAT_LABELS_ATTACK,
84
        }
85
    )
86
    payload_scanner = json.dumps(
87
        {
88
            "start_time": 0,
89
            "end_time": 0,
90
            "intranet": 0,
91
            "source": 0,
92
            "threat_label": THREAT_LABELS_SCANNER,
93
        }
94
    )
95

96
    attack_data = fetch_data(payload_attacker)
97
    scanner_data = fetch_data(payload_scanner)
98

99
    attacker_ips = match_ips(attack_data)
100
    scanner_ips = match_ips(scanner_data)
101

102
    attacker_ips_path = os.path.join(RESULT_DIR, "attack_ips.txt")
103
    scan_ips_path = os.path.join(RESULT_DIR, "scan_ips.txt")
104

105
    previous_attacker_ips = []
106
    previous_scan_ips = []
107

108
    if os.path.exists(attacker_ips_path):
109
        with open(attacker_ips_path, "r") as file:
110
            previous_attacker_ips = file.read().splitlines()
111

112
    if os.path.exists(scan_ips_path):
113
        with open(scan_ips_path, "r") as file:
114
            previous_scan_ips = file.read().splitlines()
115

116
    new_attack_ips = find_new_ips(previous_attacker_ips, attacker_ips)
117
    new_scan_ips = find_new_ips(previous_scan_ips, scanner_ips)
118

119
    write_ips_to_file(attacker_ips_path, attacker_ips)
120

121
    if new_attack_ips:
122
        logger.info("有憨包踩蜜罐咯，尝试下溯源吧! hacking!")
123
        logger.info("\n".join(new_attack_ips))
124
        attack_content = "有憨包踩蜜罐咯，尝试下溯源吧! hacking!\n" + "\n".join(new_attack_ips)
125
    else:
126
        logger.info("还没新的憨包踩到蜜罐! 请静待有缘人")
127
        attack_content = "还没新的憨包踩到蜜罐! 请静待有缘人\n"
128

129
    write_ips_to_file(scan_ips_path, scanner_ips)
130

131
    if new_scan_ips:
132
        logger.info("快把这些扫描器、爬虫封了！")
133
        logger.info("\n".join(new_scan_ips))
134
        scan_content = "快把这些扫描器、爬虫封了！\n" + "\n".join(new_scan_ips)
135
    else:
136
        logger.info("目前没得需要封禁的IP")
137
        scan_content = "目前没得需要封禁的IP\n"
138

139
    send_notification(attack_webhook, attack_content)
140
    send_notification(scan_webhook, scan_content)
141

142

143
if __name__ == "__main__":
144
    main()