diff --git a/analysis/field_manual.py b/analysis/field_manual.py new file mode 100644 index 0000000..bbe26c5 --- /dev/null +++ b/analysis/field_manual.py @@ -0,0 +1,396 @@ +"""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' diff --git a/mosaic.py b/mosaic.py index 84088e8..bb45409 100755 --- a/mosaic.py +++ b/mosaic.py @@ -826,6 +826,35 @@ def _run_verify(db, output_dir: str, source: Optional[str] = None): console.print("\n[green]All cached files passed integrity checks.[/green]") +# --------------------------------------------------------------------------- +# manual +# --------------------------------------------------------------------------- + +@cli.command() +@click.option("--output", "-o", type=click.Path(), default="output/FIELD_MANUAL.md", + help="Output file path.") +@click.option("--title", type=str, default="Operator's Field Manual", + help="Manual title.") +@click.pass_context +def manual(ctx, output, title): + """Generate Markdown field manual from extracted intelligence.""" + from analysis.field_manual import generate_field_manual + + db = open_db(ctx.obj["db_path"]) + try: + path = generate_field_manual(db, output_path=output, title=title) + console.print(f"\n[bold green]Field manual generated:[/bold green] {path}") + + # Show stats + import os + size = os.path.getsize(path) + with open(path) as f: + lines = sum(1 for _ in f) + console.print(f" Size: {size:,} bytes ({lines:,} lines)") + finally: + db.close() + + # --------------------------------------------------------------------------- # sources # ---------------------------------------------------------------------------