Weipeng Xu

"""
# By repository (existing behavior, now uses repository_upload_sessions)
python start.py --s3 --repo bazaar --index_prefix redb

# By repository with notes filter
python start.py --s3 --repo vx-itw --s3-notes "ITW.0138" --index_prefix redb

# By single date (all repo samples first seen on Jan 15, 2025)
python start.py --date 2025-01-15 --index_prefix redb

# By date with repository filter
python start.py --date 2025-01-15 --repo bazaar --index_prefix redb

# By date range (inclusive)
python start.py --range 2025-01-01 2025-01-31 --index_prefix redb

# By date range with repository and notes filters
python start.py --range 2025-01-01 2025-01-31 --repo malshare --s3-notes "batch1" --index_prefix redb

# By filetype (magika) standalone - process all ELF samples across all repos
python start.py --s3 --magika elf --index_prefix redb

# By filetype with repository filter
python start.py --s3 --repo bazaar --magika elf --index_prefix redb

# By filetype with date range - process only PE samples in date range
python start.py --range 2025-01-01 2025-01-31 --magika pebin --index_prefix redb
"""

import argparse
import os
import sys
from datetime import datetime, timedelta
from redb.ingestor import *

"""
        # General modules
        'BasicPropertiesExtractor': BasicPropertiesExtractor,
        'HashExtractor': HashExtractor,
        'DIEExtractor': DIEExtractor,
        'CAPAExtractor': CAPAExtractor,
        'StringsExtractor': StringsExtractor,
        # PE modules
        'PEFeaturesExtractor': PEFeaturesExtractor,
        'PEImportExtractor': PEImportExtractor,
        'PEResourceExtractor': PEResourceExtractor,
        'PEOverlayExtractor': PEOverlayExtractor,
        'PESectionExtractor': PESectionExtractor,
        'PESignatureExtractor': PESignatureExtractor,
        'PEExtraFindings': PEExtraFindings,
        'PEInconstistencyTestsExtractor': PEInconstistencyTestsExtractor,
        'PEDotNetExtractor': PEDotNetExtractor,
"""


def main():
    # # path = "/mnt/samples/consilience/malware/test/test-unzipped/0242d90dc48a8931bad72ddbdba34bdd568fd30610dfe049c84968d425088c71/"
    # path = "/Users/p4c0/_samples/test-unzipped/0242d90dc48a8931bad72ddbdba34bdd568fd30610dfe049c84968d425088c71" #Stuxnet
    # # path = "/Volumes/backup/consilience/malware/test/test-unzipped/rhpv-673f91a2085358e3266f466845366f30cf741060edeb31e9a93e2c92033bba28"
    # # path = "/mnt/samples/consilience/malware/test/test-redb/"
    # # path = "/mnt/samples/consilience/malware/malpedia-pe/9bc81280113473de9ebfe54f689b4440287c37fff562e070d3a28f5269cadcf0_dump7_0x00400000"
    # # path = "/mnt/samples/consilience/malware/test/test-unzipped/379251974ebcd5c397f92ca45bb9620d"
    # path = "0242d90dc48a8931bad72ddbdba34bdd568fd30610dfe049c84968d425088c71" # rich header, UPX packer
    # # #path = "d8637bdbcfc9112fcb1f0167b398e771" #dotnet
    # path = "/Users/p4c0/_samples/test-unzipped/sig-8e035beb02a411f8a9e92d4cf184ad34f52bbd0a81a50c222cdd4706e4e45104" #code signed, protector use case for sections
    # path = "/Users/p4c0/_samples/test-unzipped/vsinfo-39d8ad95b0323c37bd3134ab93ac4af44c66a1a8443a41c1ac02cec19bb2816a"
    # # path = "/Volumes/backup/consilience/malware/_sorted_samples/vx-apt/pebin/69e679daaaff3832c39671bf2b813b5530a70fb763d381f9a6e22e3bc493c8a9.7z"
    # # path = "test_files/hello"
    # path = "/Volumes/backup/consilience/malware/test/test-re2db/"
    # # path = "/Users/p4c0/_samples/HEUR-Trojan-PSW.MSIL.Maslog.gen-0c9ae5cd740c1da7060b92ddb33f3a3893e361aad45a2accc64d43bd9a1a4106"
    # # path = "/Users/p4c0/_samples/test-unzipped/"
    # # # path = "/Volumes/backup/consilience/malware/_sorted_samples/vx-apt/pebin/7156bd8056c4b6b4e179a64370067d3f7a7cce0044f1352d41f3c2c73038d273.7z"
    # decompile = False
    # repo = "test-fixing"
    # index_prefix = "test4"
    # selected_modules = "all"
    # exporter_types = ['ClickHouse']

    # parser = argparse.ArgumentParser(
    #     description="Process binary files in a given path."
    # )
    # parser.add_argument("path", 
    #     help="File path, directory path, or path to a .txt file containing a list of files to process (one per line)")
    # parser.add_argument(
    #     "--repo", help="Repository name for sample source, used for loggfile name"
    # )
    # parser.add_argument("--index_prefix", help="Index prefix for ElasticSearch")
    # parser.add_argument(
    #     "-d",
    #     "--decompile",
    #     action="store_true",
    #     help="Optional flag, if set it will run ONLY the decompiler on the binary files",
    # )
    # parser.add_argument(
    #     "-m",
    #     "--modules",
    #     help="Comma-separated list of modules to run (e.g., 'BasicPropertiesExtractor,HashExtractor') or 'all' for all modules",
    #     default="all",
    # )
    # args = parser.parse_args()
    # path = args.path
    # index_prefix = args.index_prefix
    # decompile = args.decompile
    # repo = args.repo
    # selected_modules = args.modules

    # print(f"Decompile flag: {decompile}")
    # print(f"Repo: {repo}")
    # print(f"Selected modules: {selected_modules}")
    
    # if path.endswith('.txt'):
    #     print(f"Reading file list from: {path}")

    # Ingestor(path, decompile, repo, index_prefix, selected_modules).ingest()

    parser = argparse.ArgumentParser(
        description="Process binary files from local paths or S3 storage."
    )
    
    # Create a mutually exclusive group for input sources
    # Not required because --analyzed can be used standalone
    input_group = parser.add_mutually_exclusive_group(required=False)
    input_group.add_argument(
        "--path", 
        help="File path, directory path, or path to a .txt file containing a list of files to process (one per line)")
    input_group.add_argument(
        "--s3",
        action="store_true",
        help="Use S3 mode to fetch files from repository specified by --repo")
    input_group.add_argument(
        "--s3-solo",
        metavar="S3_KEY",
        help="Process a single S3 file by providing the S3 key (e.g., 09/f7/09f7d02a....zip)")
    input_group.add_argument(
        "--nomad-job",
        action="store_true",
        help="Run as Nomad job using environment variables for job parameters")
    input_group.add_argument(
        "--date",
        metavar="YYYY-MM-DD",
        help="Process samples first seen on a specific date (from repository_upload_sessions only)")
    input_group.add_argument(
        "--range",
        nargs=2,
        metavar=("START_DATE", "END_DATE"),
        help="Process samples first seen in a date range (inclusive, from repository_upload_sessions only). Format: YYYY-MM-DD YYYY-MM-DD")
    parser.add_argument(
        "--analyzed",
        action="store_true",
        help="Filter to samples already in the database (from basic_properties). "
             "Can be used standalone or combined with --range/--date to partition large runs")

    parser.add_argument(
        "--repo",
        required=False,
        help="Repository name for sample source, used for logging and S3 filtering (optional for --date/--range modes)")
    
    parser.add_argument(
        "--s3-notes",
        help="Optional filter for S3 files based on notes field (S3 mode only)")

    parser.add_argument(
        "--magika",
        help="Filter by filetype_magika (e.g., 'elf', 'pebin'). Overrides SUPPORTED_FORMATS env var. Can combine with --repo, --date, --range")

    parser.add_argument(
        "--index_prefix", default="redb",
        help="Index prefix for database (default: redb)")
    
    parser.add_argument(
        "-d",
        "--decompile",
        action="store_true",
        help="Optional flag, if set it will run ONLY the decompiler on the binary files",
    )

    parser.add_argument(
        "-y",
        "--yara",
        action="store_true",
        help="Optional flag, if set it will run ONLY the YARA scanner on the binary files",
    )

    parser.add_argument(
        "--with-yara",
        action="store_true",
        help="Add YARA scanning to feature extraction (runs both features and YARA)",
    )

    parser.add_argument(
        "-m",
        "--modules",
        help="Comma-separated list of modules to run (e.g., 'BasicPropertiesExtractor,HashExtractor') or 'all' for all modules",
        default="all",
    )

    parser.add_argument(
        "--decompile-modules",
        help="Comma-separated list of decompiler sub-modules to run when using -d/--decompile. "
             "Available: decompilation, disassembly, cfg, llil, strings, or 'all' (default: all)",
        default="all",
    )
    
    parser.add_argument(
        "--force",
        action="store_true",
        help="Force reprocessing of samples already in the database (bypasses deduplication check)",
    )

    parser.add_argument(
        "--rerun",
        action="store_true",
        help="Re-run decompiler modules on already-disassembled samples only. "
             "Queries code_binja_disassembled_functions_references instead of basic_properties. "
             "Requires --analyzed and --decompile.",
    )

    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Print results instead of uploading to database (useful for testing)",
    )
    
    args = parser.parse_args()

    # Extract arguments
    path = args.path
    index_prefix = args.index_prefix
    decompile = args.decompile
    yara_scan = args.yara
    with_yara = args.with_yara
    repo = args.repo
    selected_modules = args.modules
    decompile_modules = args.decompile_modules
    s3_mode = args.s3
    s3_notes = args.s3_notes
    magika_filter = args.magika
    dry_run = args.dry_run
    force = args.force
    rerun = args.rerun
    s3_solo = args.s3_solo
    s3_key = args.s3_solo if args.s3_solo else None
    nomad_job = args.nomad_job
    analyzed = args.analyzed
    date_filter = args.date
    date_range = args.range

    # Validate that at least one input source is provided
    has_input = any([path, s3_mode, s3_solo, nomad_job, date_filter, date_range, analyzed])
    if not has_input:
        print("ERROR: Must specify an input source: --path, --s3, --s3-solo, --nomad-job, --date, --range, or --analyzed")
        sys.exit(1)

    # Validate --analyzed combinations
    if analyzed and any([path, s3_mode, s3_solo, nomad_job]):
        print("ERROR: --analyzed cannot be combined with --path, --s3, --s3-solo, or --nomad-job")
        sys.exit(1)

    # Validate flag combinations
    if yara_scan and with_yara:
        print("ERROR: Cannot use both --yara and --with-yara")
        sys.exit(1)
    if decompile and with_yara:
        print("ERROR: --with-yara only works with feature extraction, not decompile")
        sys.exit(1)

    # Parse and validate --decompile-modules
    VALID_DECOMPILE_MODULES = {"all", "decompilation", "disassembly", "cfg", "llil", "strings"}
    if decompile_modules == "all":
        decompile_modules_set = {"all"}
    else:
        decompile_modules_set = {m.strip() for m in decompile_modules.split(",")}
        invalid = decompile_modules_set - VALID_DECOMPILE_MODULES
        if invalid:
            print(f"ERROR: Invalid decompile module(s): {', '.join(sorted(invalid))}")
            print(f"Available: {', '.join(sorted(VALID_DECOMPILE_MODULES - {'all'}))}")
            sys.exit(1)

    if not decompile and decompile_modules != "all":
        print("ERROR: --decompile-modules requires -d/--decompile flag")
        sys.exit(1)

    if rerun and not analyzed:
        print("ERROR: --rerun requires --analyzed flag")
        sys.exit(1)
    if rerun and not decompile:
        print("ERROR: --rerun requires -d/--decompile flag")
        sys.exit(1)

    if rerun and force:
        print("ERROR: --rerun and --force are mutually exclusive. "
              "--rerun targets already-disassembled samples, --force targets all analyzed samples.")
        sys.exit(1)

    # Validate and parse date arguments
    start_date = None
    end_date = None

    if date_filter:
        # Single date mode: process samples from that day
        try:
            parsed_date = datetime.strptime(date_filter, "%Y-%m-%d")
            start_date = date_filter
            # End date is the next day (exclusive)
            end_date = (parsed_date + timedelta(days=1)).strftime("%Y-%m-%d")
        except ValueError:
            print(f"ERROR: Invalid date format '{date_filter}'. Use YYYY-MM-DD")
            sys.exit(1)

    if date_range:
        # Date range mode: process samples between start and end dates
        try:
            start_date = date_range[0]
            datetime.strptime(start_date, "%Y-%m-%d")  # Validate format
            parsed_end = datetime.strptime(date_range[1], "%Y-%m-%d")
            # End date is the day after the provided end date (to make it inclusive)
            end_date = (parsed_end + timedelta(days=1)).strftime("%Y-%m-%d")
        except ValueError:
            print(f"ERROR: Invalid date format in range '{date_range}'. Use YYYY-MM-DD YYYY-MM-DD")
            sys.exit(1)

    # Validate required parameters based on mode
    if nomad_job or s3_solo:
        # For nomad job and s3-solo modes, set default repo if not provided (used for log filename)
        if not repo:
            repo = "s3-solo" if s3_solo else "nomad-worker"
    elif analyzed:
        # For analyzed mode, repo is optional (used for logging only)
        if not repo:
            repo = "analyzed"
    elif date_filter or date_range:
        # For date/range modes, repo is optional (used for filtering within date range)
        # Set a default repo name for logging if not provided
        if not repo:
            repo = "date-range"
    elif s3_mode and magika_filter and not repo:
        # For S3 mode with magika filter, repo is optional (query all repos for that filetype)
        repo = "all-repos"
    elif not s3_mode and path:
        # For local mode, repo is required
        if not repo:
            print("ERROR: --repo is required for local mode")
            sys.exit(1)
    elif s3_mode and not repo:
        # For S3 mode without magika filter, repo is required
        print("ERROR: --repo is required for S3 catalog mode (or use --magika to query all repos)")
        sys.exit(1)

    print(f"Decompile flag: {decompile}")
    if decompile and decompile_modules != "all":
        print(f"Decompile modules: {', '.join(sorted(decompile_modules_set))}")
    print(f"YARA scan flag: {yara_scan}")
    print(f"With YARA flag: {with_yara}")
    print(f"Repo: {repo}")
    print(f"Selected modules: {selected_modules}")
    print(f"Dry run mode: {dry_run}")
    print(f"Force reprocessing: {force}")
    print(f"S3 solo mode: {s3_solo}")
    print(f"Nomad job mode: {nomad_job}")
    if magika_filter:
        print(f"Magika filter: {magika_filter}")
    if start_date:
        print(f"Date filter: {start_date} to {end_date}")
    if rerun:
        print(f"Rerun mode: targeting already-disassembled samples from code_binja_disassembled_functions_references")
    if analyzed:
        print(f"Analyzed mode: processing already-analyzed samples from basic_properties")

    if analyzed and not (date_filter or date_range):
        # Analyzed mode (standalone): process samples already in basic_properties via S3
        print(f"Processing already-analyzed samples from {index_prefix}_basic_properties")
        if magika_filter:
            print(f"Filetype filter: {magika_filter}")
        Ingestor(
            path=None,
            decompile=decompile,
            yara_scan=yara_scan,
            with_yara=with_yara,
            repository=repo or "analyzed",
            index_prefix=index_prefix,
            selected_modules=selected_modules,
            s3_mode=True,
            magika_filter=magika_filter,
            dry_run=dry_run,
            force=force,
            analyzed=True,
            decompile_modules=decompile_modules_set,
            rerun=rerun,
        ).ingest()

    elif date_filter or date_range:
        # Date-based S3 mode
        print(f"Date-based S3 mode enabled")
        if analyzed:
            print(f"Filtered to already-analyzed samples in {index_prefix}_basic_properties")
        if repo and repo != "date-range":
            print(f"Repository filter: {repo}")
        if s3_notes:
            print(f"Notes filter: {s3_notes}")
        Ingestor(
            path=None,
            decompile=decompile,
            yara_scan=yara_scan,
            with_yara=with_yara,
            repository=repo,
            index_prefix=index_prefix,
            selected_modules=selected_modules,
            s3_mode=True,
            s3_notes=s3_notes,
            magika_filter=magika_filter,
            dry_run=dry_run,
            force=force,
            start_date=start_date,
            end_date=end_date,
            analyzed=analyzed,
            decompile_modules=decompile_modules_set,
        ).ingest()

    elif s3_solo:
        # Process a single S3 file using S3 key provided as argument
        print(f"Starting S3 solo mode with S3 key: {s3_key}")

        # Override with environment variables if not provided via command line
        if not index_prefix:
            index_prefix = os.getenv('INDEX_PREFIX', 'redb')
        if not repo:
            repo = os.getenv('REPO', 's3-solo')

        # Validate required parameters
        if not s3_key:
            print("ERROR: S3 key is required for S3-solo mode")
            sys.exit(1)

        # Extract hash from S3 key by splitting and taking the last chunk
        # S3 key format examples:
        # - 09/f7/09f7d02a3c2382199458c98a62b045145ee54ab6aba86166aecf3d10c3c1444c.zip
        # - private/ab/cd/abcd1234567890abcdef1234567890abcdef1234567890abcdef123456.zip
        try:
            # Remove .zip extension and split by '/'
            sample_hash = s3_key.replace('.zip', '').split('/')[-1]
        except Exception as e:
            print(f"ERROR: Failed to extract hash from S3 key {s3_key}: {e}")
            sys.exit(1)

        print(f"S3 Key: {s3_key}")
        print(f"Extracted hash: {sample_hash}")
        print(f"Using index_prefix: {index_prefix}")
        print(f"Using repo: {repo}")
        print(f"Dry run mode: {dry_run}")
        print(f"Selected modules: {selected_modules}")

        # Use ingestor with S3-solo mode
        Ingestor(
            path=None,
            decompile=decompile,
            yara_scan=yara_scan,
            with_yara=with_yara,
            repository=repo,
            index_prefix=index_prefix or "s3_solo",
            selected_modules=selected_modules,
            s3_mode=True,
            s3_solo=True,
            s3_solo_hash=sample_hash,
            s3_solo_key=s3_key,
            dry_run=dry_run,
            force=force,
            decompile_modules=decompile_modules_set,
        ).ingest()

    elif nomad_job:
        # Run as Nomad job using environment variables - convert to S3-solo mode
        print("Starting Nomad job processor...")
        
        # Process Nomad environment variables
        job_id = os.getenv('JOB_ID')
        s3_key = os.getenv('S3_KEY')
        worker_type = os.getenv('WORKER_TYPE')
        callback_url = os.getenv('CALLBACK_URL')
        modules = os.getenv('ANALYSIS_MODULES', 'all')
        
        # Validate required parameters
        if not all([job_id, s3_key, worker_type, callback_url]):
            print("ERROR: Missing required Nomad job parameters")
            print("Required: JOB_ID, S3_KEY, WORKER_TYPE, CALLBACK_URL")
            sys.exit(1)
        
        print(f"Job ID: {job_id}")
        print(f"S3 Key: {s3_key}")
        print(f"Worker Type: {worker_type}")
        print(f"Callback URL: {callback_url}")
        print(f"Analysis Modules: {modules}")
        
        # Extract hash from S3 key (remove sharding structure and .zip extension)
        # S3 key format: 09/f7/09f7d02a3c2382199458c98a62b045145ee54ab6aba86166aecf3d10c3c1444c.zip
        # Extract: 09f7d02a3c2382199458c98a62b045145ee54ab6aba86166aecf3d10c3c1444c
        try:
            parts = s3_key.split('/')
            if len(parts) == 3:
                sample_hash = parts[2].replace('.zip', '')
            else:
                # Fallback for non-sharded keys
                sample_hash = s3_key.replace('.zip', '')
        except Exception as e:
            print(f"ERROR: Failed to extract hash from S3 key {s3_key}: {e}")
            sys.exit(1)
        
        print(f"Extracted hash: {sample_hash}")
        
        # Set decompile flag based on worker type
        decompile = worker_type == 'decompilation'
        
        # Override modules if specified
        if modules != 'all':
            selected_modules = modules
        
        # Use S3-solo mode with extracted hash
        Ingestor(
            path=None,
            decompile=decompile,
            yara_scan=yara_scan,
            with_yara=with_yara,
            repository=repo,
            index_prefix=index_prefix or "nomad",
            selected_modules=selected_modules,
            s3_mode=False,  # Not bulk S3 mode
            s3_solo=True,   # Use S3-solo mode
            s3_solo_hash=sample_hash,
            dry_run=dry_run,
            force=force,
            decompile_modules=decompile_modules_set,
        ).ingest()

        # TODO: Add callback to send results to callback_url
        print(f"[INFO] Nomad job {job_id} completed. Callback URL: {callback_url}")
    
    elif s3_mode:
        print(f"S3 mode enabled")
        if s3_notes:
            print(f"S3 notes filter: {s3_notes}")
        Ingestor(
            path=None,
            decompile=decompile,
            yara_scan=yara_scan,
            with_yara=with_yara,
            repository=repo,
            index_prefix=index_prefix,
            selected_modules=selected_modules,
            s3_mode=True,
            s3_notes=s3_notes,
            magika_filter=magika_filter,
            dry_run=dry_run,
            force=force,
            decompile_modules=decompile_modules_set,
        ).ingest()
    else:
        print(f"Local mode with path: {path}")
        if path.endswith('.txt'):
            print(f"Reading file list from: {path}")

        Ingestor(
            path=path,
            decompile=decompile,
            yara_scan=yara_scan,
            with_yara=with_yara,
            repository=repo,
            index_prefix=index_prefix,
            selected_modules=selected_modules,
            dry_run=dry_run,
            force=force,
            decompile_modules=decompile_modules_set,
        ).ingest()

if __name__ == "__main__":
    main()