적대 인프라 추적 시스템 구축은(는) 무엇을 하나요?
개요
적대 인프라 추적은 수동 DNS 레코드, 인증서 투명성 로그, WHOIS 등록 데이터, IP 보강을 사용하여 위협 행위자의 명령 및 제어(C2) 네트워크를 발견, 매핑, 모니터링합니다. 공격자는 종종 캠페인 전반에 걸쳐 호스팅 제공업체, 등록 대행자, SSL 인증서, 명명 패턴을 재사용하므로 분석가는 알려진 지표에서 출발하여 새로운 인프라를 발견할 수 있습니다. 이 스킬은 인프라 관계를 식별하고, 적대 패턴과 일치하는 새로 등록된 도메인을 감지하며, 위협 행위자 네트워크의 지속적으로 업데이트되는 맵을 유지하는 자동화된 추적 시스템 구축을 다룹니다.
사용 시기
- 환경에 적대 인프라 추적 시스템 기능을 배포하거나 구성할 때
- 규정 준수 요구 사항에 맞는 보안 제어를 수립할 때
- 이 도메인에 대한 보안 아키텍처를 구축하거나 개선할 때
- 이 구현을 필요로 하는 보안 평가를 수행할 때
사전 요구 사항
- Python 3.9+ 및
requests,dnspython,python-whois,shodan,networkx라이브러리 - API 키: SecurityTrails, PassiveTotal/RiskIQ, Shodan, VirusTotal
- 수동 DNS 데이터 소스에 대한 접근
- DNS 인프라, 호스팅 및 도메인 등록에 대한 이해
- 관계 시각화를 위한 그래프 데이터베이스(Neo4j) 또는 NetworkX
주요 개념
수동 DNS
수동 DNS는 과거 DNS 확인 데이터를 캡처하여 어떤 도메인이 어떤 IP에 언제 확인되었는지 기록합니다. 활성 DNS 쿼리와 달리 수동 DNS는 레코드가 변경된 후에도 과거 관계를 보존하므로 분석가는 인프라 변경을 추적하고, 공유 호스팅 패턴을 식별하며, 시간이 지남에 따라 동일한 IP 주소로 확인된 관련 도메인을 발견할 수 있습니다.
인프라 피벗
피벗은 연결을 따라 관련 인프라를 식별합니다: IP 피벗(IP의 모든 도메인 찾기), 도메인 피벗(도메인이 확인된 모든 IP 찾기), WHOIS 피벗(동일한 등록자의 도메인 찾기), 인증서 피벗(SSL 인증서를 공유하는 호스트 찾기), NS/MX 피벗(동일한 네임 서버 또는 메일 서버를 사용하는 도메인 찾기).
적대 인프라 패턴
위협 행위자는 패턴을 보입니다: 선호하는 등록 대행자(Namecheap, REG.RU, Tucows), 선호하는 호스팅(방화벽 호스팅 제공업체, 클라우드 서비스), 도메인 생성 알고리즘(DGA), 일관된 명명 패턴 및 캠페인 전반에 걸친 인증서 재사용.
워크플로우
1단계: 수동 DNS 인프라 발견
import requests
import json
from collections import defaultdict
from datetime import datetime
class 인프라추적자:
def __init__(self, 보안트레일키=None, vt키=None, 쇼단키=None):
self.st_키 = 보안트레일키
self.vt_키 = vt키
self.쇼단_키 = 쇼단키
self.인프라_그래프 = defaultdict(lambda: {"노드": set(), "에지": []})
def 수동_dns_조회(self, 도메인):
"""도메인 확인 이력에 대한 수동 DNS 쿼리."""
헤더 = {"apikey": self.st_키}
url = f"https://api.securitytrails.com/v1/history/{도메인}/dns/a"
resp = requests.get(url, headers=헤더, timeout=30)
if resp.status_code == 200:
레코드 = resp.json().get("records", [])
이력 = []
for 레코드 in 레코드:
for 값 in 레코드.get("values", []):
이력.append({
"도메인": 도메인,
"ip": 값.get("ip", ""),
"최초_확인": 레코드.get("first_seen", ""),
"최종_확인": 레코드.get("last_seen", ""),
"유형": 레코드.get("type", "a"),
})
print(f"[+] {도메인}에 대한 수동 DNS: {len(이력)}개 레코드")
return 이력
return []
def 역ip_조회(self, ip_주소):
"""IP 주소에 호스팅된 모든 도메인 찾기."""
헤더 = {"apikey": self.st_키}
url = f"https://api.securitytrails.com/v1/ips/nearby/{ip_주소}"
resp = requests.get(url, headers=헤더, timeout=30)
if resp.status_code == 200:
블록 = resp.json().get("blocks", [])
도메인들 = []
for 블록 in 블록:
for 사이트 in 블록.get("sites", []):
도메인들.append(사이트)
print(f"[+] {ip_주소}에 대한 역IP: {len(도메인들)}개 도메인")
return 도메인들
return []
def whois_조회(self, 도메인):
"""피벗을 위한 WHOIS 등록 데이터 가져오기."""
헤더 = {"apikey": self.st_키}
url = f"https://api.securitytrails.com/v1/domain/{도메인}/whois"
resp = requests.get(url, headers=헤더, timeout=30)
if resp.status_code == 200:
데이터 = resp.json()
whois_데이터 = {
"도메인": 도메인,
"등록대행자": 데이터.get("registrar", ""),
"등록자_조직": 데이터.get("registrant_org", ""),
"등록자_이메일": 데이터.get("registrant_email", ""),
"네임_서버": 데이터.get("nameServers", []),
"생성일": 데이터.get("createdDate", ""),
"갱신일": 데이터.get("updatedDate", ""),
"만료일": 데이터.get("expiresDate", ""),
}
return whois_데이터
return {}
def 시드에서_피벗(self, 시드_지표, 지표_유형="도메인", 깊이=2):
"""시드 지표에서 재귀적으로 피벗하여 인프라 발견."""
발견 = {"도메인": set(), "ip": set(), "관계": []}
if 지표_유형 == "도메인":
발견["도메인"].add(시드_지표)
# 도메인에 대한 IP 가져오기
pdns = self.수동_dns_조회(시드_지표)
for 레코드 in pdns:
ip = 레코드["ip"]
발견["ip"].add(ip)
발견["관계"].append({
"소스": 시드_지표, "대상": ip,
"유형": "확인됨",
"최초_확인": 레코드["first_seen"],
"최종_확인": 레코드["last_seen"],
})
if 깊이 > 1:
# 발견된 IP에 대한 역 조회
역_도메인 = self.역ip_조회(ip)
for rd in 역_도메인[:20]:
발견["도메인"].add(rd)
발견["관계"].append({
"소스": rd, "대상": ip,
"유형": "호스팅됨",
})
elif 지표_유형 == "ip":
발견["ip"].add(시드_지표)
도메인 = self.역ip_조회(시드_지표)
for 도메인 in 도메인[:20]:
발견["도메인"].add(도메인)
발견["관계"].append({
"소스": 도메인, "대상": 시드_지표,
"유형": "호스팅됨",
})
print(f"[+] {시드_지표}에서 피벗: "
f"{len(발견['도메인'])}개 도메인, "
f"{len(발견['ip'])}개 IP, "
f"{len(발견['관계'])}개 관계")
return 발견
추적자 = 인프라추적자(
보안트레일키="YOUR_ST_KEY",
vt키="YOUR_VT_KEY",
)
2단계: 인프라 그래프 구축
import networkx as nx
class 인프라그래프:
def __init__(self):
self.그래프 = nx.Graph()
def 발견_추가(self, 발견_데이터):
"""발견된 인프라를 그래프에 추가."""
for 도메인 in 발견_데이터["도메인"]:
self.그래프.add_node(도메인, type="도메인")
for ip in 발견_데이터["ip"]:
self.그래프.add_node(ip, type="ip")
for 관계 in 발견_데이터["관계"]:
self.그래프.add_edge(
관계["소스"], 관계["대상"],
relationship=관계["유형"],
first_seen=관계.get("최초_확인", ""),
last_seen=관계.get("최종_확인", ""),
)
def 클러스터_찾기(self):
"""인프라 클러스터 식별."""
구성요소 = list(nx.connected_components(self.그래프))
클러스터 = []
for 구성요소 in 구성요소:
도메인 = [n for n in 구성요소 if self.그래프.nodes[n].get("type") == "도메인"]
ips = [n for n in 구성요소 if self.그래프.nodes[n].get("type") == "ip"]
클러스터.append({
"크기": len(구성요소),
"도메인": sorted(도메인),
"ip": sorted(ips),
"도메인_수": len(도메인),
"ip_수": len(ips),
})
클러스터.sort(key=lambda x: x["크기"], reverse=True)
print(f"[+] 인프라 클러스터: {len(클러스터)}개")
return 클러스터
def 허브_노드_찾기(self, 상위_n=10):
"""높은 중심성을 가진 노드(공유 인프라) 찾기."""
중심성 = nx.degree_centrality(self.그래프)
상위_노드 = sorted(중심성.items(), key=lambda x: x[1], reverse=True)[:상위_n]
허브 = []
for 노드, 점수 in 상위_노드:
허브.append({
"노드": 노드,
"유형": self.그래프.nodes[노드].get("type", "unknown"),
"중심성": round(점수, 4),
"연결": self.그래프.degree(노드),
})
return 허브
def 그래프_내보내기(self, 출력_파일="infrastructure_graph.json"):
데이터 = nx.node_link_data(self.그래프)
with open(출력_파일, "w") as f:
json.dump(데이터, f, indent=2)
print(f"[+] 그래프 내보냄: {self.그래프.number_of_nodes()}개 노드, "
f"{self.그래프.number_of_edges()}개 에지")
인프라_그래프 = 인프라그래프()
발견 = 추적자.시드에서_피벗("evil-domain.com", 깊이=2)
인프라_그래프.발견_추가(발견)
클러스터 = 인프라_그래프.클러스터_찾기()
허브 = 인프라_그래프.허브_노드_찾기()
인프라_그래프.그래프_내보내기()
3단계: 새로운 인프라 모니터링
import time
class 인프라모니터:
def __init__(self, 추적자, 알려진_지표):
self.추적자 = 추적자
self.알려진 = set(알려진_지표)
self.경보 = []
def 신규_등록_확인(self, 패턴):
"""적대 패턴과 일치하는 새로 등록된 도메인 확인."""
import re
새_도메인 = []
for 패턴 in 패턴:
# SecurityTrails에 패턴과 일치하는 새 도메인 쿼리
헤더 = {"apikey": self.추적자.st_키}
url = "https://api.securitytrails.com/v1/domains/list"
params = {"include_ips": "true", "page": 1}
바디 = {"filter": {"keyword": 패턴}}
resp = requests.post(url, headers=헤더, json=바디, timeout=30)
if resp.status_code == 200:
레코드 = resp.json().get("records", [])
for 레코드 in 레코드:
도메인 = 레코드.get("hostname", "")
if 도메인 not in self.알려진:
새_도메인.append({
"도메인": 도메인,
"일치_패턴": 패턴,
"최초_확인": datetime.now().isoformat(),
})
self.알려진.add(도메인)
if 새_도메인:
print(f"[경보] 패턴과 일치하는 {len(새_도메인)}개 신규 도메인")
self.경보.extend(새_도메인)
return 새_도메인
def 인프라_보고서_생성(self, 클러스터, 허브):
보고서 = f"""# 적대 인프라 추적 보고서
생성일: {datetime.now().isoformat()}
## 요약
- 식별된 인프라 클러스터: {len(클러스터)}개
- 추적된 총 도메인: {sum(c['도메인_수'] for c in 클러스터)}개
- 추적된 총 IP: {sum(c['ip_수'] for c in 클러스터)}개
- 감지된 신규 도메인: {len(self.경보)}개
## 주요 인프라 허브
| 노드 | 유형 | 연결 | 중심성 |
|------|------|------|--------|
"""
for 허브 in 허브[:10]:
보고서 += (f"| {허브['노드']} | {허브['유형']} "
f"| {허브['연결']} | {허브['중심성']} |\n")
보고서 += "\n## 인프라 클러스터\n"
for i, 클러스터 in enumerate(클러스터[:5], 1):
보고서 += f"\n### 클러스터 {i} ({클러스터['크기']}개 노드)\n"
보고서 += f"- 도메인: {', '.join(클러스터['도메인'][:5])}\n"
보고서 += f"- IP: {', '.join(클러스터['ip'][:5])}\n"
with open("infrastructure_report.md", "w") as f:
f.write(보고서)
print("[+] 인프라 보고서 저장됨")
모니터 = 인프라모니터(추적자, 알려진_지표=set())
검증 기준
- 수동 DNS 쿼리가 과거 확인 데이터를 반환
- 역IP 조회가 공동 호스팅 도메인을 발견
- 인프라 피벗이 시드 지표에서 확장
- 그래프 분석이 클러스터와 허브 노드를 식별
- 패턴 모니터링을 통해 새로운 인프라 감지
- 실행 가능한 권장 사항이 포함된 보고서 생성