abuse_autopilot.py (4326B)
1 import csv
2 import requests
3 from urllib.parse import urlparse
4 import re
5 import socket
6 from socket import getaddrinfo
7 import geoip2.database
8 import datetime
9 import pandas as pd
10 import random
11 from modules.County_Codes import country_code_to_korean
12
13 # 국가 코드를 한국어 국가 이름으로 변환하는 함수
14 def get_korean_country_name(country_code):
15 return country_code_to_korean.get(country_code, '미국')
16
17 def get_country(reader, ip_address):
18 try:
19 response = reader.country(ip_address)
20 return response.country.iso_code
21 except:
22 return 'US'
23
24 # spam_email_list.txt 파일에서 이메일 주소를 읽어옵니다.
25 with open("spam_email_list.txt", "r") as file:
26 email_addresses = file.readlines()
27
28 # 이메일 주소를 랜덤하게 2개 선택합니다.
29 random_emails = random.sample(email_addresses, 2)
30
31 # 다운로드 URL 지정
32 url = "https://urlhaus.abuse.ch/downloads/text/"
33
34 # 파일 다운로드
35 response = requests.get(url)
36 text_data = response.text.splitlines()
37
38 # 텍스트 데이터를 CSV 형식으로 변환
39 reader = csv.reader(text_data)
40
41 # 3열에서 URL 추출 (3열이 없거나 빈 행을 무시)
42 urls = [row[0] for row in reader if len(row) > 0 and row[0]]
43
44 # URL에서 "http://" 또는 "https://"부터 첫 번째 "/"까지의 값을 추출
45 domains = [urlparse(url).netloc.split(':')[0] for url in urls]
46
47 # 중복 값 제거
48 unique_domains = list(set(domains))
49
50 # IP 주소와 도메인 이름 구분
51 ip_regex = re.compile(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}')
52 domain_name_regex = re.compile(r'(?:(?:[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?\.)+[A-Za-z]{2,6})')
53 ips = [domain for domain in unique_domains if ip_regex.fullmatch(domain)]
54 domains = [domain for domain in unique_domains if domain_name_regex.fullmatch(domain)]
55
56 # 중복체크.xlsx 파일과 비교하기 위해 일단 중복 값 제거된 IP 주소와 도메인 이름을 하나의 리스트로 결합
57 check_df = pd.read_excel("중복체크.xlsx", header=None, engine='openpyxl')
58 check_list = check_df.iloc[:, 0].tolist()
59
60 # 중복되지 않은 IP 주소와 도메인 이름 저장
61 unique_ips = [ip for ip in ips if ip not in check_list]
62 unique_domains = [domain for domain in domains if domain not in check_list]
63
64 # 요청된 개수만큼 IP 주소와 도메인 이름 선택
65 selected_ips = unique_ips[:198]
66 selected_domains = unique_domains[:2] if len(unique_domains) >= 2 else unique_domains
67
68 # 선택된 IP 주소와 도메인 이름을 하나의 리스트로 결합
69 final_data = selected_ips + selected_domains
70
71 # GeoLite2 데이터베이스 읽기
72 reader = geoip2.database.Reader('GeoLite2-Country.mmdb')
73
74 # 국가 코드가 KR인 것들을 제거
75 final_data = [domain for domain in final_data if get_country(reader, domain) != "KR"]
76
77 # 오늘 날짜 기반으로 파일명 설정
78 today = datetime.datetime.today().strftime('%y%m%d')
79 filename = f"{today}_악성IP&URL.xlsx"
80
81 # 데이터를 DataFrame으로 변환
82 data = []
83 for i, domain in enumerate(final_data):
84 if ip_regex.fullmatch(domain):
85 country_code = get_country(reader, domain)
86 else:
87 try:
88 ip_address = getaddrinfo(domain, None)[0][4][0]
89 country_code = get_country(reader, ip_address)
90 except (socket.gaierror, IndexError) as e:
91 print(f"Domain resolution error for {domain}: {e}")
92 country_code = 'US' # 기본값 설정
93
94 country_name = get_korean_country_name(country_code)
95 data.append(['' if i < 198 else domain, domain if i < 198 else '', country_name])
96
97 df = pd.DataFrame(data, columns=['URL', '공격 IP', '공격국가'])
98
99 # Excel 파일로 저장
100 df.to_excel(filename, index=False, engine='openpyxl', sheet_name=f"{today}_악성IP&URL")
101
102 # GeoLite2 데이터베이스 닫기
103 reader.close()
104
105 # 중복체크.xlsx 업데이트
106 last_row = check_df.iloc[:, 0].last_valid_index()
107 for index, value in enumerate(final_data):
108 check_df.at[last_row + 1 + index, 0] = value
109
110 for email in random_emails:
111 email = email.strip()
112 if email not in check_list:
113 last_row = check_df.iloc[:, 0].last_valid_index()
114 check_df.at[last_row + 1, 0] = email
115 check_df.at[last_row + 1, 1] = datetime.datetime.today().strftime('%Y-%m-%d')
116
117 check_df.to_excel("중복체크.xlsx", index=False, header=None, engine='openpyxl')