Security Skills Atlas
threat hunting

Hunting for Data Staging Before Exfiltration

Detect data staging activity before exfiltration by monitoring for archive creation with 7-Zip/RAR, unusual temp folder access, large file consolidation, and staging directory patterns via EDR and process telemetry

archive-detectiondata-stagingdlpedrexfiltrationt1074threat-hunting
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

MITRE ATT&CK

T1046 on the official MITRE ATT&CK siteT1048 on the official MITRE ATT&CK siteT1057 on the official MITRE ATT&CK siteT1074 on the official MITRE ATT&CK siteT1074.001 on the official MITRE ATT&CK siteT1074.002 on the official MITRE ATT&CK siteT1082 on the official MITRE ATT&CK siteT1083 on the official MITRE ATT&CK siteT1560.001 on the official MITRE ATT&CK site

NIST CSF 2.0

DE.AE-02 on the official NIST CSF 2.0 siteDE.AE-07 on the official NIST CSF 2.0 siteDE.CM-01 on the official NIST CSF 2.0 siteID.RA-05 on the official NIST CSF 2.0 site

MITRE D3FEND

Content Format Conversion on the official MITRE D3FEND siteFile Content Analysis on the official MITRE D3FEND siteFile Format Verification on the official MITRE D3FEND siteFile Metadata Consistency Validation on the official MITRE D3FEND sitePlatform Hardening on the official MITRE D3FEND site

Overview

Before exfiltrating data, adversaries typically stage collected files in a central location (MITRE ATT&CK T1074). This involves creating archives with tools like 7-Zip, RAR, or tar, consolidating files from multiple directories, and using temporary or hidden staging directories. This skill detects staging behavior by analyzing process creation logs for archiver activity, monitoring file system events in common staging paths, and identifying anomalous file consolidation patterns.

When to Use

  • When investigating security incidents that require hunting for data staging before exfiltration
  • When building detection rules or threat hunting queries for this domain
  • When SOC analysts need structured procedures for this analysis type
  • When validating security monitoring coverage for related attack techniques

Prerequisites

  • EDR or Sysmon telemetry with process creation and file system events
  • Windows Event Logs (Event ID 4688) or Sysmon Event ID 1, 11
  • Python 3.8+ with standard library
  • Access to process creation logs in JSON/CSV format

Steps

  1. Detect Archive Tool Execution — Monitor for 7z.exe, rar.exe, tar, zip, and WinRAR process creation with compression arguments
  2. Identify Staging Directories — Flag file writes to common staging locations (Recycle Bin, %TEMP%, ProgramData, hidden directories)
  3. Detect Large File Consolidation — Identify patterns of multiple file reads followed by writes to a single directory
  4. Monitor Sensitive Path Access — Track bulk reads from document directories, database paths, and network shares
  5. Analyze Archive Metadata — Extract and analyze archive file sizes, creation times, and source paths
  6. Score Staging Risk — Apply heuristic scoring based on archive size, source diversity, staging path suspicion, and timing
  7. Generate Hunt Report — Produce a structured report with staging event timeline and MITRE ATT&CK mapping

Expected Output

  • JSON report of detected staging events with risk scores
  • Archive creation timeline with source file analysis
  • MITRE ATT&CK mapping (T1074.001, T1074.002, T1560)
  • Staging directory heat map showing suspicious write activity
Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md2.3 KB

Data Staging Detection API Reference

MITRE ATT&CK T1074 — Data Staged

  • T1074.001 Local Data Staging — Files consolidated on the local system before exfiltration
  • T1074.002 Remote Data Staging — Files staged on a remote system or network share
  • T1560.001 Archive via Utility — Data compressed using 7-Zip, RAR, tar, etc.

Archive Tool Detection Signatures

7-Zip (7z.exe)

7z.exe a -pPASSWORD -mhe=on archive.7z C:\Users\target\Documents\*
7z.exe a -v50m split_archive.7z C:\sensitive\data\

WinRAR (rar.exe)

rar.exe a -hp"password" -r archive.rar C:\Users\*\Documents\
winrar.exe a -p"pass" -ep1 exfil.rar @filelist.txt

PowerShell Compress-Archive

Compress-Archive -Path C:\StagingDir\* -DestinationPath C:\Temp\data.zip

makecab.exe (LOLBin)

makecab.exe C:\sensitive\file.docx C:\Temp\file.cab

Sysmon Detection Queries

Sysmon Event ID 1 — Archive Process Creation

EventID=1 AND (Image="*7z.exe" OR Image="*rar.exe" OR Image="*tar*")
  AND CommandLine="*-p*"

Sysmon Event ID 11 — File Create in Staging Directory

EventID=11 AND (TargetFilename="*\\Temp\\*.zip" OR TargetFilename="*\\Temp\\*.7z"
  OR TargetFilename="*\\ProgramData\\*.rar")

Splunk SPL Queries

index=sysmon EventCode=1
| where match(Image, "(?i)(7z|rar|winrar|tar|zip)\.exe$")
| stats count by User, Image, CommandLine, ParentImage
| where count > 3
 
index=sysmon EventCode=11
| where match(TargetFilename, "(?i)\\\\(temp|programdata|\\$recycle)\\\\.*\.(zip|7z|rar|tar)")
| stats dc(TargetFilename) as unique_files, sum(FileSize) as total_bytes by User
| where total_bytes > 104857600

Elastic KQL Queries

process.name: ("7z.exe" or "rar.exe" or "winrar.exe")
  and process.args: ("-p*" or "-hp*" or "-mhe=on")
 
file.path: (*\\Temp\\*.zip or *\\Temp\\*.7z or *\\ProgramData\\*.rar)
  and event.action: "creation"

Common Staging Directories

OS Directories
Windows C:\Windows\Temp, C:\Users\Public, C:\ProgramData, C:\$Recycle.Bin, C:\PerfLogs
Linux /tmp, /var/tmp, /dev/shm

Atomic Red Team Test (T1074.001)

# Simulate local data staging
mkdir /tmp/staging
cp /home/user/Documents/*.pdf /tmp/staging/
tar -czf /tmp/staging/exfil.tar.gz /tmp/staging/*.pdf

Scripts 1

agent.py8.7 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Detect data staging activity before exfiltration via archive creation and file consolidation monitoring."""

import json
import re
import argparse
from datetime import datetime
from collections import defaultdict

ARCHIVE_TOOLS = {
    "7z.exe": {"args": ["a", "-p", "-mhe=on"], "mitre": "T1560.001"},
    "7z": {"args": ["a", "-p", "-mhe=on"], "mitre": "T1560.001"},
    "rar.exe": {"args": ["a", "-hp", "-p"], "mitre": "T1560.001"},
    "rar": {"args": ["a", "-hp", "-p"], "mitre": "T1560.001"},
    "winrar.exe": {"args": ["a", "-p"], "mitre": "T1560.001"},
    "zip": {"args": ["-r", "-e", "--encrypt"], "mitre": "T1560.001"},
    "tar": {"args": ["-czf", "-cjf", "-cJf", "cf"], "mitre": "T1560.001"},
    "compress-archive": {"args": ["-path", "-destinationpath"], "mitre": "T1560.001"},
    "makecab.exe": {"args": ["/d", "/f"], "mitre": "T1560.001"},
}

STAGING_PATHS_WINDOWS = [
    r"c:\windows\temp", r"c:\users\public", r"c:\programdata",
    r"c:\$recycle.bin", r"c:\perflogs", r"c:\windows\debug",
    r"c:\users\default", r"c:\intel",
]

STAGING_PATHS_LINUX = [
    "/tmp", "/var/tmp", "/dev/shm", "/var/spool/cron",
    "/opt/.hidden", "/usr/local/tmp",
]

SENSITIVE_SOURCE_DIRS = [
    "documents", "desktop", "downloads", "database", "backup",
    "finance", "hr", "confidential", "shared", "onedrive",
]


def parse_process_logs(log_path):
    """Parse process creation logs (Sysmon/EDR JSON format)."""
    events = []
    with open(log_path) as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                events.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return events


def detect_archive_creation(events):
    """Detect archive tool execution with compression arguments."""
    detections = []
    for event in events:
        image = event.get("Image", event.get("NewProcessName", "")).lower()
        cmdline = event.get("CommandLine", event.get("ProcessCommandLine", "")).lower()
        binary = image.rsplit("\\", 1)[-1] if "\\" in image else image.rsplit("/", 1)[-1]

        if binary not in ARCHIVE_TOOLS:
            continue

        tool_info = ARCHIVE_TOOLS[binary]
        matched_args = [a for a in tool_info["args"] if a.lower() in cmdline]

        output_files = re.findall(r'[\w\-./\\]+\.(zip|7z|rar|tar\.gz|tar\.bz2|tar|cab)', cmdline)
        encrypted = any(flag in cmdline for flag in ["-p", "-hp", "-mhe=on", "--encrypt", "-e"])

        archive_size = _estimate_archive_size(event)
        severity = "medium"
        if encrypted:
            severity = "high"
        if archive_size and archive_size > 100 * 1024 * 1024:
            severity = "critical"

        detections.append({
            "timestamp": event.get("UtcTime", event.get("TimeCreated", "")),
            "binary": binary,
            "command_line": event.get("CommandLine", event.get("ProcessCommandLine", "")),
            "output_archives": output_files,
            "encrypted": encrypted,
            "matched_args": matched_args,
            "user": event.get("User", event.get("SubjectUserName", "unknown")),
            "parent_process": event.get("ParentImage", event.get("ParentProcessName", "")),
            "severity": severity,
            "mitre": tool_info["mitre"],
            "indicator": "Archive creation (potential data staging)",
        })
    return detections


def _estimate_archive_size(event):
    """Estimate output file size from event metadata if available."""
    size_str = event.get("FileSize", event.get("TargetFileSize", ""))
    if size_str:
        try:
            return int(size_str)
        except ValueError:
            pass
    return None


def detect_staging_directory_writes(events):
    """Detect file writes to known staging directories."""
    staging_writes = []
    all_staging = STAGING_PATHS_WINDOWS + STAGING_PATHS_LINUX
    for event in events:
        target = event.get("TargetFilename", event.get("ObjectName", "")).lower()
        if not target:
            continue
        for staging_path in all_staging:
            if target.startswith(staging_path.lower()):
                staging_writes.append({
                    "timestamp": event.get("UtcTime", event.get("TimeCreated", "")),
                    "target_path": event.get("TargetFilename", event.get("ObjectName", "")),
                    "staging_directory": staging_path,
                    "process": event.get("Image", event.get("NewProcessName", "")),
                    "user": event.get("User", event.get("SubjectUserName", "unknown")),
                    "severity": "medium",
                    "mitre": "T1074.001",
                    "indicator": f"File write to staging directory: {staging_path}",
                })
                break
    return staging_writes


def detect_sensitive_source_reads(events, time_window_minutes=30):
    """Detect bulk reads from sensitive directories (document consolidation)."""
    user_reads = defaultdict(list)
    for event in events:
        source = event.get("SourceFilename", event.get("ObjectName", "")).lower()
        if not source:
            continue
        for sensitive_dir in SENSITIVE_SOURCE_DIRS:
            if sensitive_dir in source:
                user = event.get("User", event.get("SubjectUserName", "unknown"))
                user_reads[user].append({
                    "path": source,
                    "timestamp": event.get("UtcTime", event.get("TimeCreated", "")),
                    "sensitive_dir": sensitive_dir,
                })
                break
    consolidation_alerts = []
    for user, reads in user_reads.items():
        unique_dirs = len({r["sensitive_dir"] for r in reads})
        if len(reads) >= 20 or unique_dirs >= 3:
            consolidation_alerts.append({
                "user": user,
                "total_reads": len(reads),
                "unique_sensitive_dirs": unique_dirs,
                "directories_accessed": list({r["sensitive_dir"] for r in reads}),
                "first_access": reads[0]["timestamp"],
                "last_access": reads[-1]["timestamp"],
                "severity": "critical" if unique_dirs >= 3 else "high",
                "mitre": "T1074.001",
                "indicator": "Bulk reads from sensitive directories (data consolidation)",
            })
    return consolidation_alerts


def score_staging_risk(archives, staging_writes, consolidation):
    """Compute overall staging risk score (0-100)."""
    score = 0
    if archives:
        score += min(30, len(archives) * 10)
        if any(a["encrypted"] for a in archives):
            score += 20
    if staging_writes:
        score += min(25, len(staging_writes) * 5)
    if consolidation:
        score += min(25, len(consolidation) * 15)
    return min(100, score)


def generate_report(archives, staging_writes, consolidation, log_path):
    """Generate data staging hunt report."""
    risk_score = score_staging_risk(archives, staging_writes, consolidation)
    return {
        "report_time": datetime.utcnow().isoformat(),
        "log_source": log_path,
        "risk_score": risk_score,
        "risk_level": "critical" if risk_score >= 70 else "high" if risk_score >= 40 else "medium" if risk_score >= 20 else "low",
        "archive_creation_events": len(archives),
        "staging_directory_writes": len(staging_writes),
        "data_consolidation_alerts": len(consolidation),
        "mitre_techniques": ["T1074.001", "T1074.002", "T1560.001"],
        "archive_detections": archives,
        "staging_write_detections": staging_writes[:20],
        "consolidation_alerts": consolidation,
    }


def main():
    parser = argparse.ArgumentParser(description="Data Staging Detection Agent (MITRE T1074)")
    parser.add_argument("--log-file", required=True, help="JSON process/file event log")
    parser.add_argument("--output", default="data_staging_report.json")
    parser.add_argument("--read-threshold", type=int, default=20, help="Bulk read threshold")
    args = parser.parse_args()

    events = parse_process_logs(args.log_file)
    archives = detect_archive_creation(events)
    staging_writes = detect_staging_directory_writes(events)
    consolidation = detect_sensitive_source_reads(events)

    report = generate_report(archives, staging_writes, consolidation, args.log_file)
    with open(args.output, "w") as f:
        json.dump(report, f, indent=2, default=str)
    print(f"[+] Analyzed {len(events)} events")
    print(f"[+] Archive creation: {len(archives)} | Staging writes: {len(staging_writes)} | Consolidation: {len(consolidation)}")
    print(f"[+] Risk score: {report['risk_score']}/100 ({report['risk_level']})")
    print(f"[+] Report saved to {args.output}")


if __name__ == "__main__":
    main()
Keep exploring
Detecting DLL Sideloading AttacksDetect DLL side-loading attacks where adversaries place malicious DLLs alongside legitimate applications to hijack execution flow for defense evasion.Detecting Mimikatz Execution PatternsDetect Mimikatz execution through command-line patterns, LSASS access signatures, binary indicators, and in-memory detection of known modules.Detecting Process Hollowing TechniqueDetect process hollowing (T1055.012) by analyzing memory-mapped sections, hollowed process indicators, and parent-child process anomalies in EDR telemetry.Detecting T1003 Credential Dumping with EDRDetect OS credential dumping techniques targeting LSASS memory, SAM database, NTDS.dit, and cached credentials using EDR telemetry, Sysmon process access monitoring, and Windows security event correlation.Hunting for Data Exfiltration IndicatorsHunt for data exfiltration through network traffic analysis, detecting unusual data flows, DNS tunneling, cloud storage uploads, and encrypted channel abuse.Hunting for Living-off-the-Land Binaries (LOLBins)Proactively hunt for adversary abuse of legitimate system binaries (LOLBins) to execute malicious payloads while evading detection.