n0mad1k
397 lines
15 KiB
Python
397 lines
15 KiB
Python
"""Generate structured Markdown field manual from Mosaic extracted intelligence."""
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from collections import defaultdict
|
|
|
|
from storage.db import MosaicDB
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Domain display order and titles
|
|
TRADECRAFT_DOMAINS = {
|
|
'humint': 'HUMINT Operations',
|
|
'surveillance': 'Surveillance',
|
|
'counter_surveillance': 'Counter-Surveillance',
|
|
'covert_comms': 'Covert Communications',
|
|
'physical_access': 'Physical Access & Entry',
|
|
'escape_evasion': 'Escape & Evasion',
|
|
'cyber': 'Cyber Operations',
|
|
}
|
|
|
|
TTP_CATEGORIES = {
|
|
'reconnaissance': 'Reconnaissance',
|
|
'resource_development': 'Resource Development',
|
|
'initial_access': 'Initial Access',
|
|
'execution': 'Execution',
|
|
'persistence': 'Persistence',
|
|
'privilege_escalation': 'Privilege Escalation',
|
|
'defense_evasion': 'Defense Evasion',
|
|
'credential_access': 'Credential Access',
|
|
'discovery': 'Discovery',
|
|
'lateral_movement': 'Lateral Movement',
|
|
'collection': 'Collection',
|
|
'exfiltration': 'Exfiltration',
|
|
'command_and_control': 'Command & Control',
|
|
'impact': 'Impact',
|
|
'counter_surveillance': 'Counter-Surveillance',
|
|
'covert_comms': 'Covert Communications',
|
|
'physical_access': 'Physical Access',
|
|
'humint': 'HUMINT',
|
|
}
|
|
|
|
TOOL_CAPABILITIES = {
|
|
'implant': 'Implants & Backdoors',
|
|
'framework': 'Frameworks',
|
|
'exploit': 'Exploits',
|
|
'credential_theft': 'Credential Theft',
|
|
'collection': 'Collection Tools',
|
|
'evasion': 'Evasion & Anti-Forensics',
|
|
'rootkit': 'Rootkits & Covert Storage',
|
|
'router_exploit': 'Network Device Exploitation',
|
|
'proxy': 'Proxies & Redirectors',
|
|
'c2_framework': 'C2 Frameworks',
|
|
'obfuscation': 'Obfuscation & Attribution',
|
|
'spyware': 'Spyware',
|
|
'weapon': 'Weapons Systems',
|
|
'utility': 'Utilities',
|
|
'trojan': 'Trojans',
|
|
'platform': 'Multi-Component Platforms',
|
|
}
|
|
|
|
|
|
def generate_field_manual(db: MosaicDB, output_path: str = "output/FIELD_MANUAL.md",
|
|
title: str = "Operator's Field Manual") -> str:
|
|
"""Generate the complete field manual from database contents.
|
|
|
|
Returns the path to the generated file.
|
|
"""
|
|
tools = db.get_all_tools()
|
|
ttps = db.get_all_ttps()
|
|
tradecraft = db.get_all_tradecraft()
|
|
entities = db.get_all_entities()
|
|
infrastructure = db.get_all_infrastructure()
|
|
|
|
# Get corpus stats
|
|
stats = db.get_status_counts()
|
|
token_stats = db.get_total_tokens()
|
|
|
|
sections = []
|
|
|
|
# --- HEADER ---
|
|
sections.append(_header(title, stats, len(tools), len(ttps), len(tradecraft), len(entities), len(infrastructure)))
|
|
|
|
# --- TABLE OF CONTENTS ---
|
|
sections.append(_toc(tools, ttps, tradecraft, entities, infrastructure))
|
|
|
|
# --- PART I: TRADECRAFT ---
|
|
if tradecraft:
|
|
sections.append("\n---\n\n# Part I: Tradecraft\n")
|
|
by_domain = defaultdict(list)
|
|
for tc in tradecraft:
|
|
by_domain[tc.domain].append(tc)
|
|
|
|
for domain_key, domain_title in TRADECRAFT_DOMAINS.items():
|
|
items = by_domain.get(domain_key, [])
|
|
if not items:
|
|
# Check for unlisted domains
|
|
continue
|
|
sections.append(f"\n## {domain_title}\n")
|
|
for tc in sorted(items, key=lambda x: x.method):
|
|
sections.append(_format_tradecraft(tc))
|
|
|
|
# Any domains not in our predefined list
|
|
for domain_key, items in by_domain.items():
|
|
if domain_key not in TRADECRAFT_DOMAINS:
|
|
sections.append(f"\n## {domain_key.replace('_', ' ').title()}\n")
|
|
for tc in sorted(items, key=lambda x: x.method):
|
|
sections.append(_format_tradecraft(tc))
|
|
|
|
# --- PART II: TTPs ---
|
|
if ttps:
|
|
sections.append("\n---\n\n# Part II: Techniques, Tactics & Procedures\n")
|
|
by_category = defaultdict(list)
|
|
for ttp in ttps:
|
|
by_category[ttp.category].append(ttp)
|
|
|
|
for cat_key, cat_title in TTP_CATEGORIES.items():
|
|
items = by_category.get(cat_key, [])
|
|
if not items:
|
|
continue
|
|
sections.append(f"\n## {cat_title}\n")
|
|
for ttp in sorted(items, key=lambda x: x.technique):
|
|
sections.append(_format_ttp(ttp))
|
|
|
|
# Unlisted categories
|
|
for cat_key, items in by_category.items():
|
|
if cat_key not in TTP_CATEGORIES:
|
|
sections.append(f"\n## {cat_key.replace('_', ' ').title()}\n")
|
|
for ttp in sorted(items, key=lambda x: x.technique):
|
|
sections.append(_format_ttp(ttp))
|
|
|
|
# --- PART III: TOOLS ---
|
|
if tools:
|
|
sections.append("\n---\n\n# Part III: Tools & Capabilities\n")
|
|
by_cap = defaultdict(list)
|
|
for tool in tools:
|
|
by_cap[tool.capability].append(tool)
|
|
|
|
for cap_key, cap_title in TOOL_CAPABILITIES.items():
|
|
items = by_cap.get(cap_key, [])
|
|
if not items:
|
|
continue
|
|
sections.append(f"\n## {cap_title}\n")
|
|
for tool in sorted(items, key=lambda x: x.name):
|
|
sections.append(_format_tool(tool))
|
|
|
|
# Unlisted capabilities
|
|
for cap_key, items in by_cap.items():
|
|
if cap_key not in TOOL_CAPABILITIES:
|
|
sections.append(f"\n## {cap_key.replace('_', ' ').title()}\n")
|
|
for tool in sorted(items, key=lambda x: x.name):
|
|
sections.append(_format_tool(tool))
|
|
|
|
# --- PART IV: ENTITIES ---
|
|
if entities:
|
|
sections.append("\n---\n\n# Part IV: Entities & Organizations\n")
|
|
by_type = defaultdict(list)
|
|
for entity in entities:
|
|
by_type[entity.type].append(entity)
|
|
|
|
for etype in ['unit', 'org', 'program', 'person', 'codename']:
|
|
items = by_type.get(etype, [])
|
|
if not items:
|
|
continue
|
|
sections.append(f"\n## {etype.title()}s\n")
|
|
for entity in sorted(items, key=lambda x: x.name):
|
|
sections.append(_format_entity(entity))
|
|
|
|
# Unlisted types
|
|
for etype, items in by_type.items():
|
|
if etype not in ['unit', 'org', 'program', 'person', 'codename']:
|
|
sections.append(f"\n## {etype.title()}\n")
|
|
for entity in sorted(items, key=lambda x: x.name):
|
|
sections.append(_format_entity(entity))
|
|
|
|
# --- PART V: INFRASTRUCTURE ---
|
|
if infrastructure:
|
|
sections.append("\n---\n\n# Part V: Infrastructure Indicators\n")
|
|
by_type = defaultdict(list)
|
|
for infra in infrastructure:
|
|
by_type[infra.indicator_type].append(infra)
|
|
|
|
for itype in ['ipv4', 'ipv6', 'domain', 'url', 'hash', 'email']:
|
|
items = by_type.get(itype, [])
|
|
if not items:
|
|
continue
|
|
sections.append(f"\n## {itype.upper()} Indicators\n")
|
|
sections.append("| Indicator | Context | Source | Date |\n|---|---|---|---|\n")
|
|
for infra in sorted(items, key=lambda x: x.indicator):
|
|
ctx = infra.context[:80].replace('|', '/').replace('\n', ' ')
|
|
sections.append(f"| `{infra.indicator}` | {ctx} | {infra.collection} | {infra.doc_date or 'N/A'} |\n")
|
|
|
|
# --- APPENDIX: CROSS-REFERENCES ---
|
|
sections.append("\n---\n\n# Appendix A: Cross-Reference Index\n")
|
|
sections.append(_cross_reference_index(tools, ttps, tradecraft, entities))
|
|
|
|
# --- APPENDIX: SOURCES ---
|
|
sections.append("\n---\n\n# Appendix B: Sources\n")
|
|
sections.append(_sources_appendix(db))
|
|
|
|
# Write output
|
|
content = '\n'.join(sections)
|
|
out = Path(output_path)
|
|
out.parent.mkdir(parents=True, exist_ok=True)
|
|
out.write_text(content)
|
|
|
|
logger.info("Field manual generated: %s (%d chars)", output_path, len(content))
|
|
return str(out)
|
|
|
|
|
|
def _header(title, stats, tool_count, ttp_count, tc_count, ent_count, infra_count):
|
|
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
|
|
total_docs = stats.get('total', 0)
|
|
parsed = stats.get('parsing', {}).get('PARSED', 0)
|
|
|
|
return f"""# {title}
|
|
|
|
> Generated by **Mosaic** — Intelligence Extraction Platform
|
|
> Date: {now}
|
|
> Corpus: {total_docs} documents ({parsed} parsed)
|
|
> Extracted: {tool_count} tools, {ttp_count} TTPs, {tc_count} tradecraft, {ent_count} entities, {infra_count} infrastructure indicators
|
|
|
|
---
|
|
|
|
**Classification:** This manual is compiled from publicly available, declassified, and leaked sources.
|
|
Source provenance is noted per entry. Entries sourced from leaked materials (Vault 7, Vault 8) retain
|
|
their original classification markings for reference only.
|
|
|
|
**Purpose:** Operational reference for cyber and physical tradecraft, techniques, tactics, and
|
|
procedures. Covers the full spectrum from HUMINT field operations to cyber implant deployment.
|
|
|
|
"""
|
|
|
|
|
|
def _toc(tools, ttps, tradecraft, entities, infrastructure):
|
|
lines = ["## Table of Contents\n"]
|
|
lines.append("- **Part I: Tradecraft** — Operational methods across all domains")
|
|
lines.append("- **Part II: TTPs** — Techniques mapped to MITRE ATT&CK where applicable")
|
|
lines.append("- **Part III: Tools** — Named tools, implants, exploits, and frameworks")
|
|
lines.append("- **Part IV: Entities** — Organizations, programs, and codenames")
|
|
if infrastructure:
|
|
lines.append("- **Part V: Infrastructure** — Network indicators with provenance")
|
|
lines.append("- **Appendix A: Cross-Reference Index**")
|
|
lines.append("- **Appendix B: Sources**")
|
|
return '\n'.join(lines) + '\n'
|
|
|
|
|
|
def _format_tradecraft(tc):
|
|
lines = [f"\n### {tc.method}\n"]
|
|
lines.append(f"**Domain:** {tc.domain.replace('_', ' ').title()}\n")
|
|
lines.append(f"{tc.description}\n")
|
|
|
|
if tc.operational_notes:
|
|
lines.append(f"\n**Operational Notes:**\n{tc.operational_notes}\n")
|
|
|
|
if tc.countermeasures:
|
|
lines.append(f"\n**Countermeasures:**\n{tc.countermeasures}\n")
|
|
|
|
if tc.source_context:
|
|
# Truncate for readability
|
|
ctx = tc.source_context[:300]
|
|
lines.append(f"\n> *Source: {ctx}*\n")
|
|
|
|
return '\n'.join(lines)
|
|
|
|
|
|
def _format_ttp(ttp):
|
|
lines = [f"\n### {ttp.technique}\n"]
|
|
|
|
tags = []
|
|
if ttp.mitre_id:
|
|
tags.append(f"MITRE: **{ttp.mitre_id}**")
|
|
if ttp.actor:
|
|
tags.append(f"Actor: **{ttp.actor}**")
|
|
tags.append(f"Category: **{ttp.category.replace('_', ' ').title()}**")
|
|
lines.append(' | '.join(tags) + '\n')
|
|
|
|
lines.append(f"\n{ttp.description}\n")
|
|
|
|
if ttp.source_context:
|
|
ctx = ttp.source_context[:300]
|
|
lines.append(f"\n> *{ctx}*\n")
|
|
|
|
return '\n'.join(lines)
|
|
|
|
|
|
def _format_tool(tool):
|
|
lines = [f"\n### {tool.name}\n"]
|
|
|
|
if tool.aliases:
|
|
lines.append(f"**Aliases:** {', '.join(tool.aliases)}\n")
|
|
|
|
lines.append(f"**Capability:** {tool.capability.replace('_', ' ').title()}")
|
|
lines.append(f"**Platforms:** {', '.join(tool.target_platforms) if tool.target_platforms else 'Unknown'}")
|
|
|
|
if tool.target_software:
|
|
lines.append(f"**Targets:** {', '.join(tool.target_software)}")
|
|
|
|
if tool.mitre_techniques:
|
|
lines.append(f"**MITRE:** {', '.join(tool.mitre_techniques)}")
|
|
|
|
if tool.cves:
|
|
lines.append(f"**CVEs:** {', '.join(tool.cves)}")
|
|
|
|
lines.append(f"\n{tool.description}\n")
|
|
|
|
if tool.source_context:
|
|
ctx = tool.source_context[:300]
|
|
lines.append(f"\n> *{ctx}*\n")
|
|
|
|
return '\n'.join(lines)
|
|
|
|
|
|
def _format_entity(entity):
|
|
lines = [f"\n### {entity.name}\n"]
|
|
lines.append(f"**Type:** {entity.type.title()}\n")
|
|
lines.append(f"{entity.description}\n")
|
|
|
|
if entity.relationships:
|
|
lines.append("\n**Relationships:**")
|
|
for rel in entity.relationships:
|
|
lines.append(f"- {rel.get('relationship_type', 'related to').replace('_', ' ')}: **{rel.get('entity', 'Unknown')}**")
|
|
lines.append("")
|
|
|
|
if entity.source_context:
|
|
ctx = entity.source_context[:200]
|
|
lines.append(f"\n> *{ctx}*\n")
|
|
|
|
return '\n'.join(lines)
|
|
|
|
|
|
def _cross_reference_index(tools, ttps, tradecraft, entities):
|
|
"""Build alphabetical index of all terms with section references."""
|
|
lines = ["\nAlphabetical index of techniques, tools, and tradecraft:\n"]
|
|
|
|
all_items = []
|
|
for t in tools:
|
|
all_items.append((t.name, "Tool", t.capability.replace('_', ' ').title()))
|
|
for alias in t.aliases:
|
|
all_items.append((alias, "Tool (alias)", f"See {t.name}"))
|
|
for t in ttps:
|
|
all_items.append((t.technique, "TTP", t.category.replace('_', ' ').title()))
|
|
for t in tradecraft:
|
|
all_items.append((t.method, "Tradecraft", t.domain.replace('_', ' ').title()))
|
|
for e in entities:
|
|
all_items.append((e.name, "Entity", e.type.title()))
|
|
|
|
all_items.sort(key=lambda x: x[0].lower())
|
|
|
|
lines.append("| Term | Type | Category/Domain |")
|
|
lines.append("|---|---|---|")
|
|
for name, itype, category in all_items:
|
|
lines.append(f"| {name} | {itype} | {category} |")
|
|
|
|
return '\n'.join(lines) + '\n'
|
|
|
|
|
|
def _sources_appendix(db):
|
|
"""List all sources with document counts."""
|
|
lines = ["\nDocument sources used to compile this manual:\n"]
|
|
|
|
cursor = db.conn.execute("""
|
|
SELECT source, COUNT(*) as cnt,
|
|
SUM(CASE WHEN parse_status='PARSED' THEN 1 ELSE 0 END) as parsed
|
|
FROM documents
|
|
GROUP BY source
|
|
ORDER BY source
|
|
""")
|
|
|
|
lines.append("| Source | Documents | Parsed | Description |")
|
|
lines.append("|---|---|---|---|")
|
|
|
|
source_descriptions = {
|
|
'vault7': 'CIA Hacking Tools (WikiLeaks Vault 7) — LEAKED',
|
|
'vault8': 'CIA Source Code (WikiLeaks Vault 8) — LEAKED',
|
|
'cablegate': 'US Diplomatic Cables (WikiLeaks Cablegate) — LEAKED',
|
|
'spyfiles': 'Surveillance Industry Documents (WikiLeaks Spy Files) — LEAKED',
|
|
'emails': 'Email Dumps (WikiLeaks) — LEAKED',
|
|
'irp_fas': 'FAS Intelligence Resource Program — DECLASSIFIED/PUBLIC',
|
|
'army_pubs': 'US Army Field Manuals & Regulations — DECLASSIFIED/PUBLIC',
|
|
'grugq': 'The Grugq OPSEC & Tradecraft Writings — PUBLIC',
|
|
'cia_rdp': 'CIA Reading Room — DECLASSIFIED',
|
|
'cryptome': 'Cryptome Document Archive — MIXED',
|
|
}
|
|
|
|
for row in cursor.fetchall():
|
|
desc = source_descriptions.get(row['source'], 'Custom source')
|
|
lines.append(f"| {row['source']} | {row['cnt']} | {row['parsed']} | {desc} |")
|
|
|
|
lines.append("\n\n**Provenance Key:**")
|
|
lines.append("- **LEAKED** — Unauthorized disclosure. Original classification markings preserved for reference.")
|
|
lines.append("- **DECLASSIFIED** — Officially released through FOIA or mandatory declassification review.")
|
|
lines.append("- **PUBLIC** — Published by author, open source intelligence analysis.")
|
|
|
|
return '\n'.join(lines) + '\n'
|