🐍 Python으로 IP·도메인 평판 조회 자동화하기 — VirusTotal API 실전 활용

pip install requests python-dotenv

vt-checker/
├── .env           ← API 키 저장 (절대 커밋 금지)
├── checker.py     ← 메인 스크립트
└── .gitignore

VT_API_KEY=your_virustotal_api_key_here
SLACK_WEBHOOK_URL=your_slack_webhook_url_here

GET https://www.virustotal.com/api/v3/{리소스 타입}/{식별자}

headers = {
    "x-apikey": "YOUR_API_KEY"
}

{
  "data": {
    "attributes": {
      "last_analysis_stats": {
        "malicious": 5,
        "suspicious": 1,
        "undetected": 60,
        "harmless": 8,
        "timeout": 0
      }
    }
  }
}

import requests
import os
from dotenv import load_dotenv

load_dotenv()

VT_API_KEY = os.getenv("VT_API_KEY")
BASE_URL = "https://www.virustotal.com/api/v3"
HEADERS = {"x-apikey": VT_API_KEY}


def check_ip(ip: str) -> dict:
    """IP 주소 악성 여부 조회"""
    url = f"{BASE_URL}/ip_addresses/{ip}"
    response = requests.get(url, headers=HEADERS)

    if response.status_code == 200:
        data = response.json()
        stats = data["data"]["attributes"]["last_analysis_stats"]
        return {
            "type": "ip",
            "indicator": ip,
            "malicious": stats.get("malicious", 0),
            "suspicious": stats.get("suspicious", 0),
            "harmless": stats.get("harmless", 0),
            "undetected": stats.get("undetected", 0),
        }
    elif response.status_code == 404:
        return {"type": "ip", "indicator": ip, "error": "Not found"}
    else:
        return {"type": "ip", "indicator": ip, "error": f"HTTP {response.status_code}"}

def check_domain(domain: str) -> dict:
    """도메인 악성 여부 조회"""
    url = f"{BASE_URL}/domains/{domain}"
    return _parse_response(url, "domain", domain)


def check_hash(file_hash: str) -> dict:
    """파일 해시 악성 여부 조회"""
    url = f"{BASE_URL}/files/{file_hash}"
    return _parse_response(url, "hash", file_hash)


def _parse_response(url: str, indicator_type: str, indicator: str) -> dict:
    """공통 응답 파싱"""
    response = requests.get(url, headers=HEADERS)

    if response.status_code == 200:
        stats = (
            response.json()["data"]["attributes"]["last_analysis_stats"]
        )
        return {
            "type": indicator_type,
            "indicator": indicator,
            "malicious": stats.get("malicious", 0),
            "suspicious": stats.get("suspicious", 0),
            "harmless": stats.get("harmless", 0),
            "undetected": stats.get("undetected", 0),
        }
    elif response.status_code == 404:
        return {"type": indicator_type, "indicator": indicator, "error": "Not found"}
    else:
        return {
            "type": indicator_type,
            "indicator": indicator,
            "error": f"HTTP {response.status_code}",
        }

def judge(result: dict) -> str:
    """
    malicious >= 5  → MALICIOUS (악성)
    malicious >= 1  → SUSPICIOUS (의심)
    malicious == 0  → CLEAN (정상)
    """
    if "error" in result:
        return "UNKNOWN"

    malicious = result.get("malicious", 0)
    suspicious = result.get("suspicious", 0)

    if malicious >= 5:
        return "MALICIOUS"
    if malicious >= 1 or suspicious >= 3:
        return "SUSPICIOUS"
    return "CLEAN"

import json

SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL")

VERDICT_EMOJI = {
    "MALICIOUS":  ":red_circle:",
    "SUSPICIOUS": ":large_yellow_circle:",
    "CLEAN":      ":white_check_mark:",
    "UNKNOWN":    ":white_circle:",
}

VERDICT_COLOR = {
    "MALICIOUS":  "#e74c3c",
    "SUSPICIOUS": "#e67e22",
    "CLEAN":      "#27ae60",
    "UNKNOWN":    "#7f8c8d",
}


def send_slack(result: dict, verdict: str) -> None:
    """Slack 웹훅으로 조회 결과 발송"""
    if not SLACK_WEBHOOK_URL:
        print("[!] SLACK_WEBHOOK_URL 미설정 — Slack 알림 건너뜁니다.")
        return

    emoji = VERDICT_EMOJI.get(verdict, ":white_circle:")
    color = VERDICT_COLOR.get(verdict, "#7f8c8d")
    indicator = result["indicator"]

    if "error" in result:
        text = f"{emoji} *{indicator}* — 조회 실패: {result['error']}"
        fields = []
    else:
        text = f"{emoji} *{indicator}* — {verdict}"
        fields = [
            {"title": "Malicious",  "value": str(result["malicious"]),  "short": True},
            {"title": "Suspicious", "value": str(result["suspicious"]), "short": True},
            {"title": "Harmless",   "value": str(result["harmless"]),   "short": True},
            {"title": "Undetected", "value": str(result["undetected"]), "short": True},
        ]

    payload = {
        "attachments": [
            {
                "color": color,
                "text": text,
                "fields": fields,
                "footer": "VirusTotal API",
            }
        ]
    }

    requests.post(
        SLACK_WEBHOOK_URL,
        data=json.dumps(payload),
        headers={"Content-Type": "application/json"},
    )

import time

def check_and_notify(indicator: str) -> None:
    """타입 자동 감지 후 조회 및 알림"""
    import re

    # 타입 판단
    ip_pattern = re.compile(
        r"^(\d{1,3}\.){3}\d{1,3}$"
    )
    hash_pattern = re.compile(
        r"^[a-fA-F0-9]{32}$|^[a-fA-F0-9]{40}$|^[a-fA-F0-9]{64}$"
    )

    indicator = indicator.strip()

    if ip_pattern.match(indicator):
        result = check_ip(indicator)
    elif hash_pattern.match(indicator):
        result = check_hash(indicator)
    else:
        result = check_domain(indicator)

    verdict = judge(result)
    print(f"[{verdict}] {indicator}")
    send_slack(result, verdict)


def batch_check(indicators: list, delay: float = 15.0) -> None:
    """
    여러 지표 일괄 조회
    delay: 요청 간격 (무료 계정은 분당 4회 = 최소 15초)
    """
    for i, indicator in enumerate(indicators):
        print(f"[{i+1}/{len(indicators)}] {indicator} 조회 중...")
        check_and_notify(indicator)
        if i < len(indicators) - 1:
            time.sleep(delay)


# ── 실행 ──────────────────────────────────────────────────────
if __name__ == "__main__":
    targets = [
        "8.8.8.8",
        "185.220.101.45",          # 알려진 Tor exit node
        "malware-test.com",
        "44d88612fea8a8f36de82e1278abb02f",  # EICAR 테스트 해시 (MD5)
    ]
    batch_check(targets)

[1/4] 8.8.8.8 조회 중...
[CLEAN] 8.8.8.8

[2/4] 185.220.101.45 조회 중...
[MALICIOUS] 185.220.101.45

[3/4] malware-test.com 조회 중...
[SUSPICIOUS] malware-test.com

[4/4] 44d88612fea8a8f36de82e1278abb02f 조회 중...
[MALICIOUS] 44d88612fea8a8f36de82e1278abb02f

def safe_request(url: str, headers: dict, retries: int = 3) -> requests.Response:
    """429 발생 시 자동 재시도"""
    for attempt in range(retries):
        response = requests.get(url, headers=headers)
        if response.status_code == 429:
            wait = 60 * (attempt + 1)
            print(f"[!] Rate limit 도달. {wait}초 대기 후 재시도...")
            time.sleep(wait)
        else:
            return response
    return response

with open("indicators.txt", "r") as f:
    targets = [line.strip() for line in f if line.strip()]
batch_check(targets)

import csv
import datetime

def save_to_csv(results: list, filename: str = None) -> None:
    if not filename:
        today = datetime.date.today().isoformat()
        filename = f"vt_results_{today}.csv"

    with open(filename, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(
            f,
            fieldnames=["indicator", "type", "verdict",
                        "malicious", "suspicious", "harmless", "undetected"]
        )
        writer.writeheader()
        writer.writerows(results)
    print(f"[*] 결과 저장: {filename}")

# .gitignore
.env
*.env