Malpedia를 활용한 악성코드 패밀리 관계 분석은(는) 무엇을 하나요?
개요
Malpedia는 Fraunhofer FKIE가 운영하는 협업 플랫폼으로, 악성코드 패밀리의 별칭, YARA 규칙, 위협 행위자 연관성 및 참조 보고서를 체계적으로 수집합니다. 2,600개 이상의 악성코드 패밀리가 문서화되어 있어, 악성코드 계통 추적, 변종 진화 파악, 특정 위협 그룹과의 연결을 위한 최고의 자료입니다. 이 스킬은 Malpedia API를 쿼리하고, 악성코드 패밀리 간 관계를 매핑하며, 탐지를 위한 YARA 규칙을 추출하고, 위협 행위자가 사용하는 악성코드 생태계에 대한 인텔리전스를 구축하는 방법을 다룹니다.
사용 시기
- 보안 사고를 조사할 때 malpedia를 활용한 악성코드 패밀리 관계 분석이 필요한 경우
- 이 도메인에 대한 탐지 규칙 또는 위협 헌팅 쿼리를 구축할 때
- SOC 분석가가 해당 분석 유형에 대한 구조화된 절차가 필요할 때
- 관련 공격 기법에 대한 보안 모니터링 범위를 검증할 때
사전 요구사항
- Python 3.9+ 및
requests,yara-python,stix2라이브러리 - Malpedia API 키 (https://malpedia.caad.fkie.fraunhofer.de/ 에서 등록)
- 악성코드 분류 및 명명 규칙에 대한 이해
- 탐지를 위한 YARA 규칙 문법에 대한 familiarity
- 검증을 위한 악성코드 샘플 접근 (선택 사항)
주요 개념
Malpedia 데이터 모델
Malpedia는 악성코드를 Families(예: "win.cobalt_strike")로 구성하며, 각 패밀리는 다음을 포함합니다: 별칭(벤더별 이름: "Beacon", "CobaltStrike"), YARA 규칙(커뮤니티 및 벤더 제공), 행위자 연관성(해당 패밀리를 사용하는 위협 그룹), 참조 보고서(패밀리를 문서화한 CTI 보고서), 샘플 해시(각 변종을 대표하는 샘플).
악성코드 패밀리 명명
Malpedia는 platform.family_name 형식을 사용합니다(예: win.emotet, elf.mirai, apk.flubot). 플랫폼에는 win(Windows), elf(Linux), apk(Android), osx(macOS), py(Python)가 있습니다. 이 표준화된 명명은 서로 다른 벤더가 동일한 악성코드에 서로 다른 이름을 부여하는 "많은 이름" 문제를 해결합니다.
패밀리 관계
악성코드 패밀리 간 관계는 다음을 포함합니다: 부모-자식(코드 재사용, 포크), 로더-페이로드(Emotet이 TrickBot을 로드하고 TrickBot이 Ryuk을 로드), 공유 저작권(동일한 위협 행위자가 여러 도구 개발), 인프라 공유(공통 C2 프레임워크 사용).
워크플로
단계 1: 악성코드 패밀리에 대한 Malpedia API 쿼리
import requests
import json
from collections import defaultdict
class MalpediaClient:
BASE_URL = "https://malpedia.caad.fkie.fraunhofer.de/api"
def __init__(self, api_key):
self.headers = {"Authorization": f"apitoken {api_key}"}
def get_family_list(self):
"""Get list of all malware families."""
resp = requests.get(f"{self.BASE_URL}/list/families",
headers=self.headers, timeout=30)
if resp.status_code == 200:
families = resp.json()
print(f"[+] Malpedia: {len(families)} malware families")
return families
return {}
def get_family_info(self, family_name):
"""Get detailed information about a malware family."""
resp = requests.get(f"{self.BASE_URL}/get/family/{family_name}",
headers=self.headers, timeout=30)
if resp.status_code == 200:
info = resp.json()
print(f"[+] Family: {family_name}")
print(f" Aliases: {info.get('alt_names', [])}")
print(f" Actors: {[a.get('value', '') for a in info.get('attribution', [])]}")
print(f" URLs: {len(info.get('urls', []))} references")
return info
print(f"[-] Family not found: {family_name}")
return None
def get_family_yara(self, family_name):
"""Get YARA rules for a malware family."""
resp = requests.get(f"{self.BASE_URL}/get/yara/{family_name}",
headers=self.headers, timeout=30)
if resp.status_code == 200:
rules = resp.json()
rule_count = sum(len(v) for v in rules.values()) if isinstance(rules, dict) else 0
print(f"[+] YARA rules for {family_name}: {rule_count} rules")
return rules
return {}
def get_actor_families(self, actor_name):
"""Get malware families associated with a threat actor."""
resp = requests.get(f"{self.BASE_URL}/get/actor/{actor_name}",
headers=self.headers, timeout=30)
if resp.status_code == 200:
data = resp.json()
families = data.get("families", {})
print(f"[+] {actor_name}: {len(families)} malware families")
return data
return {}
def search_families(self, keyword):
"""Search families by keyword."""
all_families = self.get_family_list()
matches = {
name: info for name, info in all_families.items()
if keyword.lower() in name.lower()
or keyword.lower() in str(info.get("alt_names", [])).lower()
}
print(f"[+] Search '{keyword}': {len(matches)} matches")
return matches
client = MalpediaClient("YOUR_MALPEDIA_API_KEY")
families = client.get_family_list()
emotet_info = client.get_family_info("win.emotet")
단계 2: 악성코드 패밀리 관계 매핑
class MalwareFamilyMapper:
def __init__(self, malpedia_client):
self.client = malpedia_client
self.relationship_graph = defaultdict(list)
def map_actor_ecosystem(self, actor_name):
"""Map the malware ecosystem used by a threat actor."""
actor_data = self.client.get_actor_families(actor_name)
families = actor_data.get("families", {})
ecosystem = {
"actor": actor_name,
"families": [],
"family_count": len(families),
}
for family_name in families:
info = self.client.get_family_info(family_name)
if info:
ecosystem["families"].append({
"name": family_name,
"aliases": info.get("alt_names", []),
"description": info.get("description", "")[:200],
"shared_actors": [
a.get("value", "")
for a in info.get("attribution", [])
],
"reference_count": len(info.get("urls", [])),
})
print(f"\n=== {actor_name} Malware Ecosystem ===")
for fam in ecosystem["families"]:
shared = [a for a in fam["shared_actors"] if a != actor_name]
print(f" {fam['name']}")
print(f" Aliases: {fam['aliases'][:5]}")
if shared:
print(f" Also used by: {shared}")
return ecosystem
def find_shared_tooling(self, actor_names):
"""Find malware families shared between threat actors."""
actor_families = {}
for actor in actor_names:
data = self.client.get_actor_families(actor)
actor_families[actor] = set(data.get("families", {}).keys())
# Find overlaps
shared = {}
for i, actor1 in enumerate(actor_names):
for actor2 in actor_names[i+1:]:
common = actor_families[actor1] & actor_families[actor2]
if common:
shared[f"{actor1} <-> {actor2}"] = sorted(common)
print(f"\n=== Shared Tooling Analysis ===")
for pair, families in shared.items():
print(f" {pair}: {len(families)} shared families")
for f in families[:5]:
print(f" - {f}")
return shared
def build_loader_payload_chain(self, family_name):
"""Build the loader-payload delivery chain for a family."""
info = self.client.get_family_info(family_name)
if not info:
return {}
chain = {
"family": family_name,
"description": info.get("description", ""),
"known_loaders": [],
"known_payloads": [],
}
# Common known delivery chains
known_chains = {
"win.emotet": {"loaders": ["email/macro"], "payloads": ["win.trickbot", "win.qakbot", "win.cobalt_strike"]},
"win.trickbot": {"loaders": ["win.emotet"], "payloads": ["win.ryuk", "win.conti", "win.cobalt_strike"]},
"win.qakbot": {"loaders": ["email/macro", "win.emotet"], "payloads": ["win.cobalt_strike", "win.blackbasta"]},
"win.cobalt_strike": {"loaders": ["win.emotet", "win.trickbot", "win.qakbot"], "payloads": ["ransomware"]},
}
if family_name in known_chains:
chain["known_loaders"] = known_chains[family_name]["loaders"]
chain["known_payloads"] = known_chains[family_name]["payloads"]
return chain
mapper = MalwareFamilyMapper(client)
ecosystem = mapper.map_actor_ecosystem("Wizard Spider")
shared = mapper.find_shared_tooling(["Wizard Spider", "FIN7", "Lazarus Group"])
chain = mapper.build_loader_payload_chain("win.emotet")
단계 3: YARA 규칙 추출 및 컴파일
def compile_yara_ruleset(client, family_names, output_file="malware_yara_rules.yar"):
"""Compile YARA rules for multiple malware families."""
all_rules = []
for family in family_names:
yara_data = client.get_family_yara(family)
if isinstance(yara_data, dict):
for source, rules in yara_data.items():
if isinstance(rules, list):
for rule in rules:
all_rules.append(f"// Source: {source} - Family: {family}\n{rule}")
elif isinstance(rules, str):
all_rules.append(f"// Source: {source} - Family: {family}\n{rules}")
with open(output_file, "w") as f:
f.write(f"// Malpedia YARA Rules - {len(all_rules)} rules\n")
f.write(f"// Families: {', '.join(family_names)}\n\n")
for rule in all_rules:
f.write(rule + "\n\n")
print(f"[+] Compiled {len(all_rules)} YARA rules to {output_file}")
return all_rules
compile_yara_ruleset(client, ["win.emotet", "win.trickbot", "win.cobalt_strike"])
검증 기준
- Malpedia API가 악성코드 패밀리에 대해 성공적으로 쿼리됨
- 별칭, 행위자 및 참조를 포함한 패밀리 정보가 검색됨
- 행위자-패밀리 관계가 올바르게 매핑됨
- 행위자 간 공유 툴링이 식별됨
- 탐지를 위한 YARA 규칙이 추출 및 컴파일됨
- 위협 인텔리전스를 위한 로더-페이로드 체인이 문서화됨