yaraconvert.py

import sys
import json
import csv
import argparse

def show_welcome_page():
    print("\n" + "="*64)
    print("           🛸  A L I E N   P A R S E R  👽")
    print("     Convert THOR YARA logs into clean CSV reports")
    print("="*64)
    print("\nWelcome to Alien Parser!")
    print("This tool extracts YARA matches from THOR JSON logs and exports them to CSV.")
    print("\n📦 Example usage:")
    print("  python yaraconvert.py -i input_log.json -o output.csv")
    print("\n🎯 Optional filters:")
    print("  --min-score 80       Only include matches with score ≥ 80")
    print("  --rule-name RULE123  Only include matches from the specified rule")
    print("\n📄 Output:")
    print("  - Clean CSV for triage")
    print("  - JSON report with host summary")
    print("\n✨ Created for DFIR & threat hunting teams.")
    print("============================================================")
    print("H.H\n")

# Show welcome page if no arguments provided
if len(sys.argv) == 1:
    show_welcome_page()
    sys.exit(0)

def parse_thor_yara_logs(json_file, csv_file, min_score=0, rule_name_filter=None):
    results = []
    total_lines = 0
    alert_lines = 0
    parsed_lines = 0
    skipped_empty = 0
    skipped_json_error = 0
    skipped_no_reasons = 0
    skipped_invalid_match = 0
    hostnames_seen = set()

    with open(json_file, 'r', encoding='utf-8') as f:
        for i, line in enumerate(f, 1):
            total_lines += 1
            line = line.strip()
            if not line:
                skipped_empty += 1
                continue
            try:
                entry = json.loads(line)
            except json.JSONDecodeError:
                skipped_json_error += 1
                continue
            if entry.get("level") != "Alert":
                continue
            alert_lines += 1
            hostname = entry.get("hostname", "")
            if hostname:
                hostnames_seen.add(hostname)
            reasons = entry.get("reasons", [])
            file_info = entry.get("file", {})
            if not reasons:
                skipped_no_reasons += 1
                continue
            for reason in reasons:
                sig = reason.get("signature", {})
                matched = reason.get("matched", [])
                if not matched or not isinstance(matched, list):
                    skipped_invalid_match += 1
                    continue
                score = reason.get("score", 0)
                rule_name = sig.get("rulename", "")
                if score < min_score:
                    continue
                if rule_name_filter and rule_name != rule_name_filter:
                    continue
                for match in matched:
                    parsed_lines += 1
                    results.append({
                        "time": entry.get("time", ""),
                        "hostname": hostname,
                        "level": entry.get("level", ""),
                        "module": entry.get("module", ""),
                        "scanid": entry.get("scanid", ""),
                        "score": score,
                        "file_path": file_info.get("path", ""),
                        "file_ext": file_info.get("ext", ""),
                        "file_type": file_info.get("type", ""),
                        "file_size": file_info.get("size", ""),
                        "file_md5": file_info.get("md5", ""),
                        "file_sha1": file_info.get("sha1", ""),
                        "file_sha256": file_info.get("sha256", ""),
                        "file_firstbytes": file_info.get("firstbytes", ""),
                        "file_created": file_info.get("created", ""),
                        "file_modified": file_info.get("modified", ""),
                        "file_accessed": file_info.get("accessed", ""),
                        "file_permissions": file_info.get("permissions", ""),
                        "file_owner": file_info.get("owner", ""),
                        "rule_name": rule_name,
                        "rule_tags": ",".join(sig.get("tags", [])),
                        "rule_author": sig.get("author", ""),
                        "rule_class": reason.get("sigclass", ""),
                        "rule_type": reason.get("sigtype", ""),
                        "matched_data": match.get("data", ""),
                        "matched_context": match.get("context", ""),
                        "matched_offset": match.get("offset", ""),
                    })

    if results:
        fieldnames = list(results[0].keys())
        with open(csv_file, 'w', newline='', encoding='utf-8') as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            for row in results:
                writer.writerow(row)
        print(f"✅ CSV saved: {csv_file}")
    else:
        print("⚠️ No matches written to CSV.")

    metadata_output = "alien_parser_report.json"
    with open(metadata_output, 'w', encoding='utf-8') as mf:
        json.dump({
            "hosts_count": len(hostnames_seen),
            "matches_count": parsed_lines,
            "alert_lines": alert_lines,
            "unique_hosts": sorted(list(hostnames_seen))
        }, mf, indent=2)
    print(f"📤 Host summary saved to: {metadata_output}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="🛸 Alien Parser CLI")
    parser.add_argument("--input", "-i", required=True, help="Path to THOR JSON log")
    parser.add_argument("--output", "-o", required=True, help="Output CSV path")
    parser.add_argument("--min-score", type=int, default=0, help="Minimum rule score to include")
    parser.add_argument("--rule-name", type=str, help="Only include matches from this rule name")
    args = parser.parse_args()
    parse_thor_yara_logs(args.input, args.output, min_score=args.min_score, rule_name_filter=args.rule_name)