Matrix/Analyze/SDF-AN01

SDF-AN01

TTP Extraction and Cataloging

advancedDetection Value: highCost: lowTime: 3-7 days per engagement🖥 Endpoint Systems

Description

Systematically extract techniques, tactics, and procedures from raw deception telemetry and catalog them in a structured format. This transforms unstructured interaction data into organized intelligence that can be compared across incidents and used to develop detection rules.

Rationale

Raw logs and packet captures are data, not intelligence. TTP extraction applies analytical rigor to identify what the adversary did, how they did it, and what tools they used. This structured output is what enables detection rule development, adversary tracking, and defense gap analysis.

Supporting Basis

Literature-informed design

Honeypots yield observed attacker activity and ATT&CK provides a TTP vocabulary; the combined cataloguing step is a ShadowMatrix synthesis.

These sources establish design lineage and scope. They do not validate this record's operational effectiveness, planning estimates, environment labels, or ATT&CK relationships.

Implementation Steps

1

Aggregate interaction data

Collect all telemetry from the adversary engagement: host logs, network captures, session recordings, and canary alert data.

2

Identify individual techniques

Analyze the data to identify distinct techniques the adversary employed. Break down complex operations into individual technique components.

3

Document procedure details

For each technique, document the specific procedure: exact commands used, tool configurations, and operational patterns.

4

Catalog in structured format

Enter extracted TTPs into a structured catalog with standardized fields: technique name, procedure details, tools used, detection opportunities, and MITRE ATT&CK mapping.

Technical Implementation Examples(1)

TTP Extraction Script

Parse honeypot and deception logs to extract and catalog adversary techniques, tactics, and procedures

python
"""
TTP Extraction from Deception Telemetry.
Parses Cowrie SSH logs, web logs, and canary alerts to identify
and catalog adversary techniques.
"""
import json
from collections import defaultdict
from datetime import datetime
from typing import Dict, List

# Command-to-TTP mapping based on common adversary behaviors
COMMAND_TTP_MAP = {
    "whoami":        ("T1033", "System Owner/User Discovery"),
    "id":            ("T1033", "System Owner/User Discovery"),
    "uname":         ("T1082", "System Information Discovery"),
    "cat /etc/passwd": ("T1087.001", "Local Account Discovery"),
    "net user":      ("T1087.001", "Local Account Discovery"),
    "ipconfig":      ("T1016", "System Network Configuration"),
    "ifconfig":      ("T1016", "System Network Configuration"),
    "netstat":       ("T1049", "System Network Connections"),
    "ps aux":        ("T1057", "Process Discovery"),
    "wget":          ("T1105", "Ingress Tool Transfer"),
    "curl":          ("T1105", "Ingress Tool Transfer"),
    "chmod +x":      ("T1222", "File and Directory Permissions"),
    "crontab":       ("T1053.003", "Scheduled Task: Cron"),
    "ssh-keygen":    ("T1098.004", "SSH Authorized Keys"),
    "cat /etc/shadow": ("T1003.008", "OS Credential Dumping"),
    "history":       ("T1552.003", "Bash History"),
}

def extract_ttps_from_cowrie(log_file: str) -> List[Dict]:
    """Extract TTPs from Cowrie SSH honeypot JSON logs."""
    ttps = []

    with open(log_file) as f:
        for line in f:
            try:
                event = json.loads(line.strip())
            except json.JSONDecodeError:
                continue

            if event.get("eventid") == "cowrie.command.input":
                cmd = event.get("input", "").strip()
                session = event.get("session", "")
                src_ip = event.get("src_ip", "")
                timestamp = event.get("timestamp", "")

                # Match against known TTP patterns
                for pattern, (tech_id, tech_name) in COMMAND_TTP_MAP.items():
                    if pattern in cmd.lower():
                        ttps.append({
                            "technique_id": tech_id,
                            "technique_name": tech_name,
                            "evidence": cmd,
                            "source": "cowrie-ssh",
                            "session_id": session,
                            "source_ip": src_ip,
                            "timestamp": timestamp,
                            "confidence": "high",
                        })

            elif event.get("eventid") == "cowrie.session.file_download":
                ttps.append({
                    "technique_id": "T1105",
                    "technique_name": "Ingress Tool Transfer",
                    "evidence": f"Downloaded: {event.get('url', 'N/A')}",
                    "file_hash": event.get("shasum", ""),
                    "source": "cowrie-ssh",
                    "session_id": event.get("session", ""),
                    "source_ip": event.get("src_ip", ""),
                    "timestamp": event.get("timestamp", ""),
                    "confidence": "high",
                })

    return ttps

def generate_ttp_report(ttps: List[Dict]) -> Dict:
    """Generate a structured TTP report."""
    by_technique = defaultdict(list)
    for ttp in ttps:
        by_technique[ttp["technique_id"]].append(ttp)

    report = {
        "report_type": "ttp_extraction",
        "generated_at": datetime.utcnow().isoformat() + "Z",
        "total_ttps_observed": len(ttps),
        "unique_techniques": len(by_technique),
        "techniques": [
            {
                "technique_id": tid,
                "technique_name": observations[0]["technique_name"],
                "observation_count": len(observations),
                "first_seen": min(o["timestamp"] for o in observations),
                "last_seen": max(o["timestamp"] for o in observations),
                "source_ips": list({o["source_ip"] for o in observations}),
                "evidence_samples": [
                    o["evidence"] for o in observations[:5]
                ],
            }
            for tid, observations in sorted(by_technique.items())
        ],
    }

    return report

if __name__ == "__main__":
    ttps = extract_ttps_from_cowrie("/var/log/cowrie/cowrie.json")
    report = generate_ttp_report(ttps)
    print(json.dumps(report, indent=2))

Considerations

  • Distinguish between adversary choices and adversary capabilities. An adversary who uses basic techniques may be choosing efficiency, not lacking sophistication.
  • Document negative indicators as well. What did the adversary avoid doing? What detection they evaded can tell you about their awareness of security controls.
  • Compare extracted TTPs with previously cataloged adversary behavior to identify returning actors or shared toolkits.

Supported Environments

🏢On-Premise
☁Amazon Web Services
☁Microsoft Azure
☁Google Cloud Platform
🔄Hybrid Environment
🏭OT/ICS Environment

Tools & Resources

  • MITRE ATT&CK Navigator
  • Threat intelligence platforms
  • Analysis workbench tools
  • Structured reporting templates

Quick Stats

Phase
Analyze
Difficulty
advanced
Detection Value
high
Cost
low
Time
3-7 days per engagement
ATT&CK Mappings
0