使用py批量检测云服务器访问ip是否为恶意ip

日明月

由于工作需要每天要封堵云上攻击ip，因此编写此脚本举行检测，需要在同一目录下创建url.txt放置待检测ip，apikey需要到https://x.threatbook.com/举行申请，一天调用上限50条。

• import requests,re,os
  
• city_info=[]
  
• arr1=[]
  
• def check_ip(ip):
  
•     try:
  
•         url = "https://api.threatbook.cn/v3/scene/ip_reputation"
  
•         query = {
  
•             "apikey": "",
  
•             "resource": f"{ip}",
  
•             "lang": "zh"
  
•         }
  
•         response = requests.request("GET", url, params=query)
  
•         res=response.json()
  
•         f = open('attack_ip.txt', mode='a+')
  
•         if(response.json()["response_code"]==-4):
  
•             url = "https://api.threatbook.cn/v3/scene/ip_reputation"
  
•             query = {
  
•                 "apikey": "",
  
•                 "resource": f"{ip}",
  
•                 "lang": "zh"
  
•             }
  
•             response1 = requests.request("GET", url, params=query)
  
•             res1 = response1.json()
  
•             if(res1["response_code"]!=-4):
  
•                 ip_info1 = res1['data'][ip]
  
•                 is_malicious = ip_info1['is_malicious']
  
•                 # print(res1['data'])
  
•                 if is_malicious:
  
•                     city = ip_info1['basic']['location']['country'] + ' ' + ip_info1['basic']['location']['province'] + ' ' + \
  
•                            ip_info1['basic']['location']['city']
  
•                     city_info.append(ip_info1['basic']['location']['country'])
  
•                     print(ip)
  
•                     f.write(ip + ' ' + city + '\n')
  
•             else:
  
•                 print("API接口上限")
  
•                 exit()
  
•         else:
  
•             ip_info = res['data'][ip]
  
•             is_malicious = ip_info['is_malicious']
  
•             # print(res)
  
•             if is_malicious:
  
•                 city = ip_info['basic']['location']['country'] + ' ' + ip_info['basic']['location']['province'] + ' ' + \
  
•                        ip_info['basic']['location']['city']
  
•                 city_info.append(ip_info['basic']['location']['country'])
  
•                 # print(ip + ' ' + city)
  
•                 # print("-------------------------")
  
•                 print(ip)
  
•                 f.write(ip + ' ' + city + '\n')
  
•             # pass
  
•         f.close()
  
•     except Exception as e:
  
•         print(e)
  
• 
  
• 
  
• 
  
• 
  
• 
  
• if __name__ == '__main__':
  
•     arr = []
  
•     city_count=[]
  
•     with open("attack_ip.txt", 'r+') as file:
  
•         file.truncate(0)
  
•     for ip in open('url.txt', encoding='utf-8'):
  
•         result = re.match(r"\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b", ip.strip())
  
•         if result is None:  # 进一步确定错误范例
  
•           continue
  
•           # print("match()的返回值为空")
  
•         arr.append(result.group())
  
•     a=list(set(arr))
  
•     chose = input('输入选项：1、去重ip 2、威胁识别\n')
  
•     if chose == '1':
  
•         for i in a:
  
•             print(i)
  
•     elif chose == '2':
  
•         for x in a:
  
•             check_ip(x)
  
•         for x in open('attack_ip.txt'):
  
•             res = re.findall(r"^\S*\s(\S*)", x)
  
• 
  
•             arr1.append(res[0])
  
•             # print(res)
  
•         f = open('attack_ip.txt', mode='a+')
  
•         for i in set(arr1):
  
•             # count函数某一个字符在列表中的出现次数
  
• 
  
•             f.write(f"{i} {arr1.count(i)}次" + '\n')
  
•             print(f"{i} {arr1.count(i)}次")
  
•         # print(len(set(arr1)))
  
•         f.close()
  
•     else:
  
•         exit()