auto_sigma_rule_generator/cli/sigma_cli.py

#!/usr/bin/env python3
"""
SIGMA CLI - CVE-SIGMA Auto Generator Command Line Interface

A CLI tool for processing CVEs and generating SIGMA detection rules
in a file-based directory structure.

Author: CVE-SIGMA Auto Generator
"""

import click
import asyncio
import os
import sys
import json
from typing import Optional, List
from pathlib import Path
from datetime import datetime

# Add parent directories to path for imports
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'backend'))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'core'))

# Import CLI command modules
from commands.process_commands import ProcessCommands
from commands.generate_commands import GenerateCommands
from commands.search_commands import SearchCommands
from commands.stats_commands import StatsCommands
from commands.export_commands import ExportCommands
from commands.migrate_commands import MigrateCommands

# Global CLI configuration
class Config:
    def __init__(self):
        self.base_dir = Path.cwd()
        self.cves_dir = self.base_dir / "cves"
        self.templates_dir = self.base_dir / "backend" / "templates"
        self.reports_dir = self.base_dir / "reports"
        self.config_file = Path.home() / ".sigma-cli" / "config.yaml"
        
        # Ensure directories exist
        self.cves_dir.mkdir(exist_ok=True)
        self.reports_dir.mkdir(exist_ok=True)
        (Path.home() / ".sigma-cli").mkdir(exist_ok=True)

pass_config = click.make_pass_decorator(Config, ensure=True)

@click.group()
@click.option('--verbose', '-v', is_flag=True, help='Enable verbose output')
@click.option('--config', '-c', type=click.Path(), help='Path to configuration file')
@click.pass_context
def cli(ctx, verbose, config):
    """
    SIGMA CLI - CVE-SIGMA Auto Generator
    
    A command line tool for processing CVEs and generating SIGMA detection rules.
    Rules are stored in a file-based directory structure organized by year and CVE-ID.
    """
    ctx.ensure_object(Config)
    if verbose:
        click.echo("Verbose mode enabled")
    
    if config:
        ctx.obj.config_file = Path(config)
    
    # Initialize logging
    import logging
    level = logging.DEBUG if verbose else logging.INFO
    logging.basicConfig(level=level, format='%(asctime)s - %(levelname)s - %(message)s')

# Process commands
@cli.group()
@pass_config
def process(config):
    """Process CVEs and generate SIGMA rules"""
    pass

@process.command('year')
@click.argument('year', type=int)
@click.option('--method', '-m', multiple=True, type=click.Choice(['template', 'llm', 'hybrid', 'all']), 
              default=['template'], help='Rule generation method(s)')
@click.option('--force', '-f', is_flag=True, help='Force regeneration of existing rules')
@click.option('--batch-size', '-b', default=50, help='Batch size for processing')
@pass_config
def process_year(config, year, method, force, batch_size):
    """Process all CVEs for a specific year"""
    cmd = ProcessCommands(config)
    asyncio.run(cmd.process_year(year, method, force, batch_size))

@process.command('cve')
@click.argument('cve_id')
@click.option('--method', '-m', multiple=True, type=click.Choice(['template', 'llm', 'hybrid', 'all']), 
              default=['template'], help='Rule generation method(s)')
@click.option('--force', '-f', is_flag=True, help='Force regeneration of existing rules')
@pass_config
def process_cve(config, cve_id, method, force):
    """Process a specific CVE"""
    cmd = ProcessCommands(config)
    asyncio.run(cmd.process_cve(cve_id, method, force))

@process.command('bulk')
@click.option('--start-year', default=2022, help='Starting year for bulk processing')
@click.option('--end-year', default=datetime.now().year, help='Ending year for bulk processing')
@click.option('--method', '-m', multiple=True, type=click.Choice(['template', 'llm', 'hybrid', 'all']), 
              default=['template'], help='Rule generation method(s)')
@click.option('--batch-size', '-b', default=50, help='Batch size for processing')
@pass_config
def process_bulk(config, start_year, end_year, method, batch_size):
    """Bulk process all CVEs across multiple years"""
    cmd = ProcessCommands(config)
    asyncio.run(cmd.process_bulk(start_year, end_year, method, batch_size))

@process.command('incremental')
@click.option('--days', '-d', default=7, help='Process CVEs modified in the last N days')
@click.option('--method', '-m', multiple=True, type=click.Choice(['template', 'llm', 'hybrid', 'all']), 
              default=['template'], help='Rule generation method(s)')
@pass_config
def process_incremental(config, days, method):
    """Process recently modified CVEs"""
    cmd = ProcessCommands(config)
    asyncio.run(cmd.process_incremental(days, method))

# Generate commands
@cli.group()
@pass_config
def generate(config):
    """Generate SIGMA rules for existing CVEs"""
    pass

@generate.command('cve')
@click.argument('cve_id')
@click.option('--method', '-m', type=click.Choice(['template', 'llm', 'hybrid', 'all']), 
              default='template', help='Rule generation method')
@click.option('--provider', '-p', type=click.Choice(['openai', 'anthropic', 'ollama']), 
              help='LLM provider for LLM-based generation')
@click.option('--model', help='Specific model to use')
@click.option('--force', '-f', is_flag=True, help='Force regeneration of existing rules')
@pass_config
def generate_cve(config, cve_id, method, provider, model, force):
    """Generate SIGMA rules for a specific CVE"""
    cmd = GenerateCommands(config)
    asyncio.run(cmd.generate_cve(cve_id, method, provider, model, force))

@generate.command('regenerate')
@click.option('--year', type=int, help='Regenerate rules for specific year')
@click.option('--method', '-m', type=click.Choice(['template', 'llm', 'hybrid', 'all']), 
              default='all', help='Rule generation method')
@click.option('--filter-quality', type=click.Choice(['excellent', 'good', 'fair']), 
              help='Only regenerate rules for CVEs with specific PoC quality')
@pass_config
def generate_regenerate(config, year, method, filter_quality):
    """Regenerate existing SIGMA rules"""
    cmd = GenerateCommands(config)
    asyncio.run(cmd.regenerate_rules(year, method, filter_quality))

# Search commands
@cli.group()
@pass_config
def search(config):
    """Search CVEs and SIGMA rules"""
    pass

@search.command('cve')
@click.argument('pattern')
@click.option('--year', type=int, help='Search within specific year')
@click.option('--severity', type=click.Choice(['low', 'medium', 'high', 'critical']), help='Filter by severity')
@click.option('--has-poc', is_flag=True, help='Only show CVEs with PoC data')
@click.option('--has-rules', is_flag=True, help='Only show CVEs with generated rules')
@click.option('--limit', '-l', default=20, help='Limit number of results')
@pass_config
def search_cve(config, pattern, year, severity, has_poc, has_rules, limit):
    """Search for CVEs by pattern"""
    cmd = SearchCommands(config)
    asyncio.run(cmd.search_cves(pattern, year, severity, has_poc, has_rules, limit))

@search.command('rules')
@click.argument('pattern')
@click.option('--rule-type', help='Filter by rule type (e.g., process, network, file)')
@click.option('--method', type=click.Choice(['template', 'llm', 'hybrid']), help='Filter by generation method')
@click.option('--limit', '-l', default=20, help='Limit number of results')
@pass_config
def search_rules(config, pattern, rule_type, method, limit):
    """Search for SIGMA rules by pattern"""
    cmd = SearchCommands(config)
    asyncio.run(cmd.search_rules(pattern, rule_type, method, limit))

# Statistics commands
@cli.group()
@pass_config
def stats(config):
    """Generate statistics and reports"""
    pass

@stats.command('overview')
@click.option('--year', type=int, help='Statistics for specific year')
@click.option('--output', '-o', type=click.Path(), help='Save output to file')
@pass_config
def stats_overview(config, year, output):
    """Generate overview statistics"""
    cmd = StatsCommands(config)
    asyncio.run(cmd.overview(year, output))

@stats.command('poc')
@click.option('--year', type=int, help='PoC statistics for specific year')
@pass_config
def stats_poc(config, year):
    """Generate PoC coverage statistics"""
    cmd = StatsCommands(config)
    asyncio.run(cmd.poc_stats(year))

@stats.command('rules')
@click.option('--year', type=int, help='Rule statistics for specific year')
@click.option('--method', type=click.Choice(['template', 'llm', 'hybrid']), help='Filter by generation method')
@pass_config
def stats_rules(config, year, method):
    """Generate rule generation statistics"""
    cmd = StatsCommands(config)
    asyncio.run(cmd.rule_stats(year, method))

# Export commands
@cli.group()
@pass_config
def export(config):
    """Export rules in various formats"""
    pass

@export.command('sigma')
@click.argument('output_dir', type=click.Path())
@click.option('--year', type=int, help='Export rules for specific year')
@click.option('--format', type=click.Choice(['yaml', 'json']), default='yaml', help='Output format')
@click.option('--method', type=click.Choice(['template', 'llm', 'hybrid']), help='Filter by generation method')
@pass_config
def export_sigma(config, output_dir, year, format, method):
    """Export SIGMA rules to a directory"""
    cmd = ExportCommands(config)
    asyncio.run(cmd.export_sigma_rules(output_dir, year, format, method))

@export.command('metadata')
@click.argument('output_file', type=click.Path())
@click.option('--year', type=int, help='Export metadata for specific year')
@click.option('--format', type=click.Choice(['json', 'csv']), default='json', help='Output format')
@pass_config
def export_metadata(config, output_file, year, format):
    """Export CVE metadata"""
    cmd = ExportCommands(config)
    asyncio.run(cmd.export_metadata(output_file, year, format))

# Migration commands (for transitioning from web app)
@cli.group()
@pass_config
def migrate(config):
    """Migration utilities for transitioning from web application"""
    pass

@migrate.command('from-database')
@click.option('--database-url', help='Database URL to migrate from')
@click.option('--batch-size', '-b', default=100, help='Batch size for migration')
@click.option('--dry-run', is_flag=True, help='Show what would be migrated without doing it')
@pass_config
def migrate_from_database(config, database_url, batch_size, dry_run):
    """Migrate data from existing database to file structure"""
    cmd = MigrateCommands(config)
    asyncio.run(cmd.migrate_from_database(database_url, batch_size, dry_run))

@migrate.command('validate')
@click.option('--year', type=int, help='Validate specific year')
@pass_config
def migrate_validate(config, year):
    """Validate migrated data integrity"""
    cmd = MigrateCommands(config)
    asyncio.run(cmd.validate_migration(year))

# Utility commands
@cli.command()
@pass_config
def version(config):
    """Show version information"""
    click.echo("SIGMA CLI v1.0.0")
    click.echo("CVE-SIGMA Auto Generator - File-based Edition")

@cli.command()
@pass_config
def config_init(config):
    """Initialize CLI configuration"""
    config_data = {
        'base_dir': str(config.base_dir),
        'api_keys': {
            'nvd_api_key': '',
            'github_token': '',
            'openai_api_key': '',
            'anthropic_api_key': ''
        },
        'llm_settings': {
            'default_provider': 'ollama',
            'default_model': 'llama3.2',
            'ollama_base_url': 'http://localhost:11434'
        },
        'processing': {
            'default_batch_size': 50,
            'default_methods': ['template']
        }
    }
    
    config.config_file.parent.mkdir(exist_ok=True)
    with open(config.config_file, 'w') as f:
        import yaml
        yaml.dump(config_data, f, default_flow_style=False)
    
    click.echo(f"Configuration initialized at {config.config_file}")
    click.echo("Please edit the configuration file to add your API keys and preferences.")

if __name__ == '__main__':
    cli()