Wazuh에 Threat Intelligence 연동

VirusTotal

Target-EC2(에이전트)

sudo mkdir -p /opt/vt-test        # 감시할 폴더 생성

sudo vi /var/ossec/etc/ossec.conf # 설정 파일 

#설정 추가
<directories check_all="yes" realtime="yes">/opt/vt-test</directories>

# directories 감시할 폴더 지정
# /opt/vt-test 감시 대상 폴더
# chekck_all = "yes" 파일의 해시, 크기, 권한, 소유자 등 전부 체크
# realtime="yes"     주기적으로 훑는 게 아니라 변화 즉시 탐지 

sudo systemctl restart wazuh-agent # 설정 파일 변경했으니 agent 재시작 

1. FIM이 /opt/vt-test에 새 파일 생긴 걸 탐지 한다.

2. Wazuh Manager 에 “파일 추가됨” 알림

3. 그 해시를 VirousTotal에 조회

4. 악성이면 알림

Wazuh Manager-EC2

sudo vi /var/ossec/etc/ossec.conf # 설정 파일

<integration>
  <name>virustotal</name>
  <api_key>API키</api_key>
  <rule_id>550,554</rule_id>
  <alert_format>json</alert_format>
</integration>     #rule_id 554= 파일 새로 생김, 550 = 파일 내용 바뀜 -> 이 두 알림이 뜰 때만 VirusTotal 조회 


sudo systemctl restart wazuh-manager # 설정 파일 변경했으니 재시작

EICAR 파일 넣기

echo 'X5O!P%@AP[4\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*' | sudo tee /opt/vt-test/eicar.txt

로그 확인하기

sudo grep -i virustotal /var/ossec/logs/alerts/alerts.json | tail -n 5
# alerts에 쌓인 로그 보기 

VirusTotalVirusTotal

VirusTotal

VirusTotal

→ VirusTotal에서 더 자세하게 확인할 수 있다.

AbuseIPDB

AbuseIPDBAbuseIPDB

AbuseIPDB

Check the abuse confidence score, reports, and geolocation of any IP address.

AbuseIPDB

→ AbuseIPDB에서 가입하고 API 키를 가져온다.(무료는 하루 1000회 조회 가능)

Wazuh Manager 연동 스크립트 작성

sudo vi /var/ossec/integrations/abuseipdb.py

import sys, json, ipaddress
from socket import socket, AF_UNIX, SOCK_DGRAM
import requests

SOCKET_ADDR = "/var/ossec/queue/sockets/queue"
THRESHOLD = 50  # 이 점수(0~100) 이상이면 악성으로 표시

def send_event(msg, agent): # 정리한 결과를 Wazuh로 보내는 함수 
    body = json.dumps(msg)
    if not agent or agent.get("id") == "000":
        line = f"1:abuseipdb:{body}"
    else:
        line = f"1:[{agent['id']}] ({agent['name']}) {agent.get('ip', 'any')}->abuseipdb:{body}"
    sock = socket(AF_UNIX, SOCK_DGRAM)
    sock.connect(SOCKET_ADDR)
    sock.send(line.encode())
    sock.close()

def main():     #알림 파일 읽기           
    alert_file, api_key = sys.argv[1], sys.argv[2] 
    with open(alert_file) as f:                      
        alert = json.load(f)   
# Wazuh가 스크립트를 실행할 때 알림 내용이 담긴 파일 경로, ossec.conf에 적은 API 키를 넘겨 준다.

    srcip = alert.get("data", {}).get("srcip") # 공격자 IP 꺼내기
    if not srcip: # srcip -> source ip
        return
    try:
        ip = ipaddress.ip_address(srcip)
    except ValueError:
        return
    if ip.is_private or ip.is_loopback:   # 사설 IP는 조회 의미 없음
        return

    r = requests.get(
        "https://api.abuseipdb.com/api/v2/check",
        headers={"Key": api_key, "Accept": "application/json"},
        params={"ipAddress": srcip, "maxAgeInDays": 90},
        timeout=10,
    )
    r.raise_for_status()
    d = r.json()["data"]
    # AbuseIPDB에 접속을 시도한 IP가 90일 동안 신고되었는지 확인 
    # API 키로 인증, 10초의 시간, 받은 답을 변수 d에 담는다. 

    msg = { # 결과 정리해서 Wazuh로 보내기
        "integration": "abuseipdb",
        "abuseipdb": {
            "ip": srcip,
            "score": str(d["abuseConfidenceScore"]),
            "reports": str(d["totalReports"]),
            "country": d.get("countryCode") or "",
            "isp": d.get("isp") or "",
            "malicious": "1" if d["abuseConfidenceScore"] >= THRESHOLD else "0",
            "source": {
                "alert_id": alert.get("id"),
                "rule": alert.get("rule", {}).get("id"),
                "description": alert.get("rule", {}).get("description"),
            },
        },
    }
    send_event(msg, alert.get("agent"))

if __name__ == "__main__":
    main()

실행 래퍼 + 권한 Wazuh Manager

cd /var/ossec/integrations
cp virustotal custom-abuseipdb # VirusTotal 래퍼를 복사해서 새 이름으로 만든다. 
chown root:wazuh custom-abuseipdb custom-abuseipdb.py # 파일 소유자를 root 그룹을 wazuh로 변경한다.
chmod 750 custom-abuseipdb custom-abuseipdb.py # 파일 권한 설정 

ossec.conf 파일에 연동 설정 Wazuh Manager

<integration>
  <name>custom-abuseipdb</name>
  <api_key>AbuseIPDB API키</api_key>
  <rule_id>5710,5712,5716,5760</rule_id>
  <alert_format>json</alert_format>
</integration>

결과 판별 룰 Wazuh Manager

sudo vi /var/ossec/etc/rules/local_rules.xml

<group name="abuseipdb,">
  <rule id="100100" level="3">
    <decoded_as>json</decoded_as>
    <field name="integration">abuseipdb</field>
    <description>AbuseIPDB: $(abuseipdb.ip) 조회 (점수 $(abuseipdb.score))</description>
  </rule>

  <rule id="100101" level="10">
    <if_sid>100100</if_sid>
    <field name="abuseipdb.malicious">^1$</field>
    <description>AbuseIPDB: 악성 IP $(abuseipdb.ip) - 신뢰도 $(abuseipdb.score)%, 신고 $(abuseipdb.reports)건</description>
  </rule>
</group>

• 100100은 조회 결과가 오면 기본 알림

• 100101은 그중 점수가 50 이상이면 악성 IP 알림

테스트 Agent EC2

sudo touch /var/log/test-ssh.log

sudo vi /var/ossec/etc/ossec.conf

<localfile>
  <log_format>syslog</log_format>
  <location>/var/log/test-ssh.log</location>
</localfile>   # 가짜 로그 파일 읽으라고 설정 하기 

sudo systemctl restart wazuh-agent

 echo "$(LC_ALL=C date '+%b %d %H:%M:%S') $(hostname) sshd[12345]: Failed password for invalid user admin from 172.237.137.146 port 51234 ssh2" | sudo tee -a /var/log/test-ssh.log

{
  "_index": "wazuh-alerts-4.x-2026.09.27",
  "_id": "t9jK46ABTr43n58PHlEP",
  "_version": 1,
  "_score": null,
  "_source": {
    "input": {
      "type": "log"
    },
    "agent": {
      "ip": "10.0.28.82",
      "name": "ip-10-0-28-82.ap-northeast-2.compute.internal",
      "id": "001"
    },
    "manager": {
      "name": "ip-10-0-12-74.ap-northeast-2.compute.internal"
    },
    "data": {
      "abuseipdb": {
        "score": "100",
        "reports": "1621",
        "country": "US",
        "malicious": "1",
        "ip": "172.237.137.146",
        "isp": "Linode",
        "source": {
          "alert_id": "1790528060.2794293",
          "rule": "5710",
          "description": "sshd: Attempt to login using a non-existent user"
        }
      },
      "integration": "abuseipdb"
    },
    "rule": {
      "firedtimes": 1,
      "mail": false,
      "level": 3,
      "description": "AWS CloudWatch log event received",
      "groups": [
        "aws_cloudwatch"
      ],
      "id": "100010"
    },
    "location": "abuseipdb",
    "decoder": {
      "name": "json"
    },
    "id": "1790528061.2794911",
    "full_log": "{\"integration\": \"abuseipdb\", \"abuseipdb\": {\"ip\": \"172.237.137.146\", \"score\": \"100\", \"reports\": \"1621\", \"country\": \"US\", \"isp\": \"Linode\", \"malicious\": \"1\", \"source\": {\"alert_id\": \"1790528060.2794293\", \"rule\": \"5710\", \"description\": \"sshd: Attempt to login using a non-existent user\"}}}",
    "timestamp": "2026-09-27T16:54:21.412+0000"
  },
  "fields": {
    "timestamp": [
      "2026-09-27T16:54:21.412Z"
    ]
  },
  "highlight": {
    "manager.name": [
      "@opensearch-dashboards-highlighted-field@ip-10-0-12-74.ap-northeast-2.compute.internal@/opensearch-dashboards-highlighted-field@"
    ]
  },
  "sort": [
    1790528061412
  ]
}

echo 'X5O!P%@AP[4\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*' | sudo tee /etc/test.txt

import json, sys, time, ipaddress, socket, requests

ALERT_FILE, API_KEY = sys.argv[1], sys.argv[2]
SOCK = "/var/ossec/queue/sockets/queue"
CACHE = "/var/ossec/tmp/abuseipdb_cache.json"
LOG_FILE = "/var/ossec/logs/integrations.log"
TTL = 6 * 3600   # 같은 IP는 6시간 재조회 안 함

# 알림에서 IP를 찾을 경로 (실제 알림 JSON 기준)
IP_PATHS = [
    ("data", "srcip"),                                                                                 # SSH 등
    ("data", "aws", "httpRequest", "clientIp"),                                                        # WAF
    ("data", "aws", "sourceIPAddress"),                                                                # CloudTrail
    ("data", "aws", "service", "action", "awsApiCallAction", "remoteIpDetails", "ipAddressV4"),        # GuardDuty API 호출
    ("data", "aws", "service", "action", "networkConnectionAction", "remoteIpDetails", "ipAddressV4"), # GuardDuty 네트워크
]

def log(msg):
    try:
        with open(LOG_FILE, "a") as f:
            f.write("{} custom-abuseipdb: {}\n".format(time.strftime("%Y-%m-%d %H:%M:%S"), msg))
    except Exception:
        pass

def dig(d, path):
    for k in path:
        if not isinstance(d, dict) or k not in d:
            return None
        d = d[k]
    return d

def send(msg, agent):
    if not agent or agent.get("id") == "000":
        s = "1:abuseipdb:{}".format(json.dumps(msg))
    else:
        loc = "[{}] ({}) {}".format(agent["id"], agent["name"], agent.get("ip", "any"))
        loc = loc.replace("|", "||").replace(":", "|:")
        s = "1:{}->abuseipdb:{}".format(loc, json.dumps(msg))
    with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as c:
        c.connect(SOCK)
        c.send(s.encode())

def main():
    alert = json.load(open(ALERT_FILE))
    ip = next((v for p in IP_PATHS if (v := dig(alert, p))), None)
    if not ip:
        return
    try:
        if not ipaddress.ip_address(ip).is_global:
            return
    except ValueError:
        return

    try:
        cache = json.load(open(CACHE))
    except Exception:
        cache = {}
    hit = cache.get(ip)
    if hit and time.time() - hit["t"] < TTL:
        return

    try:
        r = requests.get(
            "https://api.abuseipdb.com/api/v2/check",
            headers={"Key": API_KEY, "Accept": "application/json"},
            params={"ipAddress": ip, "maxAgeInDays": 90},
            timeout=10,
        )
    except requests.RequestException as e:
        log("request failed for {}: {}".format(ip, e))
        return
    if r.status_code != 200:
        log("AbuseIPDB error {} for {}: {}".format(r.status_code, ip, r.text[:200]))
        return
    d = r.json()["data"]
    log("checked {} score={}".format(ip, d["abuseConfidenceScore"]))

    cache[ip] = {"t": time.time()}
    try:
        json.dump(cache, open(CACHE, "w"))
    except Exception as e:
        log("cache write failed: {}".format(e))

    send({
        "integration": "abuseipdb",
        "abuseipdb": {
            "ip": ip,
            "score": d["abuseConfidenceScore"],
            "reports": d["totalReports"],
            "country": d.get("countryCode"),
            "isp": d.get("isp"),
            "source_rule": alert.get("rule", {}).get("id"),
        },
    }, alert.get("agent"))

if __name__ == "__main__":
    main()

import json, sys, time, ipaddress, socket, requests
ALERT_FILE, API_KEY = sys.argv[1], sys.argv[2]

• json은 알림 파일을 읽고 결과를 만드는 데

• ipaddress는 사설 IP를 거르는 데

• socket은 Wazuh에 결과를 돌려보내는 데, requests는 AbuseIPDB를 호출하는데 사용한다.

• time은 캐시 만료 계산용

• sys.argv는 스크립트를 실행할 때 붙는 인자이다. Wazuh가 custom-abuseipdb <알림파일경로> <api_key>… 처럼 실행된다. 1번째가 알림 파일, 2번째가 ossec.conf에 적은 API 키이다.

SOCK = "/var/ossec/queue/sockets/queue"
CACHE = "/var/ossec/tmp/abuseipdb_cache.json"
LOG_FILE = "/var/ossec/logs/integrations.log"
TTL = 6 * 3600

• SOCK은 매니저가 이벤트를 받는 입구이다. 여기로 문자열을 보내면 Wazuh가 새 이벤트로 처리한다.

• CACHE는 “이 IP를 언제 조회했는지” 적어두는 메모 파일이다.

• LOG_FILE 은 Wazuh 연동 로그 파일이다. VT와 Discord 연동도 여기에 기록한다. 세 연동의 로그를 한 곳에서 볼 수 있다.

• TTL은 그 메모의 유효 시간(6시간, 초 단위)이다.

IP_PATHS = [
    ("data", "srcip"),                                                                                 # SSH 등
    ("data", "aws", "httpRequest", "clientIp"),                                                        # WAF
    ("data", "aws", "sourceIPAddress"),                                                                # CloudTrail
    ("data", "aws", "service", "action", "awsApiCallAction", "remoteIpDetails", "ipAddressV4"),        # GuardDuty API 호출
    ("data", "aws", "service", "action", "networkConnectionAction", "remoteIpDetails", "ipAddressV4"), # GuardDuty 네트워크
]

def log(msg):
    try:
        with open(LOG_FILE, "a") as f:
            f.write("{} custom-abuseipdb: {}\n".format(time.strftime("%Y-%m-%d %H:%M:%S"), msg))
    except Exception:
        pass

• log(”메시지”) 를 부르면 2026-10-07 00:00:00 custom-abuseipdb: 메시지 형태로 한 줄이 추가된다. “a” 는 이어쓰기 모드이다.

• try/except: pass 는 로그를 쓰다 실패해도 스크립트 본 작업은 계속 가게 한다. 로그 때문에 AbuseIPDB 조회가 죽지 않기 위해서

def dig(d, path):
    for k in path:
        if not isinstance(d, dict) or k not in d:
            return None
        d = d[k]
    return d

• data를 따라가는 함수이다.

• 서랍 이름(k)을 하나씩 열어 들어간다.

• 가다가 서랍이 아니거나 그 이름의 서랍이 없으면 바로 None을 리턴한다.

• 끝까지 가면 그 자리의 IP를 리턴한다.

def send(msg, agent):
    if not agent or agent.get("id") == "000":
        s = "1:abuseipdb:{}".format(json.dumps(msg))
    else:
        loc = "[{}] ({}) {}".format(agent["id"], agent["name"], agent.get("ip", "any"))
        loc = loc.replace("|", "||").replace(":", "|:")
        s = "1:{}->abuseipdb:{}".format(loc, json.dumps(msg))
    with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as c:
        c.connect(SOCK)
        c.send(s.encode())

• msg : 보낼 내용 (점수, IP등이 담긴 딕셔너리) 이다.

• agent : 원래 알림이 어느 agent에서 왔는지 정보 이다.

• AF_UNIX는 같은 서버 안에서 파일 경로(SOCK)로 통신하는 방식이다.

• SOCK_DGRAM은 연결 없이 메시지 한 통씩 던지는 방식이다.

• connect로 매니저의 이벤트 입구에 접속하고, send로 문자열을 보낸다. .encode()는 글자를 바이트로 바꾸는 것이다.

• with 블록이 끝나면 소켓이 자동으로 닫힌다.

def main():
    alert = json.load(open(ALERT_FILE))
    ip = next((v for p in IP_PATHS if (v := dig(alert, p))), None)
    if not ip:
        return
    try:
        if not ipaddress.ip_address(ip).is_global:
            return
    except ValueError:
        return

• alert = json.load(open(ALERT_FILE)) Wazuh가 넘겨준 알림 파일을 열어서 파이썬 딕셔너리로 바꾼다.

• ip = next((v for p in IP_PATHS if (v := dig(alert, p))), None) 는 IP_PARTHS 를 위에서부터 하나씩 dig 로 따라가 보고, 값이 나오는 첫 번째를 ip에 담는 부분이다.

try:
    if not ipaddress.ip_address(ip).is_global:
        return
except ValueError:
    return

• ipaddress.ip_address(ip) 는 문자열을 IP 객체로 바꾼다. .is_global 은 인터넷에서 쓰이는 공인 IP인지를 알려준다.

• 사설 IP면 is_global 이 False 라서 조회해도 의미없으니까 종료한다.

• IP가 아닌 문자열이 들어오면 (CloudTrail의 ec2.amazonaws.com 등) ip_address() 가 ValueError를 낸다.

try:
    cache = json.load(open(CACHE))
except Exception:
    cache = {}
hit = cache.get(ip)
if hit and time.time() - hit["t"] < TTL:
    return

try:
        r = requests.get(
            "https://api.abuseipdb.com/api/v2/check",
            headers={"Key": API_KEY, "Accept": "application/json"},
            params={"ipAddress": ip, "maxAgeInDays": 90},
            timeout=10,
        )
    except requests.RequestException as e:
        log("request failed for {}: {}".format(ip, e))
        return
    if r.status_code != 200:
        log("AbuseIPDB error {} for {}: {}".format(r.status_code, ip, r.text[:200]))
        return
    d = r.json()["data"]
    log("checked {} score={}".format(ip, d["abuseConfidenceScore"]))

try:
    cache = json.load(open(CACHE))
except Exception:
    cache = {}

hit = cache.get(ip)
if hit and time.time() - hit["t"] < TTL:
    return

• cache.get(ip) 는 이 IP의 기록을 꺼낸다. 없으면 None

• 기록이 있고, 지금 시각에서 조회 시각을 뺀 값이 TTL (6시간) 보다 작으면 최근에 이미 조회한 것이기 때문에 종료한다.

try:
    r = requests.get(
        "https://api.abuseipdb.com/api/v2/check",
        headers={"Key": API_KEY, "Accept": "application/json"},
        params={"ipAddress": ip, "maxAgeInDays": 90},
        timeout=10,
    )
except requests.RequestException as e:              
    log("request failed for {}: {}".format(ip, e))  
    return   

• 주소는 AbuseIPDB의 chekck API이다.

• headers 의 Key 에 API 키를 넣어서 등록된 사용자임을 확인하고 응답을 JSON으로 받는다.

• params 는 질문 내용이다. ipAddress 는 조회할 IP이다. maxAgeInDays: 90 은 최근 90일 동안의 신고 내용만 확인

• timeout=10 은 10초 안에 응답이 없으면 포기

• DNS 실패, 연결 끊김, 10초 타임아웃이 나면RequestException 을 잡아서 request failed for 1.2.3.4: … 를 integrations.log 에 남기고 return으로 종료

 if r.status_code != 200:
        log("AbuseIPDB error {} for {}: {}".format(r.status_code, ip, r.text[:200]))
        return

• 200은 정상 응답을 뜻한다. 키가 틀렸거나, 한도를 넘었으면 다른 숫자가 와서 조건문 실행, 200이면 조건문 통과

• 정상 응답이면 JSON의 data 부분을 변수 d에 담는다. 안에 점수, 신고횟수, 국가 등이 들어있다.

• 앞 블록은 네트워크 오류에 대한 try/except 이다. 이 블록은 요청은 갔는데 AbuseIPDB가 거절한 경우이다.

d = r.json()["data"]
log("checked {} score={}".format(ip, d["abuseConfidenceScore"]))   # 추가 (성공 로그)

cache[ip] = {"t": time.time()}
try:
    json.dump(cache, open(CACHE, "w"))
except Exception as e:                                             # pass → 로그로 변경
    log("cache write failed: {}".format(e))

• 성공 로그(checked …)

• 조회가 성공하면 check 1.2.3.4 score = 100 처럼 IP와 점수를 남긴다.

• 캐시 쓰기 실패

• 캐시 파일은 /var/ossec/tmp/abuseipdb_cache.json 이다. 권한이 틀리면 저장 실패

send({
        "integration": "abuseipdb",
        "abuseipdb": {
            "ip": ip,
            "score": d["abuseConfidenceScore"],
            "reports": d["totalReports"],
            "country": d.get("countryCode"),
            "isp": d.get("isp"),
            "source_rule": alert.get("rule", {}).get("id"),
        },
    }, alert.get("agent"))

if __name__ == "__main__":
    main()

• 앞에서 AbuseIPDB에서 받은 결과(d)를 JSON으로 만들어 Wazuh에 돌려보낸다.

• 첫 번째 인자가 보낼 내용이고, 두번째 인자 alert.get(”agent”) 는 원래 알림이 온 에이전트 정보이다. send() 안에서 이걸 보고 매니저 자신의 알림인지, 에이전트의 알림인지에 따라 소켓 메시지 형식을 정한다.