npx skills add mukul975/Anthropic-Cybersecurity-SkillsMITRE ATT&CK
Overview
Malpedia is a collaborative platform maintained by Fraunhofer FKIE that catalogs malware families with their aliases, YARA rules, threat actor associations, and reference reports. With over 2,600 malware families documented, it serves as the definitive resource for understanding malware lineages, tracking variant evolution, and linking malware to specific threat groups. This skill covers querying the Malpedia API, mapping malware family relationships, extracting YARA rules for detection, and building intelligence on malware ecosystems used by adversaries.
When to Use
- When investigating security incidents that require analyzing malware family relationships with malpedia
- When building detection rules or threat hunting queries for this domain
- When SOC analysts need structured procedures for this analysis type
- When validating security monitoring coverage for related attack techniques
Prerequisites
- Python 3.9+ with
requests,yara-python,stix2libraries - Malpedia API key (register at https://malpedia.caad.fkie.fraunhofer.de/)
- Understanding of malware classification and naming conventions
- Familiarity with YARA rule syntax for detection
- Access to malware samples for validation (optional)
Key Concepts
Malpedia Data Model
Malpedia organizes malware into Families (e.g., "win.cobalt_strike"), each containing: aliases (vendor-specific names like "Beacon", "CobaltStrike"), YARA rules (community and vendor-contributed), actor associations (threat groups using the family), reference reports (CTI reports documenting the family), and sample hashes (representative samples for each variant).
Malware Family Naming
Malpedia uses the format platform.family_name (e.g., win.emotet, elf.mirai, apk.flubot). Platforms include win (Windows), elf (Linux), apk (Android), osx (macOS), and py (Python). This standardized naming resolves the "many names" problem where different vendors assign different names to the same malware.
Family Relationships
Malware families have relationships including: parent-child (code reuse, forks), loader-payload (Emotet loads TrickBot loads Ryuk), shared authorship (same threat actor develops multiple tools), and infrastructure sharing (common C2 frameworks).
Workflow
Step 1: Query Malpedia API for Malware Families
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")Step 2: Map Malware Family Relationships
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")Step 3: Extract and Compile YARA Rules
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"])Validation Criteria
- Malpedia API queried successfully for malware families
- Family information retrieved with aliases, actors, and references
- Actor-family relationships mapped correctly
- Shared tooling between actors identified
- YARA rules extracted and compiled for detection
- Loader-payload chains documented for threat intelligence
References
References and resources
Everything below is rendered for inspection. Script files are read-only and never run.
References 1
api-reference.md1.4 KB
API Reference: Malpedia Malware Family Analysis
Base URL
https://malpedia.caad.fkie.fraunhofer.de/apiAuthentication
Authorization: apitoken YOUR_API_KEYList Families
GET /list/familiesReturns dict of {family_name: {alt_names, description, attribution, urls}}.
Get Family Details
GET /get/family/{family_name}| Field | Description |
|---|---|
common_name |
Primary family name |
alt_names |
List of alternative names |
description |
Family description |
attribution |
List of attributed threat actors |
urls |
Reference URLs |
Get YARA Rules
GET /get/yara/{family_name}Returns dict of YARA rules keyed by rule source.
List Actors
GET /list/actorsReturns dict of {actor_name: {alt_names, description, families}}.
Get Actor Details
GET /get/actor/{actor_name}| Field | Description |
|---|---|
common_name |
Actor name |
description |
Actor profile |
families |
Associated malware families |
alt_names |
Alternative names (APT designations) |
Get Sample
GET /get/sample/{sha256}
GET /get/sample/{sha256}/zipRelationship Types
| Relation | Description |
|---|---|
also_known_as |
Family alias |
shared_actor |
Families used by same threat actor |
variant_of |
Derived malware variant |
MITRE ATT&CK
- T1587.001 - Develop Capabilities: Malware
Scripts 1
agent.py5.4 KB
#!/usr/bin/env python3
"""Malpedia Malware Family Relationship Agent - Queries Malpedia API for malware family intelligence."""
import json
import logging
import argparse
from datetime import datetime
from collections import defaultdict
import requests
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)
MALPEDIA_API = "https://malpedia.caad.fkie.fraunhofer.de/api"
def malpedia_get(endpoint, api_key):
"""Make authenticated GET request to Malpedia API."""
headers = {"Authorization": f"apitoken {api_key}"}
resp = requests.get(f"{MALPEDIA_API}{endpoint}", headers=headers, timeout=30)
resp.raise_for_status()
return resp.json()
def list_families(api_key):
"""List all malware families from Malpedia."""
data = malpedia_get("/list/families", api_key)
logger.info("Retrieved %d malware families", len(data))
return data
def get_family_info(family_name, api_key):
"""Get detailed info for a malware family."""
return malpedia_get(f"/get/family/{family_name}", api_key)
def get_family_yara(family_name, api_key):
"""Get YARA rules for a malware family."""
return malpedia_get(f"/get/yara/{family_name}", api_key)
def list_actors(api_key):
"""List all threat actors from Malpedia."""
data = malpedia_get("/list/actors", api_key)
logger.info("Retrieved %d threat actors", len(data))
return data
def get_actor_info(actor_name, api_key):
"""Get detailed info for a threat actor."""
return malpedia_get(f"/get/actor/{actor_name}", api_key)
def build_family_graph(families_data):
"""Build relationship graph between malware families."""
relationships = []
family_actors = defaultdict(list)
for family_name, info in families_data.items():
if not isinstance(info, dict):
continue
alt_names = info.get("alt_names", [])
actors = info.get("attribution", [])
urls = info.get("urls", [])
for actor in actors:
family_actors[actor].append(family_name)
for alt in alt_names:
relationships.append({
"source": family_name,
"target": alt,
"relation": "also_known_as",
})
for actor, actor_families in family_actors.items():
if len(actor_families) > 1:
for i in range(len(actor_families)):
for j in range(i + 1, len(actor_families)):
relationships.append({
"source": actor_families[i],
"target": actor_families[j],
"relation": "shared_actor",
"actor": actor,
})
return relationships, dict(family_actors)
def analyze_family(family_name, api_key):
"""Analyze a specific malware family and its relationships."""
info = get_family_info(family_name, api_key)
result = {
"family": family_name,
"description": info.get("description", ""),
"alt_names": info.get("alt_names", []),
"attribution": info.get("attribution", []),
"urls": info.get("urls", [])[:10],
"common_name": info.get("common_name", ""),
}
try:
yara_data = get_family_yara(family_name, api_key)
result["yara_rule_count"] = len(yara_data) if isinstance(yara_data, dict) else 0
except requests.RequestException:
result["yara_rule_count"] = 0
return result
def generate_report(families_analyzed, relationships, actor_map):
"""Generate malware family relationship report."""
report = {
"timestamp": datetime.utcnow().isoformat(),
"families_analyzed": len(families_analyzed),
"relationships_found": len(relationships),
"actors_mapped": len(actor_map),
"family_details": families_analyzed,
"relationships": relationships[:200],
"actor_family_map": {a: f for a, f in list(actor_map.items())[:50]},
}
print(f"MALPEDIA REPORT: {len(families_analyzed)} families, {len(relationships)} relationships, {len(actor_map)} actors")
return report
def main():
parser = argparse.ArgumentParser(description="Malpedia Malware Family Analysis Agent")
parser.add_argument("--api-key", required=True, help="Malpedia API key")
parser.add_argument("--family", help="Specific family to analyze")
parser.add_argument("--list-families", action="store_true")
parser.add_argument("--build-graph", action="store_true", help="Build full relationship graph")
parser.add_argument("--output", default="malpedia_report.json")
args = parser.parse_args()
families_analyzed = []
relationships = []
actor_map = {}
if args.family:
result = analyze_family(args.family, args.api_key)
families_analyzed.append(result)
elif args.build_graph:
all_families = list_families(args.api_key)
relationships, actor_map = build_family_graph(all_families)
elif args.list_families:
all_families = list_families(args.api_key)
families_analyzed = [{"family": k, "alt_names": v.get("alt_names", []) if isinstance(v, dict) else []} for k, v in list(all_families.items())[:100]]
report = generate_report(families_analyzed, relationships, actor_map)
with open(args.output, "w") as f:
json.dump(report, f, indent=2)
logger.info("Report saved to %s", args.output)
if __name__ == "__main__":
main()