
사이버 위협 방어 세계 모델링
사이버 위협 방어 세계 모델링 시스템
Bandjacks는 포괄적인 사이버 위협 인텔리전스(CTI) 시스템으로:
| 가이드 | 설명 |
|---|---|
| 빠른 시작 | 5분 안에 실행하기 |
| 전체 설정 | 완전한 환경 설정 |
| CLI 사용법 | 명령줄 인터페이스 가이드 |
| API 참조 | REST API 문서 |
| 동시 발생 분석 | 분석 문서 |
| AttackFlow 생성 | 흐름 생성 가이드 |
| 검토 시스템 | 인간 참여 검토 |
git clone https://github.com/yourusername/bandjacks.git cd bandjacks
uv sync
pip install -e .
cd ui && npm install && cd ..
### 환경 설정
**중요:** 애플리케이션을 시작하기 전에 환경 변수를 설정해야 합니다. 애플리케이션은 `NEO4J_PASSWORD`가 설정되어 있어야 합니다.
프로젝트 루트에 `.env` 파일을 생성하세요:```bash
# Copy the sample file
cp infra/env.sample .env
# Edit .env and set your actual passwords
nano .env
.env에 필요한 구성:```bash
NEO4J_URI=bolt://localhost:7687 NEO4J_USER=neo4j NEO4J_PASSWORD=your-actual-neo4j-password # MUST BE SET - no default provided
OPENSEARCH_URL=http://localhost:9200 OPENSEARCH_USER=admin OPENSEARCH_PASSWORD=your-opensearch-password # Optional if security is disabled
LOCAL_LLM_API_BASE=http://192.168.1.100:8080/v1 # Base URL of your local server LOCAL_LLM_MODEL=mistral-nemo # Model name as the server reports it LOCAL_LLM_API_KEY=no-key # Most local servers accept any value
PRIMARY_LLM=gemini GOOGLE_API_KEY=your-gemini-api-key
OPENAI_API_KEY=your-openai-api-key
ATTACK_INDEX_URL=https://raw.githubusercontent.com/mitre-attack/attack-stix-data/master/index.json ATTACK_COLLECTION=enterprise-attack ATTACK_VERSION=latest
REDIS_URL=redis://localhost:6379
**참고:** `NEO4J_PASSWORD`가 설정되지 않으면 애플리케이션이 시작되지 않습니다. 자세한 내용은 [환경 변수 수정](https://github.com/blevene/bandjacks/blob/HEAD/ENV_VARIABLES_FIX.md)을 참조하세요.
### 서비스 시작```bash
# Start the FastAPI backend server
uv run uvicorn bandjacks.services.api.main:app --reload --port 8000
# In another terminal, start the Next.js frontend
cd ui && npm run dev
# Access the applications
open http://localhost:8000/docs # API documentation
open http://localhost:3000 # Frontend UI
Bandjacks는 위협 인텔리전스 작업을 위한 포괄적인 CLI를 포함합니다:```bash
uv run python -m bandjacks.cli.main --help
> **참고:** CLI는 환경 변수(NEO4J_PASSWORD 등)가 설정되어 있어야 합니다. `.env` 파일이 있는 프로젝트 루트에서 실행하세요.
### 쿼리 명령어```bash
# Search for threat intelligence
uv run python -m bandjacks.cli.main query search "ransomware encryption techniques" --top-k 10
# Explore graph relationships
uv run python -m bandjacks.cli.main query graph "attack-pattern--abc123" --depth 2
uv run python -m bandjacks.cli.main review queue --status pending --limit 20
uv run python -m bandjacks.cli.main review approve "candidate-123" --reviewer analyst-1
uv run python -m bandjacks.cli.main review reject "candidate-456" --reviewer analyst-1 --reason "False positive"
### 문서 추출```bash
# Extract CTI from a document
uv run python -m bandjacks.cli.main extract document ./report.pdf --confidence-threshold 80 --show-evidence
참고: 분석 명령어가 결과를 반환하려면 Neo4j에
AttackEpisode데이터가 필요합니다.```bash
uv run python -m bandjacks.cli.main analytics top-cooccurrence --limit 25 --min-episode-size 2
uv run python -m bandjacks.cli.main analytics conditional "attack-pattern--abc123" --limit 25
uv run python -m bandjacks.cli.main analytics actor "intrusion-set--xyz789" --metric npmi
uv run python -m bandjacks.cli.main analytics bundles --min-support 3 --min-size 3 --max-size 5 --format json --output bundles.json
uv run python -m bandjacks.cli.main analytics global --min-support 2 --limit 50 --format csv --output pairs.csv
### 워크플로 명령```bash
# Process a directory of reports with analytics
uv run python -m bandjacks.cli.main workflow process-reports ./reports/ --workers 3 --analyze --export-dir ./results/
# Bulk export all analytics data
uv run python -m bandjacks.cli.main workflow bulk-export --export-dir ./analytics_export/
uv run python -m bandjacks.cli.main admin health
uv run python -m bandjacks.cli.main admin cache-stats
uv run python -m bandjacks.cli.main admin cache-clear --pattern "search:*"
uv run python -m bandjacks.cli.main admin optimize
## Frontend UI
Next.js 프론트엔드는 시스템 작업을 위한 현대적인 인터페이스를 제공합니다.
### Report Management (`/reports`)
- **Report List**: 모든 수집된 보고서를 상태 및 기법 카운트와 함께 확인
- **New Report** (`/reports/new`): PDF/TXT 파일 업로드 또는 보고서 내용 붙여넣기
- **Report Detail** (`/reports/[id]`): 추출된 기법, 엔터티 및 증거 확인
- **Review Interface** (`/reports/[id]/review`): 사람이 참여하는 검토 워크플로우
### Co-occurrence Analytics (`/analytics/cooccurrence`)
> **Note:** 이 페이지들은 Neo4j의 `AttackEpisode` 데이터가 필요합니다. 먼저 추출 파이프라인을 통해 보고서를 처리하거나, `POST /v1/flows/build`를 사용하여 침입 세트 데이터에서 에피소드를 생성하세요.
- **Hub Page**: 에피소드/기법/행위자 카운트 개요
- **Top Pairs** (`/pairs`): NPMI/Lift 메트릭으로 동시 발생 기법 쌍
- **Conditional** (`/conditional`): P(B|A) 조건부 확률
- **Bundles** (`/bundles`): 자주 동시 발생하는 기법 번들
- **Actors** (`/actors`): 행위자별 기법 패턴
- **Bridging** (`/bridging`): 여러 행위자에 걸쳐 사용된 기법
### System Health (`/health`)
- 모든 구성 요소(Neo4j, OpenSearch, Redis)의 실시간 상태
- 캐시 통계 및 메모리 사용량
- Kubernetes 호환 상태 엔드포인트
### Starting the Frontend```bash
cd ui
npm run dev # Development mode with hot reload
npm run build # Production build
npm run start # Start production server
# Ensure backend is running
# API_URL defaults to http://localhost:8000/v1
먼저, MITRE ATT&CK 프레임워크를 지식 그래프에 로드하십시오:```bash
curl -X POST "http://localhost:8000/v1/stix/load/attack"
-H "Content-Type: application/json"
-d '{
"collection": "enterprise-attack",
"version": "latest",
"adm_strict": false
}'
### 2. 보고서에서 기법 추출
위협 인텔리전스 보고서에서 MITRE ATT&CK 기법을 추출하세요:```python
import httpx
import time
# For small reports (<5KB) - synchronous processing
response = httpx.post(
"http://localhost:8000/v1/reports/ingest",
json={
"content": "APT29 used spearphishing emails with malicious attachments...",
"title": "APT29 Campaign Analysis",
"config": {
"use_optimized_extractor": True,
"span_score_threshold": 0.7,
"top_k": 5
}
}
)
result = response.json()
print(f"Extracted {len(result['extraction']['techniques'])} techniques")
# For large reports (>5KB) - asynchronous processing
response = httpx.post(
"http://localhost:8000/v1/reports/ingest_async",
json={
"content": large_report_text,
"title": "Large Report Analysis"
}
)
job_id = response.json()["job_id"]
# Check job status
status = httpx.get(f"http://localhost:8000/v1/reports/jobs/{job_id}/status")
while status.json()["status"] == "processing":
time.sleep(2)
status = httpx.get(f"http://localhost:8000/v1/reports/jobs/{job_id}/status")
# Get results from completed job
result = status.json()["result"]
print(f"Extracted {result['techniques_count']} techniques in {result['elapsed_time']} seconds")
API 없이 프로그래밍 방식으로 접근하려면:```python from bandjacks.llm.extraction_pipeline import run_extraction_pipeline
config = { "use_optimized_extractor": True, # Use optimized pipeline "span_score_threshold": 0.7, # Minimum span confidence "max_spans": 20, "top_k": 5, "chunk_size": 2000, # For large documents "max_chunks": 100 }
result = run_extraction_pipeline( report_text, config, source_id="report_123", neo4j_config=neo4j_config )
techniques = result["techniques"] # Dict of technique_id -> details bundle = result.get("bundle") # STIX 2.1 bundle if configured entities = result.get("entities") # Extracted entities
for tech_id, info in techniques.items(): print(f"{tech_id}: {info['name']}") print(f" Confidence: {info['confidence']}%") print(f" Evidence: {info['evidence']}")
## 추출 파이프라인 아키텍처
Bandjacks 추출 파이프라인은 다중 에이전트 아키텍처를 사용하여 구조화된 위협 인텔리전스를 추출합니다:
### 파이프라인 구성 요소
추출 파이프라인은 순차적으로 9개의 특화된 에이전트를 사용합니다:
#### 1. **EntityExtractionAgent** - 엔터티 인식
- 위협 행위자, 악성코드, 도구 및 캠페인 추출
- 기술(technique) 추출을 위한 컨텍스트 제공을 위해 먼저 실행
- JSON 스키마 검증을 통한 퓨샷 프롬프팅 사용
- 점진적 윈도우 추출로 청크된 문서 처리
#### 2. **SpanFinderAgent** - 행동 텍스트 탐지
- 14개의 전술별 정규식 패턴을 사용하여 위협 행동을 포함하는 텍스트 스팬 탐지
- 명시적 기술 ID(T1566.001) 및 행동 패턴 식별
- 키워드 인덱스 부스팅을 통해 신뢰도에 따라 스팬 점수 매기기
- LLM 호출 없음 — 속도를 위한 순수 패턴 매칭
#### 3. **BatchRetrieverAgent** - 후보 검색
- OpenSearch KNN 벡터 검색을 사용하여 스팬별 후보 기술 찾기
- 중복 임베딩을 방지하기 위해 인코딩 전에 동일한 스팬 텍스트 중복 제거
- 각 스팬에 대해 유사도 점수와 함께 상위 k개 후보 반환
#### 4. **사전 필터** - 스팬 축소
- 후보 기술당 `max_spans_per_technique`(기본값 2)으로 스팬 제한
- 증거 품질 유지를 위해 후보당 가장 높은 점수의 스팬 유지
- 최소한의 기술 손실로 매퍼 LLM 호출 약 46% 감소
#### 5. **DiscoveryAgent** - LLM 발견 (조건부)
- 검색기 신뢰도가 낮을 때(평균 <0.7) 트리거됨
- 벡터 검색이 놓친 기술을 발견하기 위해 LLM 사용
- 모든 저신뢰 스팬에 대한 단일 배치 호출
#### 6. **BatchMapperAgent** - 기술 매핑 (LLM)
- 최대 10개의 그룹으로 스팬 배치 처리 (`MAX_MAPPER_BATCH_SIZE`, 클라우드 LLM 잘림 제한을 위해 2026-05에 기본값 25에서 낮춤)
- 신뢰도 점수와 함께 스팬당 모든 관련 기술 추출
- 구조화된 출력을 위한 JSON 스키마 검증 사용
#### 7. **EvidenceVerifierAgent** - 증거 검증
- 인용 및 라인 참조의 패턴 기반 검증
- 40-100점 척도로 증거 품질 점수 매기기
- LLM 호출 없음 — 정규식 및 텍스트 매칭
#### 8. **ConsolidatorAgent** - 증거 통합
- 여러 스팬에서 발견된 중복 기술 병합
- 자카드 유사도(>85% 임계값)를 사용하여 증거 집계
- 통합된 신뢰도 점수로 최종 기술 목록 생성
#### 9. **AttackFlowSynthesizer** - 시퀀스 생성 (LLM)
- 시간적 마커 분석 ("first", "then", "after")
- 내러티브에서 인과 관계 추론
- 확률적 엣지를 가진 STIX Attack Flow 객체 생성
- 시퀀스가 불명확할 때 동시 발생 모델링으로 대체
### 성능 최적화
- **스마트 청킹**: 문서를 2KB 청크로 분할, 중복 포함
- **배치 처리**: 매퍼가 LLM 호출당 최대 25개의 스팬 처리
- **병렬 처리**: 청크를 작업자 스레드에서 동시에 처리
- **응답 캐싱**: 중복 호출 방지를 위해 LLM 응답 캐시
- **조기 종료**: 고신뢰도 추출은 검증 건너뛰기
- **TechniqueCache**: 모든 ATT&CK 기술을 시작 시 로드하여 O(1) 조회
- **사전 필터**: LLM 매퍼 전에 후보 기술당 스팬 제한 (호출 46% 감소)
- **배치 임베딩**: 기술 임베딩을 배치로 생성 (2-5배 빠름)
- **연결 풀링**: 요청 간 공유 Neo4j/OpenSearch 연결
- **UNWIND 배치**: UNWIND를 통해 Neo4j 쓰기 배치 (30-40 쿼리 → 6-7)
- **모델 예열**: 콜드 스타트 지연 방지를 위해 시작 시 임베딩 모델 로드
### 처리 시간
| 문서 크기 | 처리 시간 | 추출된 기법 |
|--------------|-----------------|---------------------|
| 작음 (<5KB) | 10-20 초 | 5-10 기법 |
| 중간 (5-15KB) | 20-40 초 | 10-15 기법 |
| 큼 (>15KB) | 30-60 초 | 15-25 기법 |
## 동시 발생 분석
Bandjacks는 기술 관계를 이해하기 위한 분석을 제공합니다.
> **참고:** 분석을 위해서는 Neo4j에 `AttackEpisode` 및 `AttackAction` 데이터가 필요합니다. 이 데이터는 다음과 같은 경우 생성됩니다:
> - 보고서가 추출 파이프라인을 통해 처리될 때
> - `/v1/flows/build`를 통해 공격 흐름이 구축될 때
> - 공격 에피소드가 포함된 STIX 번들이 수집될 때
>
> 에피소드가 없으면 분석은 빈 결과를 반환합니다.
### 글로벌 동시 발생
모든 공격 에피소드에서 어떤 기법들이 자주 함께 나타나는지 계산합니다:```python
# Via API
response = httpx.post(
"http://localhost:8000/v1/analytics/cooccurrence/global",
json={"min_support": 2, "min_episodes_per_pair": 2, "limit": 50}
)
for pair in response.json()["pairs"]:
print(f"{pair['name_a']} + {pair['name_b']}: NPMI={pair['npmi']:.3f}")
P(B|A) 계산 - 기법 A가 사용되었을 때, 기법 B의 확률은 얼마인가:```python response = httpx.get( "http://localhost:8000/v1/analytics/cooccurrence/conditional", params={"technique_id": "attack-pattern--abc123", "limit": 25} )
### 기법 번들
자주 함께 발생하는 기법 번들(3-5개 기법)을 식별하세요:```python
response = httpx.post(
"http://localhost:8000/v1/analytics/cooccurrence/bundles",
json={"min_support": 3, "min_size": 3, "max_size": 5}
)
특정 위협 행위자의 기술 패턴 분석:```python response = httpx.post( "http://localhost:8000/v1/analytics/cooccurrence/actor", json={"intrusion_set_id": "intrusion-set--xyz789", "min_support": 1} )
## 인간 참여 검토 시스템
Bandjacks는 추출된 인텔리전스를 검증하기 위한 포괄적인 검토 시스템을 포함합니다:
### 통합 검토 인터페이스
검토 시스템은 모든 추출된 항목을 단일 인터페이스에 표시합니다:```typescript
// Review workflow
1. Upload/ingest report → Extraction pipeline runs
2. Navigate to /reports/{id}/review
3. Review extracted items across three tabs:
- Entities (threat actors, malware, tools)
- Techniques (ATT&CK mappings with evidence)
- Attack Flow (sequenced steps)
4. Take actions on each item:
- Approve: Accept as correct
- Reject: Mark as incorrect
- Edit: Modify details (name, confidence, etc.)
5. Submit all decisions atomically
response = httpx.post( f"http://localhost:8000/v1/reports/{report_id}/unified-review", json={ "decisions": [ { "item_id": "technique-0", "action": "approve", "confidence_adjustment": 5, "notes": "Confirmed via external CTI" }, { "item_id": "entity-malware-1", "action": "edit", "edited_value": { "name": "Corrected Malware Name", "confidence": 95 } } ], "global_notes": "Review completed by analyst-1" } )
### 4. 기술 검색
자연어를 사용하여 ATT&CK 기술 검색:```python
# Vector search for similar techniques
response = httpx.post(
"http://localhost:8000/v1/search/ttx",
json={
"query": "ransomware that encrypts files and demands payment",
"top_k": 5
}
)
techniques = response.json()["results"]
for tech in techniques:
print(f"{tech['external_id']}: {tech['name']} (score: {tech['score']:.2f})")
지식 그래프에서 관계를 쿼리합니다:```python
response = httpx.get( "http://localhost:8000/v1/graph/group/G0016/techniques" )
response = httpx.get( "http://localhost:8000/v1/defense/technique/T1566.001" )
### 6. AttackFlow 모델 생성
위협 행위자가 기법을 함께 사용하는 방식을 보여주는 동시 발생 모델을 생성합니다:```python
# Generate flow for a specific intrusion set (e.g., APT29)
response = httpx.post(
"http://localhost:8000/v1/flows/build",
json={
"intrusion_set_id": "intrusion-set--899ce53f-13a0-479b-a0e4-67d46e241542"
}
)
flow = response.json()
print(f"Generated flow '{flow['name']}' with {len(flow['steps'])} techniques")
print(f"Co-occurrence edges: {len(flow['edges'])}")
대량 생성: 모든 위협 행위자와 기법에 대한 플로우 생성:```bash
uv run python scripts/build_intrusion_flows_simple.py
AttackFlow 모델은 침해 집합에 본질적인 순서 정보가 없으므로 순차적 순서보다 **동시 발생**을 사용합니다. 기법은 다음으로 연결됩니다:
- **전술 내 에지**: 동일한 킬 체인 전술 내 기법 간
- **전술 간 에지**: 인접한 전술 간 기법 간
- **허브-스포크 패턴**: 에지 폭발을 방지하기 위한 대규모 기법 집합
자세한 사용법은 [AttackFlow 생성 가이드](https://github.com/blevene/bandjacks/blob/HEAD/docs/ATTACKFLOW_GENERATION.md)를 참조하세요.
## 지원되는 입력 형식
추출 파이프라인은 여러 입력 형식을 지원합니다:
- **일반 텍스트** - 직접 텍스트 콘텐츠
- **마크다운** - 서식이 있는 마크다운 문서
- **PDF** - pdfplumber 추출을 통해
- **HTML** - BeautifulSoup 파싱을 통해
- **JSON** - 구조화된 데이터 추출
### 일반 텍스트에서 추출```python
# Direct text extraction
plaintext_report = """
The threat actors used spearphishing emails with malicious attachments.
After gaining access, they deployed Mimikatz to harvest credentials and
used RDP for lateral movement across the network.
"""
result = asyncio.run(run_agentic_v2_async(plaintext_report, {
"cache_llm_responses": True,
"single_pass_threshold": 500
}))
markdown_report = """
| Tool | Purpose |
|---|---|
| Mimikatz | Credential dumping |
| PsExec | Remote execution |
| Cobalt Strike | C2 communications |
| """ |
result = run_extraction_pipeline(markdown_report, { "use_optimized_extractor": True, "span_score_threshold": 0.7 }, source_id="markdown_report")
### PDF에서 추출```python
import pdfplumber
from bandjacks.llm.extraction_pipeline import run_extraction_pipeline
# Read PDF with pdfplumber (recommended)
with pdfplumber.open("threat_report.pdf") as pdf:
text = ""
for page in pdf.pages:
page_text = page.extract_text()
if page_text:
text += page_text + "\n"
# Extract techniques using extraction pipeline
result = run_extraction_pipeline(text, {
"use_optimized_extractor": True,
"span_score_threshold": 0.7,
"chunk_size": 2000
}, source_id="threat_report")
print(f"Found {len(result['techniques'])} techniques")
from pathlib import Path import json
reports_dir = Path("./reports") results = []
for pdf_file in reports_dir.glob("*.pdf"): # Extract text and techniques # ... (see above)
results.append({
"file": pdf_file.name,
"techniques": list(result["techniques"].keys()),
"count": len(result["techniques"])
})
with open("extraction_summary.json", "w") as f: json.dump(results, f, indent=2)
### 공격 흐름 구축```python
# Generate attack flow from extracted techniques
response = httpx.post(
"http://localhost:8000/v1/flows/build",
json={
"source_id": "report-123",
"technique_ids": ["T1566.001", "T1059.001", "T1003.001"]
}
)
flow = response.json()
print(f"Generated flow with {len(flow['steps'])} steps")
테스트 스위트를 실행하여 설치를 확인하십시오:```bash
uv run pytest
python tests/test_optimized_extraction.py
python tests/test_graph_upsert.py
python tests/test_bundle_validation.py
cd ui && npm test
## API 엔드포인트
### 핵심 엔드포인트
- `POST /v1/stix/load/attack` - MITRE ATT&CK 데이터 로드
- `POST /v1/reports/ingest` - 동기 보고서 수집 (<5KB)
- `POST /v1/reports/ingest_async` - 비동기 보고서 수집 (>5KB)
- `POST /v1/reports/ingest/upload` - PDF/TXT 파일 업로드
- `GET /v1/reports/jobs/{id}/status` - 작업 상태 확인
- `POST /v1/reports/{id}/unified-review` - 검토 결정 제출
- `POST /v1/search/ttx` - 기법 검색
- `GET /v1/graph/technique/{id}` - 기법 상세 정보 가져오기
### 공격 흐름
- `POST /v1/flows/build` - AttackFlow 동시발생 모델 생성
- `GET /v1/flows/{flow_id}` - 특정 AttackFlow 상세 정보 검색
- `POST /v1/flows/search` - 유사 공격 흐름 검색
- `GET /v1/flows/dump` - 페이지네이션 및 필터링을 통한 대량 흐름 내보내기
### 분석
- `GET /v1/analytics/cooccurrence/global` - 글로벌 동시발생 메트릭
- `GET /v1/analytics/cooccurrence/conditional` - 조건부 확률
- `GET /v1/analytics/cooccurrence/bundles` - 기법 번들
- `GET /v1/analytics/cooccurrence/actor` - 행위자별 패턴
- `GET /v1/coverage/gaps` - 기법 커버리지 갭
### 방어 및 탐지
- `GET /v1/defense/technique/{id}` - 방어 권장 사항 가져오기
- `GET /v1/detections/technique/{id}` - 탐지 전략
- `POST /v1/sigma/validate` - Sigma 규칙 검증
### 모니터링
- `GET /health` - 기본 상태 확인
- `GET /health/live` - Kubernetes 활성 프로브
- `GET /health/ready` - Kubernetes 준비 프로브
- `GET /health/components/{component}` - 개별 컴포넌트 상태
- `GET /v1/costs/stats` - LLM 비용 추적 (모델별 일일 집계)
- `GET /v1/cache/stats` - LLM 캐시 통계 가져오기
- `POST /v1/cache/clear` - LLM 캐시 지우기
- `GET /v1/compliance/report` - 규정 준수 메트릭
- `GET /v1/drift/status` - 드리프트 탐지 상태
- `GET /v1/ml-metrics/performance` - ML 모델 메트릭
### 행위자 및 출처
- `GET /v1/actors` - 위협 행위자 목록
- `GET /v1/actors/{id}` - 행위자 상세 정보 가져오기
- `GET /v1/provenance/{object_id}` - 객체 출처
- `GET /v1/provenance/{object_id}/lineage` - 전체 계보 체인
- `GET /v1/provenance/{object_id}/evidence` - 증거 스니펫
### API 전용 기능 (UI/CLI 없음)
이 엔드포인트는 완전히 기능하지만 REST API로만 접근 가능합니다 (프론트엔드 페이지나 CLI 명령어 없음):
#### 공격 경로 시뮬레이션
- `POST /v1/simulation/paths` - 시작 기법/그룹에서 공격 경로 시뮬레이션
- `POST /v1/simulation/predict` - 현재 상태에서 다음에 발생할 가능성이 높은 기법 예측
- `POST /v1/simulation/whatif` - 방어 시나리오에 대한 가상 분석
- `POST /v1/simulation/scenario` - 그룹/소프트웨어/기법 세트에서 시뮬레이션
- `GET /v1/simulation/statistics/{technique_id}` - 기법 사용 통계
- `GET /v1/simulation/groups/{group_id}/patterns` - 그룹 공격 패턴
- `POST /v1/simulation/compare` - 여러 공격 경로 비교
#### MDP 정책 및 롤아웃
- `POST /v1/simulate/rollout` - PTG 롤아웃 시뮬레이션
- `POST /v1/simulate/mdp` - MDP 최적 방어 정책 계산
- `GET /v1/simulate/models` - 사용 가능한 PTG 모델 목록
#### 드리프트 탐지 및 모니터링
- `GET /v1/drift/status` - 모든 메트릭에 대한 현재 드리프트 상태
- `POST /v1/drift/analyze` - 사용자 정의 임계값으로 드리프트 분석 실행
- `GET /v1/drift/alerts` - 활성 드리프트 알림 가져오기
- `POST /v1/drift/alerts/{alert_id}/acknowledge` - 알림 확인
- `GET /v1/drift/metrics/{metric_name}` - 특정 드리프트 메트릭 가져오기
#### ML 메트릭 추적
- `POST /v1/ml-metrics/prediction` - 추적을 위한 모델 예측 기록
- `POST /v1/ml-metrics/review` - 검토 결정 메트릭 기록
- `POST /v1/ml-metrics/coverage-gap` - 커버리지 갭 기록
- `GET /v1/ml-metrics/performance` - 모델 성능 메트릭 가져오기
- `GET /v1/ml-metrics/dashboard` - 대시보드 메트릭 내보내기
#### 알림
- `GET /v1/notifications/history` - 알림 기록 가져오기
- `POST /v1/notifications/clear-history` - 알림 기록 지우기
- `GET /v1/notifications/config` - 알림 설정 가져오기
- `POST /v1/notifications/test` - 테스트 알림 보내기
#### 벡터 업데이트 관리
- `GET /v1/vectors/status` - 벡터 업데이트 시스템 상태
- `GET /v1/vectors/metrics` - 상세 벡터 업데이트 메트릭
- `POST /v1/vectors/update` - 수동으로 벡터 업데이트 트리거
- `POST /v1/vectors/process-batch` - 강제 배치 처리
- `DELETE /v1/vectors/queue` - 보류 중인 업데이트 큐 지우기
- `GET /v1/vectors/health` - 벡터 시스템 상태 확인
#### 엔티티 무시 목록
- `GET /v1/ignorelist` - 현재 무시 목록 상태 가져오기
- `POST /v1/ignorelist/add` - 엔티티를 무시 목록에 추가
- `DELETE /v1/ignorelist/remove` - 엔티티를 무시 목록에서 제거
- `POST /v1/ignorelist/reload` - 디스크에서 무시 목록 다시 로드
#### 후보 패턴 검토
- `GET /v1/review/candidates` - 후보 공격 패턴 목록
- `POST /v1/review/candidates` - 후보 패턴 생성
- `GET /v1/review/candidates/{id}` - 후보 상세 정보 가져오기
- `POST /v1/review/candidates/{id}/approve` - 후보 승인
- `POST /v1/review/candidates/{id}/reject` - 후보 거부
- `GET /v1/review/candidates/{id}/similar` - 유사 패턴 찾기
- `GET /v1/review/candidates/stats/summary` - 후보 통계
### 전체 API 문서
전체 API 문서를 확인하세요:
- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc
- OpenAPI JSON: http://localhost:8000/openapi.json
## 아키텍처
### 프로젝트 구조```
bandjacks/
├── bandjacks/
│ ├── analysis/ # Graph analysis & interdiction
│ │ ├── graph_analyzer.py
│ │ └── interdiction.py
│ ├── analytics/ # Co-occurrence & clustering
│ │ ├── clustering.py
│ │ ├── cooccurrence.py
│ │ └── detection_bundles.py
│ ├── cli/ # Command-line interface
│ │ ├── main.py # CLI entry point
│ │ ├── batch_extract.py
│ │ ├── formatters.py
│ │ └── workflows.py
│ ├── config/ # Configuration files
│ │ └── entity_ignorelist.yaml
│ ├── core/ # Core utilities
│ │ ├── cache.py # Redis caching
│ │ ├── connection_pool.py
│ │ └── query_optimizer.py
│ ├── llm/ # Extraction pipeline
│ │ ├── extraction_pipeline.py
│ │ ├── agents_v2.py # Core extraction agents
│ │ ├── chunked_extractor.py
│ │ ├── optimized_chunked_extractor.py
│ │ ├── entity_extractor.py
│ │ ├── flow_builder.py
│ │ ├── cache.py # LLM response caching
│ │ └── experimental/ # Experimental features
│ ├── loaders/ # Data loading & indexing
│ │ ├── attack_catalog.py
│ │ ├── attack_upsert.py
│ │ ├── opensearch_index.py
│ │ ├── hybrid_search.py
│ │ └── sigma_loader.py
│ ├── monitoring/ # Metrics & monitoring
│ │ ├── compliance_metrics.py
│ │ ├── defense_metrics.py
│ │ ├── drift_detector.py
│ │ └── ml_metrics.py
│ ├── services/ # API & services
│ │ ├── api/ # FastAPI application
│ │ │ ├── main.py
│ │ │ ├── routes/ # API route handlers
│ │ │ └── middleware/
│ │ ├── technique_cache.py
│ │ └── actor_cache.py
│ ├── simulation/ # Attack simulation
│ │ ├── attack_simulator.py
│ │ ├── mdp_solver.py
│ │ └── ptg_rollout.py
│ └── store/ # Data stores
│ ├── report_store.py
│ ├── candidate_store.py
│ └── review_store.py
├── ui/ # Next.js frontend
│ ├── app/ # App Router pages
│ │ ├── reports/ # Report management
│ │ ├── analytics/ # Analytics dashboards
│ │ └── health/ # Health monitoring
│ ├── components/ # React components
│ └── hooks/ # Custom React hooks
├── tests/ # Test suite
├── samples/ # Sample reports
├── scripts/ # Utility scripts
└── docs/ # Documentation
추출 파이프라인 (bandjacks/llm/)
extraction_pipeline.py - 주요 추출 오케스트레이터chunked_extractor.py - 표준 청크 단위 처리optimized_chunked_extractor.py - 고급 최적화 처리agents_v2.py - 핵심 추출 에이전트 (SpanFinder, Mapper, Consolidator)entity_extractor.py - 개체 인식 에이전트flow_builder.py - 공격 흐름 생성memory.py - 공유 작업 메모리cache.py - LLM 응답 캐싱데이터 계층 (bandjacks/loaders/)
API 계층 (bandjacks/services/api/)
시스템은 클라우드 LLM과 모든 로컬 OpenAI 호환 API를 지원합니다:```bash
LOCAL_LLM_API_BASE=http://192.168.1.100:8080/v1 LOCAL_LLM_MODEL=mistral-nemo LOCAL_LLM_API_KEY=no-key # optional — most local servers don't require a key
PRIMARY_LLM=gemini # "gemini" (default) or "openai" GOOGLE_API_KEY=your-key # Gemini OPENAI_API_KEY=your-key # OpenAI (used as fallback when Gemini is primary)
**공급자 우선순위:** Local API > Gemini > OpenAI > LiteLLM proxy.
로컬 서버가 설정되면 클라우드 공급자가 자동으로 대체 수단으로 추가됩니다.
#### 일반적인 로컬 서버 예시
| 서버 | `LOCAL_LLM_API_BASE` | `LOCAL_LLM_MODEL` |
|--------|---------------------|-------------------|
| vLLM | `http://host:8000/v1` | `mistralai/Mistral-Nemo-Instruct-2407` |
| llama.cpp | `http://host:8080/v1` | `mistral-nemo` |
| Ollama | `http://host:11434/v1` | `mistral-nemo` |
| LM Studio | `http://host:1234/v1` | `mistral-nemo` |
| LocalAI | `http://host:8080/v1` | `mistral-nemo` |
### 추출 구성
시스템은 구성 가능한 옵션이 있는 단일 고성능 비동기 파이프라인을 사용합니다:```python
{
"cache_llm_responses": True, # Enable LLM caching (default: True)
"single_pass_threshold": 500, # Max words for single-pass (default: 500)
"early_termination_confidence": 90, # Skip verification above this (default: 90)
"disable_discovery": False, # Disable LLM discovery agent
"max_spans": 20, # Maximum spans to process
"span_score_threshold": 0.7, # Minimum span quality
"top_k": 5, # Candidates per span
# Cost optimization options
"max_spans_per_technique": 2, # Pre-filter: max spans per candidate technique (0=disable, default=2)
"enable_span_dedup": False, # Text-based span dedup before mapping (default=False)
}
추출 파이프라인은 litellm.completion_cost()를 통해 LLM 비용을 추적하며, 보고서별 메트릭과 일일 집계 엔드포인트를 제공합니다.
비용 제어:
모니터링:```bash
curl http://localhost:8000/v1/costs/stats
curl http://localhost:8000/v1/reports/{id} # -> extraction.metrics.cost_usd
### 신뢰 임계값
추출 품질 제어:```python
{
"confidence_threshold": 50.0, # Minimum confidence (0-100)
"auto_ingest": True # Auto-add high-confidence results
}
curl http://localhost:8000/health
curl http://localhost:8000/health/live
curl http://localhost:8000/health/ready
curl http://localhost:8000/health/components/neo4j curl http://localhost:8000/health/components/opensearch curl http://localhost:8000/health/components/redis curl http://localhost:8000/health/components/caches curl http://localhost:8000/health/components/system
### 헬스 응답 예시```json
{
"status": "healthy",
"timestamp": "2025-01-28T17:43:30.184036Z",
"version": "1.0.0",
"components": {
"neo4j": {
"status": "healthy",
"latency_ms": 5
},
"opensearch": {
"status": "degraded",
"cluster_status": "yellow",
"indices": {
"attack_nodes": false,
"bandjacks_reports": true
}
},
"redis": {
"status": "healthy",
"latency_ms": 2,
"memory_mb": 1.69
},
"caches": {
"status": "healthy",
"technique_cache": {
"count": 993,
"loaded": true
},
"actor_cache": {
"count": 145,
"loaded": true
}
},
"system": {
"status": "healthy",
"memory": {
"available_gb": 8.84,
"percent_used": 72.4
},
"disk": {
"available_gb": 353.11,
"percent_used": 2.9
},
"cpu": {
"percent_used": 7.7
}
}
}
}
Kubernetes 배포의 경우 프로브를 다음과 같이 구성하십시오:```yaml livenessProbe: httpGet: path: /health/live port: 8000 initialDelaySeconds: 30 periodSeconds: 10
readinessProbe: httpGet: path: /health/ready port: 8000 initialDelaySeconds: 45 periodSeconds: 5
## 성능 최적화
### 캐싱
시스템은 향상된 성능을 위해 자동 LLM 응답 캐싱을 포함합니다:```python
# Check cache statistics
response = httpx.get("http://localhost:8000/v1/cache/stats")
stats = response.json()
print(f"Cache hit rate: {stats['hit_rate']}")
# Clear cache if needed
httpx.post("http://localhost:8000/v1/cache/clear")
필요에 따라 프로필을 선택하세요:```python
fast_config = { "single_pass_threshold": 1000, "max_spans": 5, "skip_verification": True, "top_k": 3 }
balanced_config = { "single_pass_threshold": 500, "max_spans": 10, "early_termination_confidence": 90, "top_k": 5 }
quality_config = { "single_pass_threshold": 200, "max_spans": 20, "disable_discovery": False, "min_quotes": 3, "top_k": 10 }
## Security
### Input Validation
- **Cypher Injection Prevention**: 모든 그래프 쿼리 엔드포인트는 사용자 제공 `relationship_types` 매개변수를 알려진 관계 유형(USES, MITIGATES, HAS_TACTIC 등)의 허용 목록과 엄격한 정규식 패턴(`^[A-Z][A-Z0-9_]*$`)에 대해 검증합니다. 유효하지 않은 입력은 쿼리 구성 전에 400을 반환합니다.
- **JSON Schema Validation**: LLM 응답은 JSON 스키마에 대해 검증되어 잘못된 데이터가 파이프라인에 유입되는 것을 방지합니다.
- **ADM Validation**: 모든 STIX 콘텐츠는 수집 전에 ATT&CK 데이터 모델 검증을 통과해야 합니다.
### Authentication & Authorization
- **JWT Authentication**: API 인증을 위한 선택적 미들웨어 (`JWTAuthMiddleware`)
- **Rate Limiting**: 구성 가능한 임계값을 사용한 엔드포인트별 속도 제한
- **CORS**: 구성 가능한 교차 출처 리소스 공유
## Advanced Features
### Provenance Tracking
추출된 모든 엔티티는 전체 출처를 포함합니다:```python
# Get provenance for an object
response = httpx.get(
"http://localhost:8000/v1/provenance/attack-pattern--abc123"
)
시스템에는 추출 개선을 위한 검토 큐가 포함되어 있습니다:```python
response = httpx.get("http://localhost:8000/v1/review_queue/next")
response = httpx.post( "http://localhost:8000/v1/feedback/extraction", json={ "extraction_id": "ext-123", "correct": True, "corrections": [] } )
### 커버리지 분석
위협 인텔리전스 커버리지를 분석하세요:```python
# Get coverage analysis
response = httpx.get("http://localhost:8000/v1/analytics/coverage")
coverage = response.json()
print(f"Summary: {coverage['summary']}")
for tactic in coverage['tactics']:
print(f" {tactic['tactic']}: {tactic['coverage_percentage']}%")
참고: 플랫폼 커버리지 (
_analyze_platforms_coverage) 현재는 플레이스홀더 데이터를 반환합니다. 전술 및 그룹 커버리지는 실제 Neo4j 쿼리를 사용합니다.
시뮬레이션 모듈은 MDP 기반의 공격 경로 예측을 제공합니다:```python
from bandjacks.simulation.attack_simulator import AttackSimulator from bandjacks.simulation.mdp_solver import MDPSolver
## 기능 상태
이 섹션에서는 다양한 기능의 구현 상태를 투명하게 제공합니다:
### 완전히 기능함 ✅
- **보고서 추출 파이프라인** - LLM 기반 기술 추출이 종단간 작동
- **MITRE ATT&CK 로딩** - 엔터프라이즈/모바일/ICS ATT&CK 데이터를 Neo4j로 로드
- **벡터 검색** - 기술에 대한 OpenSearch 기반 시맨틱 검색
- **검토 시스템** - API 및 UI를 통한 인간-루프 검토 워크플로
- **상태 모니터링** - 구성 요소 상태 확인 및 Kubernetes 프로브
- **CLI 쿼리/관리 명령** - 검색, 그래프 탐색, 캐시 관리
- **공격 흐름 생성** - 침입 집합에 대한 공동 발생 기반 흐름 구축
- **공격 시뮬레이션** - `/simulation/*` 및 `/simulate/*`를 통한 MDP 기반 경로 시뮬레이션
- **커버리지 보고서** - 경영진, 기술, 전술, 운영 관점을 위한 JSON 보고서
### 데이터 종속성 있음 ⚠️
- **공동 발생 분석** - 보고서 처리에서 `AttackEpisode` 노드 필요
- **행위자 분석** - 침입 집합에 귀속된 에피소드 필요
- **기술 번들** - 패턴 마이닝에 충분한 에피소드 데이터 필요
- **CLI 분석 명령** - 작동하지만 에피소드가 없으면 빈 결과 반환
### API 전용 (UI/CLI 없음) 🔌
REST API를 통해서만 접근 가능한 완전히 구현된 기능:
- **공격 경로 시뮬레이션** - 경로 예측 및 가상 분석을 위한 `/simulation/*` 경로
- **MDP 정책 솔버** - 최적 방어 정책 계산을 위한 `/simulate/mdp`
- **드리프트 탐지** - 데이터 품질 드리프트 모니터링을 위한 `/drift/*` 경로
- **ML 메트릭** - 시간 경과에 따른 모델 성능 추적을 위한 `/ml-metrics/*`
- **벡터 관리** - 벡터 임베딩 관리를 위한 `/vectors/*`
- **엔티티 무시 목록** - 거짓 양성 엔티티 필터링을 위한 `/ignorelist/*`
- **후보 패턴** - 새로운 기술 후보를 위한 `/review/candidates/*`
- **알림** - 알림 구성 및 기록을 위한 `/notifications/*`
- **이력 추적** - 추출 계보 추적을 위한 `/provenance/*`
- **규정 준수** - 규정 준수 메트릭 보고를 위한 `/compliance/*`
### 실험적 ( `llm/experimental/` 내) 🧪
- **PTG (확률적 위협 그래프)** - 핵심 로직 구현, 제한적 테스트
- **Judge 통합** - LLM 기반 시퀀스 검증
- **공격 흐름 시뮬레이터** - 흐름 기반 시뮬레이션 엔진
- **시퀀스 추출기** - 흐름에서 시퀀스 추출
### 제거/정리됨 🗑️
다음 스텁 기능이 API에서 제거되었습니다:
- ~~플랫폼 커버리지 분석~~ - 하드코딩된 스텁 데이터를 반환했음
- ~~추세 분석~~ - 무작위 합성 데이터를 반환했음
- ~~CSV/PDF 보고서 내보내기~~ - 501을 반환했음; 현재는 JSON 전용
- ~~Gemini 시퀀스 추론~~ - 501 스텁이었음; 대신 `/sequence/propose` 사용
### 연결 매트릭스
| 기능 영역 | 프론트엔드 UI | CLI | REST API |
|--------------|-------------|-----|----------|
| 보고서 관리 | ✅ | ✅ | ✅ |
| 검토 워크플로 | ✅ | ✅ | ✅ |
| 검색 (TTX) | ✅ | ✅ | ✅ |
| 공동 발생 분석 | ✅ | ✅ | ✅ |
| 커버리지 분석 | ✅ | - | ✅ |
| 상태 모니터링 | ✅ | - | ✅ |
| 탐지/Sigma | ✅ | - | ✅ |
| 공격 흐름 | ✅ | - | ✅ |
| 방어 오버레이 | ✅ | - | ✅ |
| 시퀀스/PTG | ✅ | - | ✅ |
| 행위자 | ✅ | - | ✅ |
| 공격 시뮬레이션 | - | - | ✅ |
| 드리프트 탐지 | - | - | ✅ |
| ML 메트릭 | - | - | ✅ |
| 벡터 관리 | - | - | ✅ |
| 엔티티 무시 목록 | - | - | ✅ |
| 후보 패턴 | - | - | ✅ |
| 알림 | - | - | ✅ |
| 이력 추적 | - | - | ✅ |
| 규정 준수 | - | - | ✅ |
### 프론트엔드 페이지
| 페이지 | 상태 | 참고 |
|------|--------|-------|
| `/reports` | ✅ 작동 중 | 보고서 목록, 생성, 보기 |
| `/reports/[id]/review` | ✅ 작동 중 | 전체 검토 워크플로 |
| `/analytics/cooccurrence` | ⚠️ 데이터 종속 | 에피소드가 존재하면 KPI 표시 |
| `/analytics/cooccurrence/pairs` | ⚠️ 데이터 종속 | 실제 API 호출 |
| `/analytics/cooccurrence/bundles` | ⚠️ 데이터 종속 | 실제 API 호출 |
| `/analytics/cooccurrence/actors` | ⚠️ 데이터 종속 | 실제 API 호출 |
| `/health` | ✅ 작동 중 | 실시간 상태 |
## 문제 해결
### 일반적인 문제
1. **OpenSearch 연결 실패**
- OpenSearch가 실행 중인지 확인: `curl http://localhost:9200`
- 인덱스 존재 여부 확인: `curl http://localhost:9200/bandjacks_attack_nodes-v1`
2. **Neo4j 연결 실패**
- Neo4j가 실행 중인지 확인: `neo4j status`
- `.env` 파일에 `NEO4J_PASSWORD`가 설정되었는지 확인
- 비밀번호가 Neo4j 인스턴스와 일치하는지 확인
- "NEO4J_PASSWORD 환경 변수가 필요합니다"라는 메시지가 표시되면 `.env` 파일에 설정해야 합니다.
3. **낮은 추출 재현율**
- `agentic_v2` 방식을 사용하고 있는지 확인
- LLM API 키가 유효한지 확인
- 모델 이름이 올바른지 확인 (gemini-flash-latest)
4. **시간 초과 오류**
- 큰 문서에 대해 시간 초과 설정 증가
- 매우 큰 보고서는 청킹 고려
5. **프론트엔드가 API에 연결되지 않음**
- API가 포트 8000에서 실행 중인지 확인
- API 구성에서 CORS 설정 확인
### 디버그 모드
상세 로깅 활성화:```python
import logging
logging.basicConfig(level=logging.DEBUG)
# Run extraction with debug output
result = run_agentic_v2(text, config)
uv run pytest tests/unit
uv run pytest tests/integration
uv run pytest tests/test_agentic_v2.py::test_extraction
uv run pytest --cov=bandjacks
cd ui && npm test cd ui && npm run test:coverage
### 기여하기
1. 저장소 포크하기
2. 기능 브랜치 생성하기
3. 변경 사항 적용하기
4. 테스트 실행: `uv run pytest`
5. 린팅 실행: `uv run ruff check`
6. 풀 리퀘스트 제출하기
### 코드 품질```bash
# Format code
uv run ruff format
# Check linting
uv run ruff check
# Type checking
uv run mypy bandjacks
[여기에 라이선스 입력]
프론트엔드 (ui/)
| 옵션 | 기본값 | 효과 | 품질 영향 |
|---|
MAX_MAPPER_BATCH_SIZE (환경 변수) | 10 | LLM 매퍼 호출당 스팬 수 (2026-05에 25에서 낮춤; 클라우드 응답이 ~800 토큰으로 제한되어, 더 큰 배치의 ~12%가 잘린 JSON을 반환함) | 없음 |
max_spans_per_technique (설정) | 2 | 사전 필터: 후보 기술당 최상의 N개 스팬 | ~19% 적은 기술, 더 높은 신뢰도 |
enable_span_dedup (설정) | false | 매핑 전 중복 스팬 텍스트 제거 | ~15% 적은 기술 |