bpmcdevitt/auto_sigma_rule_generator
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

MAJOR: Transform web application to professional CLI-based SIGMA rule generator

Browse source 
🎉 **Architecture Transformation (v2.0)**
- Complete migration from web app to professional CLI tool
- File-based SIGMA rule management system
- Git-friendly directory structure organized by year/CVE-ID
- Multiple rule variants per CVE (template, LLM, hybrid)

✨ **New CLI System**
- Professional command-line interface with Click framework
- 8 command groups: process, generate, search, stats, export, migrate
- Modular command architecture for maintainability
- Comprehensive help system and configuration management

📁 **File-Based Storage Architecture**
- Individual CVE directories: cves/YEAR/CVE-ID/
- Multiple SIGMA rule variants per CVE
- JSON metadata with processing history and PoC data
- Native YAML files perfect for version control

🚀 **Core CLI Commands**
- process: CVE processing and bulk operations
- generate: SIGMA rule generation with multiple methods
- search: Advanced CVE and rule searching with filters
- stats: Comprehensive statistics and analytics
- export: Multiple output formats for different workflows
- migrate: Database-to-file migration tools

🔧 **Migration Support**
- Complete migration utilities from web database
- Data validation and integrity checking
- Backward compatibility with existing processors
- Legacy web interface maintained for transition

📊 **Enhanced Features**
- Advanced search with complex filtering (severity, PoC presence, etc.)
- Multi-format exports (YAML, JSON, CSV)
- Comprehensive statistics and coverage reports
- File-based rule versioning and management

🎯 **Production Benefits**
- No database dependency - runs anywhere
- Perfect for cybersecurity teams using git workflows
- Direct integration with SIGMA ecosystems
- Portable architecture for CI/CD pipelines
- Multiple rule variants for different detection scenarios

📝 **Documentation Updates**
- Complete README rewrite for CLI-first approach
- Updated CLAUDE.md with new architecture details
- Detailed CLI documentation with examples
- Migration guides and troubleshooting

**Perfect for security teams wanting production-ready SIGMA rules with version control\! 🛡️**

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Brendan McDevitt 2025-07-21 13:11:03 -05:00
parent d51f3ea402
commit e579c91b5e
13 changed files with 2994 additions and 279 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

224
CLAUDE.md
View file

@ -2,124 +2,120 @@
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
## Project Overview ## Project Overview - CLI-Based Architecture (v2.0)
This is an enhanced CVE-SIGMA Auto Generator that automatically processes comprehensive CVE data and generates SIGMA rules for threat detection. The application now supports: This is an enhanced CVE-SIGMA Auto Generator that has been **transformed from a web application to a professional CLI tool** with file-based SIGMA rule management. The system now supports:
1. **Bulk NVD Data Processing**: Downloads and processes complete NVD JSON datasets (2002-2025) 1. **Bulk NVD Data Processing**: Downloads and processes complete NVD JSON datasets (2002-2025)
2. **nomi-sec PoC Integration**: Uses curated PoC data from github.com/nomi-sec/PoC-in-GitHub 2. **nomi-sec PoC Integration**: Uses curated PoC data from github.com/nomi-sec/PoC-in-GitHub
3. **Enhanced SIGMA Rule Generation**: Creates intelligent rules based on real exploit indicators 3. **Enhanced SIGMA Rule Generation**: Creates intelligent rules based on real exploit indicators
4. **Comprehensive Database Seeding**: Supports both bulk and incremental data updates 4. **Comprehensive Database Seeding**: Supports both bulk and incremental data updates
## Architecture ## Architecture - CLI-Based System
### **Current Primary Architecture (v2.0)**
- **CLI Interface**: Professional command-line tool (`cli/sigma_cli.py`) with modular commands
- **File-Based Storage**: Git-friendly YAML and JSON files organized by year/CVE-ID
- **Directory Structure**:
- `cves/YEAR/CVE-ID/`: Individual CVE directories with metadata and multiple rule variants
- `cli/commands/`: Modular command system (process, generate, search, stats, export, migrate)
- `reports/`: Generated statistics and export outputs
- **Data Processing**:
- Reuses existing backend processors for CVE fetching and analysis
- File-based rule generation with multiple variants per CVE
- CLI-driven bulk operations and incremental updates
- **Storage Format**:
- `metadata.json`: CVE information, PoC data, processing history
- `rule_*.sigma`: Multiple SIGMA rule variants (template, LLM, hybrid)
- `poc_analysis.json`: Extracted exploit indicators and analysis
### **Legacy Web Architecture (Optional, for Migration)**
- **Backend**: FastAPI with SQLAlchemy ORM (`backend/main.py`) - **Backend**: FastAPI with SQLAlchemy ORM (`backend/main.py`)
- **Frontend**: React with Tailwind CSS (`frontend/src/App.js`) - **Frontend**: React with Tailwind CSS (`frontend/src/App.js`)
- **Database**: PostgreSQL with enhanced schema: - **Database**: PostgreSQL (used only for migration to file-based system)
- `cves`: CVE information with PoC metadata and bulk processing fields
- `sigma_rules`: Enhanced SIGMA rules with quality scoring and nomi-sec data
- `rule_templates`: Template patterns for rule generation
- `bulk_processing_jobs`: Job tracking for bulk operations
- **Data Processing**:
- `nvd_bulk_processor.py`: NVD JSON dataset downloader and processor
- `nomi_sec_client.py`: nomi-sec PoC-in-GitHub API integration
- `enhanced_sigma_generator.py`: Advanced SIGMA rule generation
- `bulk_seeder.py`: Coordinated bulk seeding operations
- **Cache**: Redis (optional) - **Cache**: Redis (optional)
- **Deployment**: Docker Compose orchestration - **Deployment**: Docker Compose (maintained for migration purposes)
## Common Development Commands ## Common Development Commands
### Quick Start ### **CLI Quick Start (Recommended)**
```bash ```bash
# Recommended quick start # Install CLI dependencies
chmod +x start.sh pip install -r cli/requirements.txt
./start.sh
# Or using Make # Make CLI executable
make start chmod +x cli/sigma_cli.py
# Initialize configuration
./cli/sigma_cli.py config-init
# Test CLI installation
./cli/sigma_cli.py --help
``` ```
### Build and Run ### **CLI Primary Operations**
```bash ```bash
# Build and start all services # Process CVEs and generate SIGMA rules
docker-compose up -d --build ./cli/sigma_cli.py process year 2024 # Process specific year
./cli/sigma_cli.py process cve CVE-2024-0001 # Process specific CVE
./cli/sigma_cli.py process bulk --start-year 2020 # Bulk process years
./cli/sigma_cli.py process incremental --days 7 # Process recent changes
# Start individual services # Generate rules for existing CVEs
docker-compose up -d db redis # Database and cache only ./cli/sigma_cli.py generate cve CVE-2024-0001 --method all
docker-compose up -d backend # Backend API ./cli/sigma_cli.py generate regenerate --year 2024 --method llm
docker-compose up -d frontend # React frontend
# Search and analyze
./cli/sigma_cli.py search cve "buffer overflow" --severity critical --has-poc
./cli/sigma_cli.py search rules "powershell" --method llm
# Statistics and reports
./cli/sigma_cli.py stats overview --year 2024
./cli/sigma_cli.py stats poc --year 2024
./cli/sigma_cli.py stats rules --method template
# Export data
./cli/sigma_cli.py export sigma ./output-rules --format yaml --year 2024
./cli/sigma_cli.py export metadata ./reports/cve-data.csv --format csv
``` ```
### Development Mode ### **Migration from Web Application**
```bash ```bash
# Using Make # Migrate existing database to file structure
make dev ./cli/sigma_cli.py migrate from-database --database-url "postgresql://user:pass@localhost:5432/db"
# Or manually # Validate migrated data
docker-compose up -d db redis ./cli/sigma_cli.py migrate validate --year 2024
cd backend && pip install -r requirements.txt && uvicorn main:app --reload
cd frontend && npm install && npm start # Check migration statistics
./cli/sigma_cli.py stats overview
``` ```
### Bulk Processing Commands ### **Legacy Web Interface (Optional)**
```bash ```bash
# Run bulk seeding standalone # Start legacy web interface (for migration only)
cd backend && python bulk_seeder.py docker-compose up -d db redis backend frontend
# Bulk seed specific year range # Access points:
cd backend && python -c " # - Frontend: http://localhost:3000
import asyncio # - API: http://localhost:8000
from bulk_seeder import BulkSeeder # - API Docs: http://localhost:8000/docs
from main import SessionLocal # - Flower (Celery): http://localhost:5555
seeder = BulkSeeder(SessionLocal())
asyncio.run(seeder.full_bulk_seed(start_year=2020, end_year=2025))
"
# Incremental update only
cd backend && python -c "
import asyncio
from bulk_seeder import BulkSeeder
from main import SessionLocal
seeder = BulkSeeder(SessionLocal())
asyncio.run(seeder.incremental_update())
"
``` ```
### Frontend Commands ### **Development and Testing**
```bash ```bash
cd frontend # CLI with verbose logging
npm install # Install dependencies ./cli/sigma_cli.py --verbose process year 2024
npm start # Development server (port 3000)
npm run build # Production build
npm test # Run tests
```
### Backend Commands # Test individual commands
```bash ./cli/sigma_cli.py version
cd backend ./cli/sigma_cli.py config-init
pip install -r requirements.txt ./cli/sigma_cli.py stats overview
uvicorn main:app --reload # Development server (port 8000)
uvicorn main:app --host 0.0.0.0 --port 8000 # Production server
```
### Database Operations # Check file structure
```bash ls -la cves/2024/ # View processed CVEs
# Connect to database ls -la cves/2024/CVE-2024-0001/ # View individual CVE files
docker-compose exec db psql -U cve_user -d cve_sigma_db
# View logs
docker-compose logs -f backend
docker-compose logs -f frontend
```
### Other Make Commands
```bash
make stop # Stop all services
make restart # Restart all services
make logs # View application logs
make clean # Clean up containers and volumes
make setup # Initial setup (creates .env from .env.example)
``` ```
## Key Configuration ## Key Configuration
@ -135,7 +131,14 @@ make setup # Initial setup (creates .env from .env.example)
- `DATABASE_URL`: PostgreSQL connection string - `DATABASE_URL`: PostgreSQL connection string
- `REACT_APP_API_URL`: Backend API URL for frontend - `REACT_APP_API_URL`: Backend API URL for frontend
### Service URLs ### CLI Configuration
- **Configuration File**: `~/.sigma-cli/config.yaml` (auto-created with `config-init`)
- **Directory Structure**:
- `cves/YEAR/CVE-ID/`: Individual CVE data and rules
- `reports/`: Generated statistics and exports
- `cli/`: Command-line tool and modules
### Legacy Service URLs (If Using Web Interface)
- Frontend: http://localhost:3000 - Frontend: http://localhost:3000
- Backend API: http://localhost:8000 - Backend API: http://localhost:8000
- API Documentation: http://localhost:8000/docs - API Documentation: http://localhost:8000/docs
@ -165,27 +168,40 @@ make setup # Initial setup (creates .env from .env.example)
## Code Architecture Details ## Code Architecture Details
### Enhanced Backend Structure ### **CLI Structure (Primary)**
- **main.py**: Core FastAPI application with enhanced endpoints - **cli/sigma_cli.py**: Main executable CLI with Click framework
- **nvd_bulk_processor.py**: NVD JSON dataset downloader and processor - **cli/commands/**: Modular command system
- **nomi_sec_client.py**: nomi-sec PoC-in-GitHub API integration - `base_command.py`: Common functionality and file operations
- **enhanced_sigma_generator.py**: Advanced SIGMA rule generation with PoC data - `process_commands.py`: CVE processing and bulk operations
- **llm_client.py**: Multi-provider LLM integration using LangChain for AI-enhanced rule generation - `generate_commands.py`: SIGMA rule generation
- **bulk_seeder.py**: Coordinated bulk processing operations - `search_commands.py`: Search and filtering
- `stats_commands.py`: Statistics and reporting
- `export_commands.py`: Data export in multiple formats
- `migrate_commands.py`: Database migration tools
- **cli/config/**: Configuration management
- **cli/README.md**: Detailed CLI documentation
### Database Models (Enhanced) ### **File-Based Storage Structure**
- **CVE**: Enhanced with `poc_count`, `poc_data`, `bulk_processed`, `data_source` - **CVE Directories**: `cves/YEAR/CVE-ID/` with individual metadata and rule files
- **SigmaRule**: Enhanced with `poc_source`, `poc_quality_score`, `nomi_sec_data` - **Rule Variants**: Multiple SIGMA files per CVE (template, LLM, hybrid)
- **RuleTemplate**: Template patterns for rule generation - **Metadata Format**: JSON files with processing history and PoC data
- **BulkProcessingJob**: Job tracking for bulk operations - **Reports**: Generated statistics and export outputs
### Frontend Structure (Enhanced) ### **Legacy Backend Structure (For Migration)**
- **Three Main Tabs**: Dashboard, CVEs, SIGMA Rules - **main.py**: Core FastAPI application (maintained for migration)
- **Enhanced Dashboard**: PoC coverage statistics, data synchronization controls - **Data Processors**: Reused by CLI for CVE fetching and analysis
- **Enhanced CVE/Rule Display**: PoC quality indicators, exploit-based tagging - `nvd_bulk_processor.py`: NVD JSON dataset processing
- **Task Monitoring**: Via Flower dashboard (http://localhost:5555) - `nomi_sec_client.py`: nomi-sec PoC integration
- `enhanced_sigma_generator.py`: SIGMA rule generation
- `llm_client.py`: Multi-provider LLM integration
### Data Processing Flow ### **CLI-Based Data Processing Flow**
1. **CVE Processing**: NVD data fetch → File storage → PoC analysis → Metadata generation
2. **Rule Generation**: Template/LLM/Hybrid generation → Multiple rule variants → File storage
3. **Search & Analysis**: File-based searching → Statistics generation → Export capabilities
4. **Migration Support**: Database export → File conversion → Validation → Cleanup
### **Legacy Web Processing Flow (For Reference)**
1. **Bulk Seeding**: NVD JSON downloads → Database storage → nomi-sec PoC sync → Enhanced rule generation 1. **Bulk Seeding**: NVD JSON downloads → Database storage → nomi-sec PoC sync → Enhanced rule generation
2. **Incremental Updates**: NVD modified feeds → Update existing data → Sync new PoCs 2. **Incremental Updates**: NVD modified feeds → Update existing data → Sync new PoCs
3. **Rule Enhancement**: PoC analysis → Indicator extraction → Template selection → Enhanced SIGMA rule 3. **Rule Enhancement**: PoC analysis → Indicator extraction → Template selection → Enhanced SIGMA rule

487
README.md
View file

@ -1,252 +1,368 @@
# CVE-SIGMA Auto Generator # CVE-SIGMA Auto Generator - CLI Edition
Automated platform that generates SIGMA detection rules from CVE data using AI-enhanced exploit analysis. **Professional file-based SIGMA rule generation system for cybersecurity workflows**
Automated CLI tool that generates SIGMA detection rules from CVE data using AI-enhanced exploit analysis. Now optimized for git workflows and production SIGMA rule management with a file-based architecture.
## 🌟 **Major Architecture Update**
**🎉 New in v2.0**: Transformed from web application to professional CLI tool with file-based SIGMA rule management!
- **Git-Friendly**: Native YAML files perfect for version control
- **Industry Standard**: Direct integration with SIGMA ecosystems
- **Portable**: No database dependency, works anywhere
- **Scalable**: Process specific years/CVEs as needed
- **Multiple Variants**: Different generation methods per CVE
## ✨ Key Features ## ✨ Key Features
- **Bulk CVE Processing**: Complete NVD datasets (2002-2025) with nomi-sec PoC integration - **Bulk CVE Processing**: Complete NVD datasets (2002-2025) with nomi-sec PoC integration
- **AI-Powered Rule Generation**: Multi-provider LLM support (OpenAI, Anthropic, local Ollama) - **AI-Powered Rule Generation**: Multi-provider LLM support (OpenAI, Anthropic, local Ollama)
- **File-Based Storage**: Organized directory structure for each CVE and rule variant
- **Quality-Based PoC Analysis**: 5-tier quality scoring system for exploit reliability - **Quality-Based PoC Analysis**: 5-tier quality scoring system for exploit reliability
- **Real-time Monitoring**: Live job tracking and progress dashboard - **Advanced Search & Filtering**: Find CVEs and rules with complex criteria
- **Advanced Indicators**: Extract processes, files, network patterns from actual exploits - **Comprehensive Statistics**: Coverage reports and generation analytics
- **Export Tools**: Multiple output formats for different workflows
## 🚀 Quick Start ## 🚀 Quick Start
### Prerequisites ### Prerequisites
- Docker and Docker Compose - Python 3.8+ with pip
- (Optional) Docker for legacy web interface
- (Optional) API keys for enhanced features - (Optional) API keys for enhanced features
### Installation ### Installation
```bash ```bash
# Clone and start # Clone repository
git clone <repository-url> git clone <repository-url>
cd auto_sigma_rule_generator cd auto_sigma_rule_generator
chmod +x start.sh
./start.sh # Install CLI dependencies
pip install -r cli/requirements.txt
# Make CLI executable
chmod +x cli/sigma_cli.py
# Initialize configuration
./cli/sigma_cli.py config-init
``` ```
**Access Points:** ### First Run - Migration from Web App (If Applicable)
- Frontend: http://localhost:3000
- API: http://localhost:8000
- API Docs: http://localhost:8000/docs
### First Run
The application automatically:
1. Initializes database with rule templates
2. Fetches recent CVEs from NVD
3. Generates SIGMA rules with AI enhancement
4. Polls for new CVEs hourly
## 🎯 Usage
### Web Interface
- **Dashboard**: Statistics and system overview
- **CVEs**: Complete CVE listing with PoC data
- **SIGMA Rules**: Generated detection rules
- **Bulk Jobs**: Processing status and controls
### API Endpoints
#### Core Operations
```bash ```bash
# Fetch CVEs # If migrating from previous web version
curl -X POST http://localhost:8000/api/fetch-cves ./cli/sigma_cli.py migrate from-database --database-url "postgresql://user:pass@localhost:5432/db"
# Bulk processing # Validate migration
curl -X POST http://localhost:8000/api/bulk-seed ./cli/sigma_cli.py migrate validate
curl -X POST http://localhost:8000/api/incremental-update
# LLM-enhanced rules # Or start fresh with new CVE processing
curl -X POST http://localhost:8000/api/llm-enhanced-rules ./cli/sigma_cli.py process year 2024
``` ```
#### Data Access ## 🎯 CLI Usage
- `GET /api/cves` - List CVEs
- `GET /api/sigma-rules` - List rules
- `GET /api/stats` - Statistics
- `GET /api/llm-status` - LLM provider status
## ⚙️ Configuration ### **Core Commands**
### Environment Variables
**Core Settings**
```bash ```bash
DATABASE_URL=postgresql://user:pass@db:5432/dbname # Process CVEs and generate rules
NVD_API_KEY=your_nvd_key # Optional: 5→50 req/30s ./cli/sigma_cli.py process year 2024 # Process specific year
GITHUB_TOKEN=your_github_token # Optional: Enhanced PoC analysis ./cli/sigma_cli.py process cve CVE-2024-0001 # Process specific CVE
./cli/sigma_cli.py process bulk --start-year 2020 # Bulk process multiple years
./cli/sigma_cli.py process incremental --days 7 # Process recent changes
# Generate rules for existing CVEs
./cli/sigma_cli.py generate cve CVE-2024-0001 --method all # All generation methods
./cli/sigma_cli.py generate regenerate --year 2024 --method llm # Regenerate with LLM
# Search CVEs and rules
./cli/sigma_cli.py search cve "buffer overflow" --severity critical --has-poc
./cli/sigma_cli.py search rules "powershell" --method llm
# View statistics and reports
./cli/sigma_cli.py stats overview --year 2024 --output ./reports/2024-stats.json
./cli/sigma_cli.py stats poc --year 2024 # PoC coverage statistics
./cli/sigma_cli.py stats rules --method template # Rule generation statistics
# Export data
./cli/sigma_cli.py export sigma ./output-rules --format yaml --year 2024
./cli/sigma_cli.py export metadata ./reports/cve-data.csv --format csv
``` ```
**LLM Configuration** ### **Available Generation Methods**
```bash - `template` - Template-based rule generation
LLM_PROVIDER=ollama # Default: ollama (local) - `llm` - AI/LLM-enhanced generation (OpenAI, Anthropic, Ollama)
LLM_MODEL=llama3.2 # Provider-specific model - `hybrid` - Combined template + LLM approach
OLLAMA_BASE_URL=http://ollama:11434 - `all` - Generate all variants
# External providers (optional) ## 📁 File Structure
OPENAI_API_KEY=your_openai_key
ANTHROPIC_API_KEY=your_anthropic_key The CLI organizes everything in a clean, git-friendly structure:
```
auto_sigma_rule_generator/
├── cves/ # CVE data organized by year
│ ├── 2024/
│ │ ├── CVE-2024-0001/
│ │ │ ├── metadata.json # CVE info & generation metadata
│ │ │ ├── rule_template.sigma # Template-based rule
│ │ │ ├── rule_llm_openai.sigma # OpenAI-generated rule
│ │ │ ├── rule_llm_anthropic.sigma# Anthropic-generated rule
│ │ │ ├── rule_hybrid.sigma # Hybrid-generated rule
│ │ │ └── poc_analysis.json # PoC analysis data
│ │ └── CVE-2024-0002/...
│ └── 2023/...
├── cli/ # CLI tool and commands
│ ├── sigma_cli.py # Main CLI executable
│ ├── commands/ # Command modules
│ └── README.md # Detailed CLI documentation
└── reports/ # Generated reports and exports
``` ```
### API Keys Setup ### **File Formats**
**NVD API** (Recommended) **metadata.json** - CVE information and processing history
1. Get key: https://nvd.nist.gov/developers/request-an-api-key ```json
2. Add to `.env`: `NVD_API_KEY=your_key` {
3. Benefit: 10x rate limit increase "cve_info": {
"cve_id": "CVE-2024-0001",
"description": "Remote code execution vulnerability...",
"cvss_score": 9.8,
"severity": "critical",
"published_date": "2024-01-01T00:00:00Z"
},
"poc_data": {
"poc_count": 3,
"poc_data": {"nomi_sec": [...], "github": [...]}
},
"rule_generation": {
"template": {"generated_at": "2024-01-01T12:00:00Z"},
"llm_openai": {"generated_at": "2024-01-01T12:30:00Z"}
}
}
```
**GitHub Token** (Optional) **SIGMA Rule Files** - Ready-to-use detection rules
1. Create: https://github.com/settings/tokens (public_repo scope)
2. Add to `.env`: `GITHUB_TOKEN=your_token`
3. Benefit: Enhanced exploit-based rules
**LLM APIs** (Optional)
- **Local Ollama**: No setup required (default)
- **OpenAI**: Get key from https://platform.openai.com/api-keys
- **Anthropic**: Get key from https://console.anthropic.com/
## 🧠 Rule Generation
### AI-Enhanced Generation
1. **PoC Analysis**: LLM analyzes actual exploit code
2. **Intelligent Detection**: Creates sophisticated SIGMA rules
3. **Context Awareness**: Maps CVE descriptions to detection patterns
4. **Validation**: Automatic SIGMA syntax verification
5. **Fallback**: Template-based generation if LLM unavailable
### Quality Tiers
- **Excellent** (80+ pts): High-quality PoCs with recent updates
- **Good** (60-79 pts): Moderate quality indicators
- **Fair** (40-59 pts): Basic PoCs with some validation
- **Poor** (20-39 pts): Minimal quality indicators
- **Very Poor** (<20 pts): Low-quality PoCs
### Rule Types
- 🤖 **AI-Enhanced**: LLM-generated with PoC analysis
- 🔍 **Exploit-Based**: Template + GitHub exploit indicators
- ⚡ **Basic**: CVE description only
### Example Output
```yaml ```yaml
title: CVE-2025-1234 AI-Enhanced Detection # rule_llm_openai.sigma
description: Detection for CVE-2025-1234 RCE [AI-Enhanced with PoC analysis] title: CVE-2024-0001 Remote Code Execution Detection
id: 12345678-1234-5678-9abc-123456789012
status: experimental
description: Detects exploitation attempts for CVE-2024-0001
author: CVE-SIGMA Auto Generator (OpenAI Enhanced)
date: 2024/01/01
references:
- https://nvd.nist.gov/vuln/detail/CVE-2024-0001
tags: tags:
- attack.t1059.001 - attack.t1059.001
- cve-2025-1234 - cve.2024.0001
- ai.enhanced - ai.enhanced
logsource:
category: process_creation
product: windows
detection: detection:
selection_process: selection:
Image|endswith: '\powershell.exe' Image|endswith: '\powershell.exe'
CommandLine|contains: CommandLine|contains:
- '-EncodedCommand' - '-EncodedCommand'
- 'bypass' - 'bypass'
selection_network: condition: selection
DestinationPort: [443, 80] falsepositives:
condition: selection_process and selection_network - Legitimate administrative scripts
level: high level: high
``` ```
## 🛠️ Development ## ⚙️ Configuration
### Local Development ### CLI Configuration (`~/.sigma-cli/config.yaml`)
```bash
# Start dependencies
docker-compose up -d db redis ollama
# Backend ```yaml
cd backend && pip install -r requirements.txt # API Keys for enhanced functionality
uvicorn main:app --reload api_keys:
nvd_api_key: "your_nvd_key" # Optional: 5→50 req/30s rate limit
github_token: "your_github_token" # Optional: Enhanced PoC analysis
openai_api_key: "your_openai_key" # Optional: AI rule generation
anthropic_api_key: "your_anthropic_key" # Optional: AI rule generation
# Frontend # LLM Settings
cd frontend && npm install && npm start llm_settings:
default_provider: "ollama" # Default: ollama (local)
default_model: "llama3.2" # Provider-specific model
ollama_base_url: "http://localhost:11434"
# Processing Settings
processing:
default_batch_size: 50 # CVEs per batch
default_methods: ["template"] # Default generation methods
``` ```
### Testing LLM Integration ### API Keys Setup
**NVD API Key** (Recommended)
- Get key: https://nvd.nist.gov/developers/request-an-api-key
- Benefit: 10x rate limit increase (5 → 50 requests/30s)
**GitHub Token** (Optional)
- Create: https://github.com/settings/tokens (public_repo scope)
- Benefit: Enhanced PoC analysis and exploit indicators
**LLM APIs** (Optional)
- **Local Ollama**: No setup required (default) - runs locally
- **OpenAI**: Get key from https://platform.openai.com/api-keys
- **Anthropic**: Get key from https://console.anthropic.com/
## 🧠 AI-Enhanced Rule Generation
### How It Works
1. **CVE Analysis**: Extract vulnerability details from NVD data
2. **PoC Collection**: Gather exploit code from nomi-sec, GitHub, ExploitDB
3. **Quality Assessment**: Score PoCs based on stars, recency, completeness
4. **AI Enhancement**: LLM analyzes actual exploit code to create detection logic
5. **SIGMA Generation**: Produce valid, tested SIGMA rules with proper syntax
6. **Multi-Variant Output**: Generate template, LLM, and hybrid versions
### Quality Tiers
- **Excellent** (80+ pts): High-star PoCs with recent updates, detailed analysis
- **Good** (60-79 pts): Moderate quality with some validation
- **Fair** (40-59 pts): Basic PoCs with minimal indicators
- **Poor** (20-39 pts): Low-quality or outdated PoCs
- **Very Poor** (<20 pts): Minimal or unreliable PoCs
### Rule Variants Generated
- 🤖 **AI-Enhanced** (`rule_llm_*.sigma`): LLM analysis of actual exploit code
- 🔧 **Template-Based** (`rule_template.sigma`): Pattern-based generation
- ⚡ **Hybrid** (`rule_hybrid.sigma`): Best of both approaches
## 📊 Advanced Features
### Search & Analytics
```bash ```bash
# Check Ollama # Complex CVE searches
curl http://localhost:11434/api/tags ./cli/sigma_cli.py search cve "remote code execution" \
--year 2024 --severity critical --has-poc --has-rules --limit 50
# Test LLM status # Rule analysis
curl http://localhost:8000/api/llm-status ./cli/sigma_cli.py search rules "powershell" \
--rule-type process --method llm --limit 20
# Switch providers # Comprehensive statistics
curl -X POST http://localhost:8000/api/llm-switch \ ./cli/sigma_cli.py stats overview # Overall system stats
-H "Content-Type: application/json" \ ./cli/sigma_cli.py stats poc --year 2024 # PoC coverage analysis
-d '{"provider": "ollama", "model": "llama3.2"}' ./cli/sigma_cli.py stats rules --method llm # AI generation statistics
``` ```
## 📊 Architecture ### Export & Integration
```bash
# Export for SIEM integration
./cli/sigma_cli.py export sigma ./siem-rules \
--format yaml --year 2024 --method llm
- **Backend**: FastAPI + SQLAlchemy ORM # Metadata for analysis
- **Frontend**: React + Tailwind CSS ./cli/sigma_cli.py export metadata ./analysis/cve-data.csv \
- **Database**: PostgreSQL with enhanced schema --format csv --year 2024
- **Cache**: Redis (optional)
- **LLM**: Ollama container + multi-provider support
- **Deployment**: Docker Compose
### Enhanced Database Schema # Consolidated ruleset
- **CVEs**: PoC metadata, bulk processing fields ./cli/sigma_cli.py export ruleset ./complete-rules.json \
- **SIGMA Rules**: Quality scoring, nomi-sec data --year 2024 --include-metadata
- **Rule Templates**: Pattern templates for generation ```
- **Bulk Jobs**: Job tracking and status
## 🛠️ Development & Legacy Support
### CLI Development
The new CLI system is built with:
- **Click**: Professional CLI framework
- **Modular Commands**: Separate modules for each command group
- **Async Processing**: Efficient handling of bulk operations
- **File-Based Storage**: Git-friendly YAML and JSON formats
### Legacy Web Interface (Optional)
The original web interface is still available for migration purposes:
```bash
# Start legacy web interface (if needed for migration)
docker-compose up -d db redis backend frontend
# Access points:
# - Frontend: http://localhost:3000
# - API: http://localhost:8000
# - Flower (Celery): http://localhost:5555
```
### Migration Path
1. **Export Data**: Use CLI migration tools to export from database
2. **Validate**: Verify all data transferred correctly
3. **Switch**: Use CLI for all new operations
4. **Cleanup**: Optionally remove web components
## 🔧 Troubleshooting ## 🔧 Troubleshooting
### Common Issues ### Common Issues
**CVE Fetch Issues** **CLI Import Errors**
- Verify NVD API key in `.env` - Ensure you're running from project root directory
- Check API connectivity: Use "Test NVD API" button - Install dependencies: `pip install -r cli/requirements.txt`
- Review logs: `docker-compose logs -f backend` - Check Python version (3.8+ required)
**CVE Processing Failures**
- Verify NVD API key in configuration
- Check network connectivity and rate limits
- Use `--verbose` flag for detailed logging
**No Rules Generated** **No Rules Generated**
- Ensure LLM provider is accessible - Ensure LLM provider is accessible (test with `./cli/sigma_cli.py stats overview`)
- Check `/api/llm-status` for provider health - Check PoC data availability with `--has-poc` filter
- Verify PoC data quality in CVE details - Verify API keys for external LLM providers
**Performance Issues** **File Permission Issues**
- Start with recent years (2020+) for faster initial setup - Ensure write permissions to `cves/` directory
- Use smaller batch sizes for bulk operations - Check CLI executable permissions: `chmod +x cli/sigma_cli.py`
- Monitor system resources during processing
**Port Conflicts** ### Performance Optimization
- Default ports: 3000 (frontend), 8000 (backend), 5432 (db) - Use `--batch-size` parameter for large datasets
- Modify `docker-compose.yml` if ports are in use - Process recent years first (2020+) for faster initial results
- Use `incremental` processing for regular updates
- Monitor system resources during bulk operations
### Rate Limits ## 🛡️ Security Best Practices
- **NVD API**: 5/30s (no key) → 50/30s (with key)
- **nomi-sec API**: 1/second (built-in limiting)
- **GitHub API**: 60/hour (no token) → 5000/hour (with token)
## 🛡️ Security - Store API keys in configuration file (`~/.sigma-cli/config.yaml`)
- Validate generated rules before production deployment
- Rules marked as "experimental" require analyst review
- Use version control to track rule changes and improvements
- Regularly update PoC data sources for current threat landscape
- Store API keys in environment variables ## 📈 Monitoring & Maintenance
- Validate generated rules before production deployment
- Rules marked as "experimental" - require analyst review
- Use strong database passwords in production
## 📈 Monitoring
```bash ```bash
# View logs # System health checks
docker-compose logs -f backend ./cli/sigma_cli.py stats overview # Overall system status
docker-compose logs -f frontend ./cli/sigma_cli.py migrate validate # Data integrity check
# Check service health # Regular maintenance
docker-compose ps ./cli/sigma_cli.py process incremental --days 7 # Weekly updates
./cli/sigma_cli.py generate regenerate --filter-quality excellent # Refresh high-quality rules
# Monitor bulk jobs # Performance monitoring
curl http://localhost:8000/api/bulk-status ./cli/sigma_cli.py stats rules --year 2024 # Generation statistics
./cli/sigma_cli.py stats poc --year 2024 # Coverage analysis
``` ```
## 🗺️ Roadmap ## 🗺️ Roadmap
- [ ] Custom rule template editor **CLI Enhancements**
- [ ] Rule quality scoring and validation
- [ ] Custom template editor
- [ ] Integration with popular SIEM platforms
- [ ] Advanced MITRE ATT&CK mapping - [ ] Advanced MITRE ATT&CK mapping
- [ ] SIEM platform export - [ ] Threat intelligence feed integration
- [ ] ML-based rule optimization
- [ ] Threat intelligence integration **Export Features**
- [ ] Splunk app export format
- [ ] Elastic Stack integration
- [ ] QRadar rule format
- [ ] YARA rule generation
- [ ] IOC extraction
## 📝 License ## 📝 License
@ -254,13 +370,34 @@ MIT License - see LICENSE file for details.
## 🤝 Contributing ## 🤝 Contributing
1. Fork repository 1. Fork the repository
2. Create feature branch 2. Create a feature branch (`git checkout -b feature/amazing-feature`)
3. Add tests and documentation 3. Test with both CLI and legacy systems
4. Submit pull request 4. Add tests and documentation
5. Submit a pull request
## 📞 Support ## 📞 Support
- Check troubleshooting section **CLI Issues**
- Review application logs - Check `cli/README.md` for detailed CLI documentation
- Open GitHub issue for bugs/questions - Use `--verbose` flag for debugging
- Ensure proper configuration in `~/.sigma-cli/config.yaml`
**General Support**
- Review troubleshooting section above
- Check application logs with `--verbose`
- Open GitHub issue with specific error details
---
## 🎉 **What's New in v2.0**
✅ **Complete CLI System** - Professional command-line interface
✅ **File-Based Storage** - Git-friendly YAML and JSON files
✅ **Multiple Rule Variants** - Template, AI, and hybrid generation
✅ **Advanced Search** - Complex filtering and analytics
✅ **Export Tools** - Multiple output formats for different workflows
✅ **Migration Tools** - Seamless transition from web application
✅ **Portable Architecture** - No database dependency, runs anywhere
**Perfect for cybersecurity teams who want production-ready SIGMA rules with version control integration! 🚀**

220
cli/README.md Normal file
View file

@ -0,0 +1,220 @@
# SIGMA CLI - CVE-SIGMA Auto Generator
A command-line interface for processing CVEs and generating SIGMA detection rules in a file-based directory structure.
## Quick Start
```bash
# Make CLI executable
chmod +x cli/sigma_cli.py
# Initialize configuration
./cli/sigma_cli.py config-init
# Migrate data from existing database (if applicable)
./cli/sigma_cli.py migrate from-database
# Process CVEs for a specific year
./cli/sigma_cli.py process year 2024
# Generate rules for a specific CVE
./cli/sigma_cli.py generate cve CVE-2024-0001
# Search CVEs
./cli/sigma_cli.py search cve "buffer overflow"
# View statistics
./cli/sigma_cli.py stats overview
# Export rules
./cli/sigma_cli.py export sigma ./output/rules
```
## Directory Structure
```
auto_sigma_rule_generator/
├── cves/
│ ├── 2024/
│ │ ├── CVE-2024-0001/
│ │ │ ├── metadata.json
│ │ │ ├── rule_template.sigma
│ │ │ ├── rule_llm_openai.sigma
│ │ │ └── poc_analysis.json
│ │ └── CVE-2024-0002/...
│ └── 2023/...
├── cli/
│ ├── sigma_cli.py (main CLI)
│ ├── commands/ (command modules)
│ └── config/ (CLI configuration)
└── reports/ (generated reports)
```
## Available Commands
### Process Commands
- `process year <year>` - Process all CVEs for a year
- `process cve <cve-id>` - Process specific CVE
- `process bulk` - Bulk process multiple years
- `process incremental` - Process recent changes
### Generate Commands
- `generate cve <cve-id>` - Generate rules for CVE
- `generate regenerate` - Regenerate existing rules
### Search Commands
- `search cve <pattern>` - Search CVEs
- `search rules <pattern>` - Search SIGMA rules
### Statistics Commands
- `stats overview` - General statistics
- `stats poc` - PoC coverage statistics
- `stats rules` - Rule generation statistics
### Export Commands
- `export sigma <dir>` - Export SIGMA rules
- `export metadata <file>` - Export CVE metadata
### Migration Commands
- `migrate from-database` - Migrate from web app database
- `migrate validate` - Validate migrated data
## Configuration
Edit `~/.sigma-cli/config.yaml` to configure API keys and settings:
```yaml
api_keys:
nvd_api_key: "your-nvd-key"
github_token: "your-github-token"
openai_api_key: "your-openai-key"
anthropic_api_key: "your-anthropic-key"
llm_settings:
default_provider: "ollama"
default_model: "llama3.2"
ollama_base_url: "http://localhost:11434"
processing:
default_batch_size: 50
default_methods: ["template"]
```
## Installation
```bash
# Install dependencies
pip install -r cli/requirements.txt
# Or if you're in a virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\\Scripts\\activate
pip install -r cli/requirements.txt
```
## Examples
### Migration from Web Application
```bash
# Migrate existing data
./cli/sigma_cli.py migrate from-database --database-url "postgresql://user:pass@localhost:5432/db"
# Validate migration
./cli/sigma_cli.py migrate validate
# Check migration statistics
./cli/sigma_cli.py stats overview
```
### Processing CVEs
```bash
# Process a specific year with multiple methods
./cli/sigma_cli.py process year 2024 --method template --method llm
# Process a specific CVE with force regeneration
./cli/sigma_cli.py process cve CVE-2024-12345 --force
# Bulk process with specific batch size
./cli/sigma_cli.py process bulk --start-year 2020 --end-year 2024 --batch-size 100
```
### Searching and Analysis
```bash
# Search for CVEs with specific patterns
./cli/sigma_cli.py search cve "remote code execution" --severity critical --has-poc
# Search SIGMA rules
./cli/sigma_cli.py search rules "powershell" --method llm
# Generate comprehensive statistics
./cli/sigma_cli.py stats overview --year 2024 --output ./reports/2024-stats.json
```
### Exporting Data
```bash
# Export all SIGMA rules as YAML
./cli/sigma_cli.py export sigma ./output/sigma-rules --format yaml
# Export CVE metadata as CSV
./cli/sigma_cli.py export metadata ./reports/cve-data.csv --format csv
# Export specific year and method
./cli/sigma_cli.py export sigma ./output/2024-llm-rules --year 2024 --method llm
```
## File Formats
### metadata.json Structure
```json
{
"cve_info": {
"cve_id": "CVE-2024-0001",
"description": "...",
"cvss_score": 9.8,
"severity": "critical"
},
"poc_data": {
"poc_count": 3,
"poc_data": {...}
},
"rule_generation": {
"template": {"generated_at": "..."},
"llm_openai": {"generated_at": "..."}
}
}
```
### SIGMA Rule Files
- `rule_template.sigma` - Template-based generation
- `rule_llm_openai.sigma` - OpenAI LLM generation
- `rule_llm_anthropic.sigma` - Anthropic LLM generation
- `rule_hybrid.sigma` - Hybrid generation method
## Development
The CLI is built using Click and follows a modular command structure:
- `sigma_cli.py` - Main CLI entry point
- `commands/base_command.py` - Base functionality
- `commands/process_commands.py` - CVE processing
- `commands/migrate_commands.py` - Database migration
- `commands/search_commands.py` - Search functionality
- `commands/stats_commands.py` - Statistics generation
- `commands/export_commands.py` - Data export
## Troubleshooting
### Common Issues
1. **Import errors**: Make sure you're running from the project root
2. **Permission errors**: Ensure directories are writable
3. **Database connection**: Check DATABASE_URL environment variable
4. **API limits**: Configure API keys for higher rate limits
### Debug Mode
```bash
# Enable verbose logging
./cli/sigma_cli.py --verbose <command>
# Check configuration
./cli/sigma_cli.py config-init
```

21
cli/commands/__init__.py Normal file
View file

@ -0,0 +1,21 @@
"""
CLI Commands Package
Contains all command implementations for the SIGMA CLI tool.
"""
from .process_commands import ProcessCommands
from .generate_commands import GenerateCommands
from .search_commands import SearchCommands
from .stats_commands import StatsCommands
from .export_commands import ExportCommands
from .migrate_commands import MigrateCommands
__all__ = [
'ProcessCommands',
'GenerateCommands',
'SearchCommands',
'StatsCommands',
'ExportCommands',
'MigrateCommands'
]

226
cli/commands/base_command.py Normal file
View file

@ -0,0 +1,226 @@
"""
Base Command Class
Provides common functionality for all CLI command classes.
"""
import json
import logging
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Optional, Any
import yaml
logger = logging.getLogger(__name__)
class BaseCommand:
"""Base class for all CLI commands"""
def __init__(self, config):
self.config = config
self.logger = logger
def get_cve_directory(self, cve_id: str) -> Path:
"""Get the directory path for a specific CVE"""
year = cve_id.split('-')[1] # Extract year from CVE-YYYY-NNNN
return self.config.cves_dir / year / cve_id
def ensure_cve_directory(self, cve_id: str) -> Path:
"""Ensure CVE directory exists and return its path"""
cve_dir = self.get_cve_directory(cve_id)
cve_dir.mkdir(parents=True, exist_ok=True)
return cve_dir
def load_cve_metadata(self, cve_id: str) -> Optional[Dict]:
"""Load metadata for a specific CVE"""
cve_dir = self.get_cve_directory(cve_id)
metadata_file = cve_dir / "metadata.json"
if not metadata_file.exists():
return None
try:
with open(metadata_file, 'r') as f:
return json.load(f)
except Exception as e:
self.logger.error(f"Error loading metadata for {cve_id}: {e}")
return None
def save_cve_metadata(self, cve_id: str, metadata: Dict) -> bool:
"""Save metadata for a specific CVE"""
cve_dir = self.ensure_cve_directory(cve_id)
metadata_file = cve_dir / "metadata.json"
# Update timestamps
if 'updated_at' not in metadata:
metadata['updated_at'] = datetime.utcnow().isoformat()
try:
with open(metadata_file, 'w') as f:
json.dump(metadata, f, indent=2, default=str)
return True
except Exception as e:
self.logger.error(f"Error saving metadata for {cve_id}: {e}")
return False
def list_cve_rules(self, cve_id: str) -> List[str]:
"""List all SIGMA rule files for a CVE"""
cve_dir = self.get_cve_directory(cve_id)
if not cve_dir.exists():
return []
rule_files = []
for file in cve_dir.glob("rule_*.sigma"):
rule_files.append(file.name)
return sorted(rule_files)
def load_sigma_rule(self, cve_id: str, rule_file: str) -> Optional[str]:
"""Load a specific SIGMA rule file content"""
cve_dir = self.get_cve_directory(cve_id)
rule_path = cve_dir / rule_file
if not rule_path.exists():
return None
try:
with open(rule_path, 'r') as f:
return f.read()
except Exception as e:
self.logger.error(f"Error loading rule {rule_file} for {cve_id}: {e}")
return None
def save_sigma_rule(self, cve_id: str, rule_file: str, content: str) -> bool:
"""Save a SIGMA rule file"""
cve_dir = self.ensure_cve_directory(cve_id)
rule_path = cve_dir / rule_file
try:
with open(rule_path, 'w') as f:
f.write(content)
# Update metadata to track this rule file
metadata = self.load_cve_metadata(cve_id) or {}
if 'file_manifest' not in metadata:
metadata['file_manifest'] = []
if rule_file not in metadata['file_manifest']:
metadata['file_manifest'].append(rule_file)
# Update rule generation info
if 'rule_generation' not in metadata:
metadata['rule_generation'] = {}
method = rule_file.replace('rule_', '').replace('.sigma', '')
metadata['rule_generation'][method] = {
'generated_at': datetime.utcnow().isoformat(),
'file': rule_file
}
self.save_cve_metadata(cve_id, metadata)
return True
except Exception as e:
self.logger.error(f"Error saving rule {rule_file} for {cve_id}: {e}")
return False
def get_all_cves(self, year: Optional[int] = None) -> List[str]:
"""Get list of all CVEs, optionally filtered by year"""
cves = []
if year:
year_dir = self.config.cves_dir / str(year)
if year_dir.exists():
for cve_dir in year_dir.iterdir():
if cve_dir.is_dir() and cve_dir.name.startswith('CVE-'):
cves.append(cve_dir.name)
else:
# Get all CVEs across all years
for year_dir in self.config.cves_dir.iterdir():
if year_dir.is_dir() and year_dir.name.isdigit():
for cve_dir in year_dir.iterdir():
if cve_dir.is_dir() and cve_dir.name.startswith('CVE-'):
cves.append(cve_dir.name)
return sorted(cves)
def get_years_with_data(self) -> List[int]:
"""Get list of years that have CVE data"""
years = []
for year_dir in self.config.cves_dir.iterdir():
if year_dir.is_dir() and year_dir.name.isdigit():
# Check if year directory has any CVE subdirectories
has_cves = any(
cve_dir.is_dir() and cve_dir.name.startswith('CVE-')
for cve_dir in year_dir.iterdir()
)
if has_cves:
years.append(int(year_dir.name))
return sorted(years)
def validate_cve_id(self, cve_id: str) -> bool:
"""Validate CVE ID format"""
import re
pattern = r'^CVE-\d{4}-\d{4,}$'
return bool(re.match(pattern, cve_id))
def print_table(self, headers: List[str], rows: List[List[str]], title: Optional[str] = None):
"""Print a formatted table"""
import click
if title:
click.echo(f"\n{title}")
click.echo("=" * len(title))
if not rows:
click.echo("No data found.")
return
# Calculate column widths
widths = [len(h) for h in headers]
for row in rows:
for i, cell in enumerate(row):
if i < len(widths):
widths[i] = max(widths[i], len(str(cell)))
# Print headers
header_line = " | ".join(h.ljust(w) for h, w in zip(headers, widths))
click.echo(header_line)
click.echo("-" * len(header_line))
# Print rows
for row in rows:
row_line = " | ".join(str(cell).ljust(w) for cell, w in zip(row, widths))
click.echo(row_line)
def format_json_output(self, data: Any, pretty: bool = True) -> str:
"""Format data as JSON"""
if pretty:
return json.dumps(data, indent=2, default=str)
else:
return json.dumps(data, default=str)
def format_yaml_output(self, data: Any) -> str:
"""Format data as YAML"""
return yaml.dump(data, default_flow_style=False)
def success(self, message: str):
"""Print success message"""
import click
click.echo(click.style(f"✓ {message}", fg='green'))
def error(self, message: str):
"""Print error message"""
import click
click.echo(click.style(f"✗ {message}", fg='red'), err=True)
def warning(self, message: str):
"""Print warning message"""
import click
click.echo(click.style(f"⚠ {message}", fg='yellow'))
def info(self, message: str):
"""Print info message"""
import click
click.echo(click.style(f"ℹ {message}", fg='blue'))

282
cli/commands/export_commands.py Normal file
View file

@ -0,0 +1,282 @@
"""
Export Commands
Commands for exporting SIGMA rules and CVE data in various formats.
"""
import json
import csv
import shutil
from pathlib import Path
from typing import Dict, List, Optional
from .base_command import BaseCommand
class ExportCommands(BaseCommand):
"""Commands for exporting data"""
async def export_sigma_rules(self, output_dir: str, year: Optional[int],
format_type: str, method: Optional[str]):
"""Export SIGMA rules to a directory"""
output_path = Path(output_dir)
output_path.mkdir(parents=True, exist_ok=True)
self.info(f"Exporting SIGMA rules to: {output_path}")
self.info(f"Format: {format_type}")
if year:
self.info(f"Filtering by year: {year}")
if method:
self.info(f"Filtering by method: {method}")
# Get CVEs to export
cves = self.get_all_cves(year)
if not cves:
self.warning("No CVEs found to export")
return
exported_count = 0
skipped_count = 0
for cve_id in cves:
try:
rules = self.list_cve_rules(cve_id)
if method:
# Filter rules by method
rules = [r for r in rules if method.lower() in r.lower()]
if not rules:
skipped_count += 1
continue
# Create CVE directory in export location
cve_export_dir = output_path / cve_id
cve_export_dir.mkdir(exist_ok=True)
for rule_file in rules:
rule_content = self.load_sigma_rule(cve_id, rule_file)
if not rule_content:
continue
if format_type == 'yaml':
# Export as YAML (original format)
export_file = cve_export_dir / rule_file
with open(export_file, 'w') as f:
f.write(rule_content)
elif format_type == 'json':
# Convert YAML to JSON (basic conversion)
try:
import yaml
rule_dict = yaml.safe_load(rule_content)
export_file = cve_export_dir / rule_file.replace('.sigma', '.json')
with open(export_file, 'w') as f:
json.dump(rule_dict, f, indent=2)
except Exception as e:
self.error(f"Error converting {rule_file} to JSON: {e}")
continue
exported_count += 1
# Export metadata for context
metadata = self.load_cve_metadata(cve_id)
if metadata:
metadata_file = cve_export_dir / "metadata.json"
with open(metadata_file, 'w') as f:
json.dump(metadata, f, indent=2, default=str)
if exported_count % 50 == 0:
self.info(f"Exported {exported_count} rules...")
except Exception as e:
self.error(f"Error exporting rules for {cve_id}: {e}")
skipped_count += 1
self.success(f"Export completed!")
self.success(f"Exported {exported_count} rules from {len(cves) - skipped_count} CVEs")
self.success(f"Skipped {skipped_count} CVEs (no matching rules)")
async def export_metadata(self, output_file: str, year: Optional[int], format_type: str):
"""Export CVE metadata"""
output_path = Path(output_file)
output_path.parent.mkdir(parents=True, exist_ok=True)
self.info(f"Exporting CVE metadata to: {output_path}")
self.info(f"Format: {format_type}")
if year:
self.info(f"Filtering by year: {year}")
# Get CVEs to export
cves = self.get_all_cves(year)
if not cves:
self.warning("No CVEs found to export")
return
metadata_list = []
for cve_id in cves:
try:
metadata = self.load_cve_metadata(cve_id)
if not metadata:
continue
# Flatten metadata for export
export_record = self._flatten_metadata(metadata)
export_record['rules_count'] = len(self.list_cve_rules(cve_id))
metadata_list.append(export_record)
except Exception as e:
self.error(f"Error processing metadata for {cve_id}: {e}")
if not metadata_list:
self.warning("No metadata found to export")
return
# Export in requested format
try:
if format_type == 'json':
with open(output_path, 'w') as f:
json.dump(metadata_list, f, indent=2, default=str)
elif format_type == 'csv':
if metadata_list:
fieldnames = metadata_list[0].keys()
with open(output_path, 'w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(metadata_list)
self.success(f"Exported metadata for {len(metadata_list)} CVEs")
except Exception as e:
self.error(f"Error writing export file: {e}")
def _flatten_metadata(self, metadata: Dict) -> Dict:
"""Flatten nested metadata structure for export"""
flattened = {}
# CVE info fields
cve_info = metadata.get('cve_info', {})
flattened.update({
'cve_id': cve_info.get('cve_id'),
'description': cve_info.get('description'),
'cvss_score': cve_info.get('cvss_score'),
'severity': cve_info.get('severity'),
'published_date': cve_info.get('published_date'),
'modified_date': cve_info.get('modified_date'),
'affected_products_count': len(cve_info.get('affected_products', [])),
'reference_urls_count': len(cve_info.get('reference_urls', []))
})
# PoC data fields
poc_data = metadata.get('poc_data', {})
flattened.update({
'poc_count': poc_data.get('poc_count', 0),
'has_nomi_sec_pocs': bool(poc_data.get('poc_data', {}).get('nomi_sec')),
'has_github_pocs': bool(poc_data.get('poc_data', {}).get('github')),
'has_exploitdb_pocs': bool(poc_data.get('poc_data', {}).get('exploitdb'))
})
# Processing fields
processing = metadata.get('processing', {})
flattened.update({
'data_source': processing.get('data_source'),
'bulk_processed': processing.get('bulk_processed', False),
'reference_sync_status': processing.get('reference_sync_status')
})
# Rule generation fields
rule_generation = metadata.get('rule_generation', {})
generation_methods = list(rule_generation.keys())
flattened.update({
'generation_methods': ','.join(generation_methods),
'generation_methods_count': len(generation_methods),
'has_template_rule': 'template' in generation_methods,
'has_llm_rule': any('llm' in method for method in generation_methods),
'has_hybrid_rule': 'hybrid' in generation_methods
})
# Timestamps
flattened.update({
'created_at': metadata.get('created_at'),
'updated_at': metadata.get('updated_at'),
'migrated_at': metadata.get('migrated_at')
})
return flattened
async def export_ruleset(self, output_file: str, year: Optional[int],
method: Optional[str], include_metadata: bool = True):
"""Export consolidated ruleset file"""
output_path = Path(output_file)
output_path.parent.mkdir(parents=True, exist_ok=True)
self.info(f"Creating consolidated ruleset: {output_path}")
if year:
self.info(f"Including year: {year}")
if method:
self.info(f"Including method: {method}")
# Get CVEs and collect rules
cves = self.get_all_cves(year)
ruleset = {
'metadata': {
'generated_at': self.format_json_output({"timestamp": "now"})[:19] + 'Z',
'filter_year': year,
'filter_method': method,
'total_cves': len(cves),
'generator': 'CVE-SIGMA Auto Generator CLI'
},
'rules': []
}
rule_count = 0
for cve_id in cves:
try:
rules = self.list_cve_rules(cve_id)
if method:
rules = [r for r in rules if method.lower() in r.lower()]
for rule_file in rules:
rule_content = self.load_sigma_rule(cve_id, rule_file)
if not rule_content:
continue
rule_entry = {
'cve_id': cve_id,
'rule_file': rule_file,
'content': rule_content
}
if include_metadata:
metadata = self.load_cve_metadata(cve_id)
if metadata:
rule_entry['cve_metadata'] = {
'severity': metadata.get('cve_info', {}).get('severity'),
'cvss_score': metadata.get('cve_info', {}).get('cvss_score'),
'poc_count': metadata.get('poc_data', {}).get('poc_count', 0)
}
ruleset['rules'].append(rule_entry)
rule_count += 1
except Exception as e:
self.error(f"Error processing {cve_id}: {e}")
# Update metadata with actual counts
ruleset['metadata']['total_rules'] = rule_count
# Save ruleset
try:
with open(output_path, 'w') as f:
json.dump(ruleset, f, indent=2, default=str)
self.success(f"Created consolidated ruleset with {rule_count} rules")
except Exception as e:
self.error(f"Error creating ruleset file: {e}")

116
cli/commands/generate_commands.py Normal file
View file

@ -0,0 +1,116 @@
"""
Generate Commands
Commands for generating SIGMA rules for existing CVEs.
"""
import asyncio
from typing import Dict, List, Optional
from .base_command import BaseCommand
from .process_commands import ProcessCommands
class GenerateCommands(BaseCommand):
"""Commands for generating SIGMA rules"""
def __init__(self, config):
super().__init__(config)
self.process_commands = ProcessCommands(config)
async def generate_cve(self, cve_id: str, method: str, provider: Optional[str], model: Optional[str], force: bool):
"""Generate SIGMA rules for a specific CVE"""
if not self.validate_cve_id(cve_id):
self.error(f"Invalid CVE ID format: {cve_id}")
return
# Check if CVE exists
metadata = self.load_cve_metadata(cve_id)
if not metadata:
self.error(f"CVE {cve_id} not found. Run 'sigma-cli process cve {cve_id}' first to fetch data.")
return
self.info(f"Generating rules for {cve_id} using method: {method}")
if provider:
self.info(f"Using LLM provider: {provider}")
if model:
self.info(f"Using model: {model}")
# Use the process command logic
methods = [method] if method != 'all' else ['template', 'llm', 'hybrid']
success = await self.process_commands._process_single_cve(cve_id, methods, force)
if success:
rules = self.list_cve_rules(cve_id)
self.success(f"Generated {len(rules)} rules for {cve_id}")
for rule in rules:
self.info(f" - {rule}")
else:
self.error(f"Failed to generate rules for {cve_id}")
async def regenerate_rules(self, year: Optional[int], method: str, filter_quality: Optional[str]):
"""Regenerate existing SIGMA rules"""
self.info(f"Regenerating rules with method: {method}")
if year:
self.info(f"Filtering by year: {year}")
if filter_quality:
self.info(f"Filtering by quality: {filter_quality}")
# Get CVEs to regenerate
cves_to_process = []
if year:
cves = self.get_all_cves(year)
else:
cves = self.get_all_cves()
# Filter by quality if specified
for cve_id in cves:
if filter_quality:
metadata = self.load_cve_metadata(cve_id)
if metadata:
poc_data = metadata.get('poc_data', {})
# Simple quality filter based on PoC count
poc_count = poc_data.get('poc_count', 0)
quality_meets_filter = False
if filter_quality == 'excellent' and poc_count >= 5:
quality_meets_filter = True
elif filter_quality == 'good' and poc_count >= 3:
quality_meets_filter = True
elif filter_quality == 'fair' and poc_count >= 1:
quality_meets_filter = True
if quality_meets_filter:
cves_to_process.append(cve_id)
else:
cves_to_process.append(cve_id)
if not cves_to_process:
self.warning("No CVEs found matching the criteria")
return
self.info(f"Will regenerate rules for {len(cves_to_process)} CVEs")
# Regenerate rules
methods = [method] if method != 'all' else ['template', 'llm', 'hybrid']
processed = 0
failed = 0
for cve_id in cves_to_process:
try:
success = await self.process_commands._process_single_cve(cve_id, methods, True) # Force=True
if success:
processed += 1
else:
failed += 1
if (processed + failed) % 10 == 0:
self.info(f"Regenerated {processed + failed}/{len(cves_to_process)} CVEs...")
except Exception as e:
self.error(f"Error regenerating {cve_id}: {e}")
failed += 1
self.success(f"Regeneration completed!")
self.success(f"Processed: {processed}, Failed: {failed}")

379
cli/commands/migrate_commands.py Normal file
View file

@ -0,0 +1,379 @@
"""
Migration Commands
Commands for migrating data from the existing web application database
to the new file-based directory structure.
"""
import asyncio
import json
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Any
import click
# Import the base command class
from .base_command import BaseCommand
# Import database models from the existing backend
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'backend'))
class MigrateCommands(BaseCommand):
"""Commands for migrating from database to file structure"""
async def migrate_from_database(self, database_url: Optional[str], batch_size: int, dry_run: bool):
"""Migrate data from existing database to file structure"""
try:
# Import database components
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from main import CVE, SigmaRule, RuleTemplate # Import from existing main.py
# Use provided database URL or default
if not database_url:
database_url = os.getenv("DATABASE_URL", "postgresql://cve_user:cve_password@localhost:5432/cve_sigma_db")
self.info(f"Connecting to database: {database_url.split('@')[1] if '@' in database_url else database_url}")
# Create database session
engine = create_engine(database_url)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db = SessionLocal()
# Get total counts
cve_count = db.query(CVE).count()
rule_count = db.query(SigmaRule).count()
template_count = db.query(RuleTemplate).count()
self.info(f"Found {cve_count} CVEs, {rule_count} SIGMA rules, {template_count} templates")
if dry_run:
self.warning("DRY RUN MODE - No files will be created")
# Show what would be migrated
sample_cves = db.query(CVE).limit(5).all()
for cve in sample_cves:
cve_dir = self.get_cve_directory(cve.cve_id)
self.info(f"Would create: {cve_dir}")
# Count rules for this CVE
rules = db.query(SigmaRule).filter(SigmaRule.cve_id == cve.cve_id).all()
self.info(f" - Would migrate {len(rules)} SIGMA rules")
return
# Migrate CVEs and rules
migrated_cves = 0
migrated_rules = 0
# Process CVEs in batches
offset = 0
while offset < cve_count:
batch_cves = db.query(CVE).offset(offset).limit(batch_size).all()
for cve in batch_cves:
try:
await self._migrate_cve(db, cve)
migrated_cves += 1
# Migrate associated rules
rules = db.query(SigmaRule).filter(SigmaRule.cve_id == cve.cve_id).all()
for rule in rules:
if await self._migrate_sigma_rule(cve.cve_id, rule):
migrated_rules += 1
if migrated_cves % 10 == 0:
self.info(f"Migrated {migrated_cves}/{cve_count} CVEs...")
except Exception as e:
self.error(f"Error migrating {cve.cve_id}: {e}")
offset += batch_size
# Migrate templates to new location
template_dir = self.config.base_dir / "backend" / "templates"
template_dir.mkdir(exist_ok=True)
templates = db.query(RuleTemplate).all()
for template in templates:
template_file = template_dir / f"{template.template_name.lower().replace(' ', '_')}.yaml"
if not template_file.exists():
try:
with open(template_file, 'w') as f:
f.write(template.template_content)
self.info(f"Migrated template: {template.template_name}")
except Exception as e:
self.error(f"Error migrating template {template.template_name}: {e}")
db.close()
self.success(f"Migration completed!")
self.success(f"Migrated {migrated_cves} CVEs and {migrated_rules} SIGMA rules")
except ImportError as e:
self.error(f"Could not import database models: {e}")
self.error("Make sure you're running from the project root directory")
except Exception as e:
self.error(f"Migration failed: {e}")
import traceback
traceback.print_exc()
async def _migrate_cve(self, db, cve) -> bool:
"""Migrate a single CVE to file structure"""
try:
# Create CVE metadata
metadata = {
"cve_info": {
"cve_id": cve.cve_id,
"description": cve.description,
"cvss_score": float(cve.cvss_score) if cve.cvss_score else None,
"severity": cve.severity,
"published_date": cve.published_date.isoformat() if cve.published_date else None,
"modified_date": cve.modified_date.isoformat() if cve.modified_date else None,
"affected_products": cve.affected_products or [],
"reference_urls": cve.reference_urls or []
},
"poc_data": {
"poc_count": getattr(cve, 'poc_count', 0),
"poc_data": getattr(cve, 'poc_data', {}),
"nomi_sec_data": getattr(cve, 'poc_data', {}).get('nomi_sec', []) if getattr(cve, 'poc_data', {}) else [],
"github_pocs": getattr(cve, 'poc_data', {}).get('github', []) if getattr(cve, 'poc_data', {}) else []
},
"processing": {
"data_source": getattr(cve, 'data_source', 'nvd_api'),
"bulk_processed": getattr(cve, 'bulk_processed', False),
"reference_sync_status": getattr(cve, 'reference_sync_status', 'pending')
},
"file_manifest": [],
"rule_generation": {},
"created_at": cve.created_at.isoformat() if cve.created_at else datetime.utcnow().isoformat(),
"updated_at": datetime.utcnow().isoformat(),
"migrated_at": datetime.utcnow().isoformat()
}
# Save PoC analysis if available
if hasattr(cve, 'poc_data') and cve.poc_data:
cve_dir = self.ensure_cve_directory(cve.cve_id)
poc_analysis_file = cve_dir / "poc_analysis.json"
with open(poc_analysis_file, 'w') as f:
json.dump(cve.poc_data, f, indent=2, default=str)
metadata["file_manifest"].append("poc_analysis.json")
# Save metadata
return self.save_cve_metadata(cve.cve_id, metadata)
except Exception as e:
self.error(f"Error migrating CVE {cve.cve_id}: {e}")
return False
async def _migrate_sigma_rule(self, cve_id: str, rule) -> bool:
"""Migrate a single SIGMA rule to file structure"""
try:
# Determine rule filename based on generation method/source
if hasattr(rule, 'poc_source') and rule.poc_source:
if 'llm' in rule.poc_source.lower() or 'openai' in rule.poc_source.lower():
filename = "rule_llm_openai.sigma"
elif 'anthropic' in rule.poc_source.lower():
filename = "rule_llm_anthropic.sigma"
elif 'hybrid' in rule.poc_source.lower():
filename = "rule_hybrid.sigma"
else:
filename = "rule_template.sigma"
else:
# Default to template-based
filename = "rule_template.sigma"
# Check if we already have a rule with this name, if so append a suffix
existing_rules = self.list_cve_rules(cve_id)
if filename in existing_rules:
base_name = filename.replace('.sigma', '')
counter = 1
while f"{base_name}_{counter}.sigma" in existing_rules:
counter += 1
filename = f"{base_name}_{counter}.sigma"
# Save the rule content
if self.save_sigma_rule(cve_id, filename, rule.rule_content):
# Update metadata with additional rule information
metadata = self.load_cve_metadata(cve_id)
if metadata:
rule_info = {
"rule_name": rule.rule_name,
"detection_type": getattr(rule, 'detection_type', ''),
"log_source": getattr(rule, 'log_source', ''),
"confidence_level": getattr(rule, 'confidence_level', ''),
"auto_generated": getattr(rule, 'auto_generated', True),
"exploit_based": getattr(rule, 'exploit_based', False),
"poc_source": getattr(rule, 'poc_source', 'template'),
"poc_quality_score": getattr(rule, 'poc_quality_score', 0),
"github_repos": getattr(rule, 'github_repos', []),
"created_at": rule.created_at.isoformat() if rule.created_at else None,
"migrated_at": datetime.utcnow().isoformat()
}
method_key = filename.replace('rule_', '').replace('.sigma', '')
if 'rule_generation' not in metadata:
metadata['rule_generation'] = {}
metadata['rule_generation'][method_key] = rule_info
self.save_cve_metadata(cve_id, metadata)
return True
except Exception as e:
self.error(f"Error migrating rule for {cve_id}: {e}")
return False
return False
async def validate_migration(self, year: Optional[int] = None):
"""Validate migrated data integrity"""
self.info("Validating migrated data...")
issues = []
validated_cves = 0
validated_rules = 0
# Get CVEs to validate
cves_to_check = self.get_all_cves(year)
for cve_id in cves_to_check:
try:
# Check if metadata exists and is valid
metadata = self.load_cve_metadata(cve_id)
if not metadata:
issues.append(f"{cve_id}: Missing metadata.json")
continue
# Validate required metadata fields
required_fields = ['cve_info', 'poc_data', 'processing']
for field in required_fields:
if field not in metadata:
issues.append(f"{cve_id}: Missing metadata field '{field}'")
# Validate CVE info
if 'cve_info' in metadata:
cve_info = metadata['cve_info']
if not cve_info.get('cve_id'):
issues.append(f"{cve_id}: Missing cve_id in metadata")
elif cve_info['cve_id'] != cve_id:
issues.append(f"{cve_id}: CVE ID mismatch in metadata")
# Validate file manifest
file_manifest = metadata.get('file_manifest', [])
cve_dir = self.get_cve_directory(cve_id)
for file_name in file_manifest:
file_path = cve_dir / file_name
if not file_path.exists():
issues.append(f"{cve_id}: Referenced file '{file_name}' does not exist")
# Check for SIGMA rule files
rule_files = self.list_cve_rules(cve_id)
for rule_file in rule_files:
rule_content = self.load_sigma_rule(cve_id, rule_file)
if not rule_content:
issues.append(f"{cve_id}: Could not load rule file '{rule_file}'")
elif not rule_content.strip():
issues.append(f"{cve_id}: Empty rule file '{rule_file}'")
else:
# Basic YAML validation for SIGMA rules
if not rule_content.strip().startswith('title:'):
issues.append(f"{cve_id}: Rule '{rule_file}' doesn't appear to be valid SIGMA format")
validated_rules += 1
validated_cves += 1
if validated_cves % 100 == 0:
self.info(f"Validated {validated_cves} CVEs...")
except Exception as e:
issues.append(f"{cve_id}: Validation error - {e}")
# Print validation results
self.info(f"\nValidation completed:")
self.info(f"- Validated {validated_cves} CVEs")
self.info(f"- Validated {validated_rules} SIGMA rules")
if issues:
self.warning(f"Found {len(issues)} validation issues:")
for issue in issues[:20]: # Show first 20 issues
self.error(f" {issue}")
if len(issues) > 20:
self.warning(f" ... and {len(issues) - 20} more issues")
else:
self.success("No validation issues found!")
async def cleanup_migration(self):
"""Clean up migration artifacts and temporary files"""
self.info("Cleaning up migration artifacts...")
# Remove empty directories
for year_dir in self.config.cves_dir.iterdir():
if year_dir.is_dir():
for cve_dir in year_dir.iterdir():
if cve_dir.is_dir():
# Check if directory is empty
if not any(cve_dir.iterdir()):
cve_dir.rmdir()
self.info(f"Removed empty directory: {cve_dir}")
# Check if year directory is now empty
if not any(year_dir.iterdir()):
year_dir.rmdir()
self.info(f"Removed empty year directory: {year_dir}")
self.success("Cleanup completed!")
async def migration_stats(self):
"""Show migration statistics"""
self.info("Migration Statistics:")
years = self.get_years_with_data()
total_cves = 0
total_rules = 0
stats_by_year = {}
for year in years:
cves = self.get_all_cves(year)
year_cves = len(cves)
year_rules = 0
for cve_id in cves:
rules = self.list_cve_rules(cve_id)
year_rules += len(rules)
stats_by_year[year] = {
'cves': year_cves,
'rules': year_rules
}
total_cves += year_cves
total_rules += year_rules
# Print statistics table
headers = ["Year", "CVEs", "Rules", "Avg Rules/CVE"]
rows = []
for year in sorted(years):
stats = stats_by_year[year]
avg_rules = stats['rules'] / stats['cves'] if stats['cves'] > 0 else 0
rows.append([
str(year),
str(stats['cves']),
str(stats['rules']),
f"{avg_rules:.1f}"
])
# Add totals
avg_total = total_rules / total_cves if total_cves > 0 else 0
rows.append(["TOTAL", str(total_cves), str(total_rules), f"{avg_total:.1f}"])
self.print_table(headers, rows, "Migration Statistics by Year")

499
cli/commands/process_commands.py Normal file
View file

@ -0,0 +1,499 @@
"""
Process Commands
Commands for processing CVEs and generating SIGMA rules in the file-based system.
"""
import asyncio
import json
import os
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Any, Tuple
import click
# Import the base command class
from .base_command import BaseCommand
# Import processing components from the existing backend
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'backend'))
class ProcessCommands(BaseCommand):
"""Commands for processing CVEs and generating rules"""
def __init__(self, config):
super().__init__(config)
self._initialize_processors()
def _initialize_processors(self):
"""Initialize the processing components"""
try:
# Import core processing modules
from nvd_bulk_processor import NVDBulkProcessor
from nomi_sec_client import NomiSecClient
from enhanced_sigma_generator import EnhancedSigmaGenerator
from poc_analyzer import PoCAnalyzer
from yaml_metadata_generator import YAMLMetadataGenerator
# Create processors (will be initialized per operation due to session requirements)
self.nvd_processor_class = NVDBulkProcessor
self.nomi_sec_client_class = NomiSecClient
self.sigma_generator_class = EnhancedSigmaGenerator
self.poc_analyzer = PoCAnalyzer()
self.yaml_generator_class = YAMLMetadataGenerator
except ImportError as e:
self.error(f"Could not import processing modules: {e}")
self.error("Make sure you're running from the project root directory")
sys.exit(1)
async def process_year(self, year: int, methods: List[str], force: bool, batch_size: int):
"""Process all CVEs for a specific year"""
self.info(f"Processing CVEs for year {year}")
self.info(f"Methods: {', '.join(methods)}")
self.info(f"Batch size: {batch_size}")
if force:
self.warning("Force mode enabled - will regenerate existing rules")
try:
# First, fetch/update CVE data for the year
await self._fetch_cve_data_for_year(year, batch_size)
# Get all CVEs for the year
cves = self.get_all_cves(year)
if not cves:
self.warning(f"No CVEs found for year {year}")
return
self.info(f"Found {len(cves)} CVEs for {year}")
# Process in batches
processed = 0
failed = 0
for i in range(0, len(cves), batch_size):
batch = cves[i:i+batch_size]
for cve_id in batch:
try:
success = await self._process_single_cve(cve_id, methods, force)
if success:
processed += 1
else:
failed += 1
if (processed + failed) % 10 == 0:
self.info(f"Processed {processed + failed}/{len(cves)} CVEs...")
except Exception as e:
self.error(f"Error processing {cve_id}: {e}")
failed += 1
# Small delay between batches
await asyncio.sleep(1)
self.success(f"Year {year} processing completed!")
self.success(f"Processed: {processed}, Failed: {failed}")
except Exception as e:
self.error(f"Error processing year {year}: {e}")
import traceback
traceback.print_exc()
async def process_cve(self, cve_id: str, methods: List[str], force: bool):
"""Process a specific CVE"""
if not self.validate_cve_id(cve_id):
self.error(f"Invalid CVE ID format: {cve_id}")
return
self.info(f"Processing CVE: {cve_id}")
self.info(f"Methods: {', '.join(methods)}")
try:
# First ensure we have the CVE data
year = int(cve_id.split('-')[1])
await self._fetch_specific_cve_data(cve_id, year)
# Process the CVE
success = await self._process_single_cve(cve_id, methods, force)
if success:
self.success(f"Successfully processed {cve_id}")
else:
self.error(f"Failed to process {cve_id}")
except Exception as e:
self.error(f"Error processing {cve_id}: {e}")
import traceback
traceback.print_exc()
async def process_bulk(self, start_year: int, end_year: int, methods: List[str], batch_size: int):
"""Bulk process CVEs across multiple years"""
self.info(f"Bulk processing CVEs from {start_year} to {end_year}")
self.info(f"Methods: {', '.join(methods)}")
total_processed = 0
total_failed = 0
for year in range(start_year, end_year + 1):
try:
self.info(f"\n--- Processing Year {year} ---")
year_start_processed = total_processed
await self.process_year(year, methods, False, batch_size)
# Update totals (approximate, since process_year doesn't return counts)
cves_in_year = len(self.get_all_cves(year))
total_processed += cves_in_year
except Exception as e:
self.error(f"Error processing year {year}: {e}")
total_failed += 1
self.success(f"\nBulk processing completed!")
self.success(f"Years processed: {end_year - start_year + 1}")
self.success(f"Approximate CVEs processed: {total_processed}")
async def process_incremental(self, days: int, methods: List[str]):
"""Process recently modified CVEs"""
self.info(f"Processing CVEs modified in the last {days} days")
cutoff_date = datetime.utcnow() - timedelta(days=days)
self.info(f"Cutoff date: {cutoff_date.isoformat()}")
# Find CVEs modified since cutoff date
recent_cves = []
for cve_id in self.get_all_cves():
metadata = self.load_cve_metadata(cve_id)
if metadata and 'cve_info' in metadata:
modified_date_str = metadata['cve_info'].get('modified_date')
if modified_date_str:
try:
modified_date = datetime.fromisoformat(modified_date_str.replace('Z', '+00:00'))
if modified_date >= cutoff_date:
recent_cves.append(cve_id)
except (ValueError, TypeError):
pass # Skip if date parsing fails
if not recent_cves:
self.warning("No recently modified CVEs found")
return
self.info(f"Found {len(recent_cves)} recently modified CVEs")
processed = 0
failed = 0
for cve_id in recent_cves:
try:
success = await self._process_single_cve(cve_id, methods, False)
if success:
processed += 1
else:
failed += 1
except Exception as e:
self.error(f"Error processing {cve_id}: {e}")
failed += 1
self.success(f"Incremental processing completed!")
self.success(f"Processed: {processed}, Failed: {failed}")
async def _fetch_cve_data_for_year(self, year: int, batch_size: int):
"""Fetch CVE data for a specific year from NVD"""
self.info(f"Fetching CVE data for year {year}...")
try:
# Use the existing NVD bulk processor
from main import SessionLocal # Import session factory
db_session = SessionLocal()
try:
processor = self.nvd_processor_class(db_session)
# Download and process NVD data for the year
result = await processor.download_and_process_year(year)
if result.get('success'):
self.info(f"Successfully fetched {result.get('processed_cves', 0)} CVEs for {year}")
# Convert database records to file structure
await self._sync_database_to_files(db_session, year)
else:
self.warning(f"Issues fetching CVE data for {year}: {result.get('error', 'Unknown error')}")
finally:
db_session.close()
except Exception as e:
self.error(f"Error fetching CVE data for year {year}: {e}")
async def _fetch_specific_cve_data(self, cve_id: str, year: int):
"""Fetch data for a specific CVE"""
# Check if we already have metadata for this CVE
existing_metadata = self.load_cve_metadata(cve_id)
if existing_metadata:
return # Already have the data
# Fetch from NVD if not already present
self.info(f"Fetching data for {cve_id}...")
try:
from main import SessionLocal
db_session = SessionLocal()
try:
processor = self.nvd_processor_class(db_session)
# Fetch single CVE data
result = await processor.fetch_single_cve(cve_id)
if result:
# Convert to file structure
await self._sync_single_cve_to_files(db_session, cve_id)
self.info(f"Successfully fetched data for {cve_id}")
else:
self.warning(f"Could not fetch data for {cve_id}")
finally:
db_session.close()
except Exception as e:
self.error(f"Error fetching data for {cve_id}: {e}")
async def _sync_database_to_files(self, db_session, year: int):
"""Sync database records to file structure for a specific year"""
try:
from main import CVE
# Get all CVEs for the year from database
year_pattern = f"CVE-{year}-%"
cves = db_session.query(CVE).filter(CVE.cve_id.like(year_pattern)).all()
for cve in cves:
await self._convert_cve_to_file(cve)
except Exception as e:
self.error(f"Error syncing database to files for year {year}: {e}")
async def _sync_single_cve_to_files(self, db_session, cve_id: str):
"""Sync a single CVE from database to file structure"""
try:
from main import CVE
cve = db_session.query(CVE).filter(CVE.cve_id == cve_id).first()
if cve:
await self._convert_cve_to_file(cve)
except Exception as e:
self.error(f"Error syncing {cve_id} to files: {e}")
async def _convert_cve_to_file(self, cve):
"""Convert a database CVE record to file structure"""
try:
# Create metadata structure
metadata = {
"cve_info": {
"cve_id": cve.cve_id,
"description": cve.description,
"cvss_score": float(cve.cvss_score) if cve.cvss_score else None,
"severity": cve.severity,
"published_date": cve.published_date.isoformat() if cve.published_date else None,
"modified_date": cve.modified_date.isoformat() if cve.modified_date else None,
"affected_products": cve.affected_products or [],
"reference_urls": cve.reference_urls or []
},
"poc_data": {
"poc_count": getattr(cve, 'poc_count', 0),
"poc_data": getattr(cve, 'poc_data', {}),
},
"processing": {
"data_source": getattr(cve, 'data_source', 'nvd_api'),
"bulk_processed": getattr(cve, 'bulk_processed', False),
"reference_sync_status": getattr(cve, 'reference_sync_status', 'pending')
},
"file_manifest": [],
"rule_generation": {},
"created_at": cve.created_at.isoformat() if cve.created_at else datetime.utcnow().isoformat(),
"updated_at": datetime.utcnow().isoformat()
}
# Save metadata
self.save_cve_metadata(cve.cve_id, metadata)
except Exception as e:
self.error(f"Error converting CVE {cve.cve_id} to file: {e}")
async def _process_single_cve(self, cve_id: str, methods: List[str], force: bool) -> bool:
"""Process a single CVE with specified methods"""
try:
# Load CVE metadata
metadata = self.load_cve_metadata(cve_id)
if not metadata:
self.error(f"No metadata found for {cve_id}")
return False
# Check if processing is needed
existing_rules = self.list_cve_rules(cve_id)
if existing_rules and not force:
self.info(f"Rules already exist for {cve_id}, skipping (use --force to regenerate)")
return True
success = True
# Process with each requested method
for method in methods:
if method == 'all':
# Generate with all available methods
await self._generate_template_rule(cve_id, metadata)
await self._generate_llm_rule(cve_id, metadata, 'openai')
await self._generate_llm_rule(cve_id, metadata, 'anthropic')
await self._generate_hybrid_rule(cve_id, metadata)
elif method == 'template':
await self._generate_template_rule(cve_id, metadata)
elif method == 'llm':
await self._generate_llm_rule(cve_id, metadata)
elif method == 'hybrid':
await self._generate_hybrid_rule(cve_id, metadata)
return success
except Exception as e:
self.error(f"Error processing {cve_id}: {e}")
return False
async def _generate_template_rule(self, cve_id: str, metadata: Dict) -> bool:
"""Generate template-based SIGMA rule"""
try:
from main import SessionLocal
db_session = SessionLocal()
try:
generator = self.sigma_generator_class(db_session)
# Create mock CVE object from metadata
class MockCVE:
def __init__(self, meta):
cve_info = meta.get('cve_info', {})
self.cve_id = cve_info.get('cve_id')
self.description = cve_info.get('description')
self.severity = cve_info.get('severity')
self.affected_products = cve_info.get('affected_products', [])
self.poc_data = meta.get('poc_data', {}).get('poc_data', {})
mock_cve = MockCVE(metadata)
# Generate rule using template method
rule_content = await generator._generate_template_based_rule(mock_cve, None, None)
if rule_content:
self.save_sigma_rule(cve_id, "rule_template.sigma", rule_content)
self.info(f"Generated template rule for {cve_id}")
return True
else:
self.warning(f"Failed to generate template rule for {cve_id}")
return False
finally:
db_session.close()
except Exception as e:
self.error(f"Error generating template rule for {cve_id}: {e}")
return False
async def _generate_llm_rule(self, cve_id: str, metadata: Dict, provider: str = 'openai') -> bool:
"""Generate LLM-based SIGMA rule"""
try:
from main import SessionLocal
db_session = SessionLocal()
try:
generator = self.sigma_generator_class(db_session, llm_provider=provider)
# Check if LLM is available
if not generator.llm_client.is_available():
self.warning(f"LLM provider {provider} not available for {cve_id}")
return False
# Create mock CVE object
class MockCVE:
def __init__(self, meta):
cve_info = meta.get('cve_info', {})
self.cve_id = cve_info.get('cve_id')
self.description = cve_info.get('description', '')
self.severity = cve_info.get('severity')
self.affected_products = cve_info.get('affected_products', [])
self.poc_data = meta.get('poc_data', {}).get('poc_data', {})
mock_cve = MockCVE(metadata)
# Get PoC data for enhanced generation
poc_data = metadata.get('poc_data', {}).get('poc_data', {})
best_poc = None
poc_content = ""
# Try to find best PoC content
if poc_data and 'nomi_sec' in poc_data:
nomi_pocs = poc_data['nomi_sec']
if nomi_pocs:
best_poc = nomi_pocs[0] # Use first PoC
poc_content = best_poc.get('content', '')
# Generate LLM-enhanced rule
rule_content = await generator.llm_client.generate_sigma_rule(
cve_id=cve_id,
poc_content=poc_content,
cve_description=mock_cve.description
)
if rule_content:
filename = f"rule_llm_{provider}.sigma"
self.save_sigma_rule(cve_id, filename, rule_content)
self.info(f"Generated {provider} LLM rule for {cve_id}")
return True
else:
self.warning(f"Failed to generate {provider} LLM rule for {cve_id}")
return False
finally:
db_session.close()
except Exception as e:
self.error(f"Error generating {provider} LLM rule for {cve_id}: {e}")
return False
async def _generate_hybrid_rule(self, cve_id: str, metadata: Dict) -> bool:
"""Generate hybrid SIGMA rule (template + LLM enhancement)"""
try:
# First generate template-based rule
template_success = await self._generate_template_rule(cve_id, metadata)
if not template_success:
return False
# Then enhance with LLM if available
llm_success = await self._generate_llm_rule(cve_id, metadata, 'openai')
if llm_success:
# Load both rules and create hybrid version
template_rule = self.load_sigma_rule(cve_id, "rule_template.sigma")
llm_rule = self.load_sigma_rule(cve_id, "rule_llm_openai.sigma")
if template_rule and llm_rule:
# Simple hybrid: use LLM rule but keep template metadata structure
# This is a simplified approach - could be made more sophisticated
hybrid_rule = llm_rule # For now, just use the LLM rule as hybrid
self.save_sigma_rule(cve_id, "rule_hybrid.sigma", hybrid_rule)
self.info(f"Generated hybrid rule for {cve_id}")
return True
# If LLM enhancement failed, template rule is still valid
return template_success
except Exception as e:
self.error(f"Error generating hybrid rule for {cve_id}: {e}")
return False

194
cli/commands/search_commands.py Normal file
View file

@ -0,0 +1,194 @@
"""
Search Commands
Commands for searching CVEs and SIGMA rules in the file-based system.
"""
import re
from typing import Dict, List, Optional, Tuple
from .base_command import BaseCommand
class SearchCommands(BaseCommand):
"""Commands for searching CVEs and rules"""
async def search_cves(self, pattern: str, year: Optional[int], severity: Optional[str],
has_poc: bool, has_rules: bool, limit: int):
"""Search for CVEs by pattern"""
self.info(f"Searching CVEs with pattern: '{pattern}'")
if year:
self.info(f"Filtering by year: {year}")
if severity:
self.info(f"Filtering by severity: {severity}")
if has_poc:
self.info("Only showing CVEs with PoC data")
if has_rules:
self.info("Only showing CVEs with generated rules")
# Get CVEs to search
cves_to_search = self.get_all_cves(year)
if not cves_to_search:
self.warning("No CVEs found to search")
return
matches = []
pattern_regex = re.compile(pattern, re.IGNORECASE)
for cve_id in cves_to_search:
try:
metadata = self.load_cve_metadata(cve_id)
if not metadata:
continue
cve_info = metadata.get('cve_info', {})
poc_data = metadata.get('poc_data', {})
# Apply filters
if severity and cve_info.get('severity', '').lower() != severity.lower():
continue
if has_poc and poc_data.get('poc_count', 0) == 0:
continue
if has_rules:
rules = self.list_cve_rules(cve_id)
if not rules:
continue
# Check pattern match
match_found = False
# Search in CVE ID
if pattern_regex.search(cve_id):
match_found = True
# Search in description
description = cve_info.get('description', '')
if description and pattern_regex.search(description):
match_found = True
# Search in affected products
products = cve_info.get('affected_products', [])
for product in products:
if pattern_regex.search(product):
match_found = True
break
if match_found:
rule_count = len(self.list_cve_rules(cve_id))
matches.append({
'cve_id': cve_id,
'severity': cve_info.get('severity', 'Unknown'),
'cvss_score': cve_info.get('cvss_score', 'N/A'),
'poc_count': poc_data.get('poc_count', 0),
'rule_count': rule_count,
'description': (description[:100] + '...') if len(description) > 100 else description
})
if len(matches) >= limit:
break
except Exception as e:
self.error(f"Error searching {cve_id}: {e}")
# Display results
if matches:
headers = ["CVE ID", "Severity", "CVSS", "PoCs", "Rules", "Description"]
rows = []
for match in matches:
rows.append([
match['cve_id'],
match['severity'],
str(match['cvss_score']),
str(match['poc_count']),
str(match['rule_count']),
match['description']
])
self.print_table(headers, rows, f"CVE Search Results ({len(matches)} matches)")
else:
self.warning("No matching CVEs found")
async def search_rules(self, pattern: str, rule_type: Optional[str], method: Optional[str], limit: int):
"""Search for SIGMA rules by pattern"""
self.info(f"Searching SIGMA rules with pattern: '{pattern}'")
if rule_type:
self.info(f"Filtering by rule type: {rule_type}")
if method:
self.info(f"Filtering by generation method: {method}")
matches = []
pattern_regex = re.compile(pattern, re.IGNORECASE)
# Search through all CVEs and their rules
all_cves = self.get_all_cves()
for cve_id in all_cves:
try:
rules = self.list_cve_rules(cve_id)
for rule_file in rules:
# Apply method filter
if method:
rule_method = rule_file.replace('rule_', '').replace('.sigma', '')
if method.lower() not in rule_method.lower():
continue
# Load and search rule content
rule_content = self.load_sigma_rule(cve_id, rule_file)
if not rule_content:
continue
# Apply rule type filter (search in logsource)
if rule_type:
if f'category: {rule_type}' not in rule_content.lower() and \
f'product: {rule_type}' not in rule_content.lower():
continue
# Check pattern match in rule content
if pattern_regex.search(rule_content):
# Extract rule title
title_match = re.search(r'^title:\s*(.+)$', rule_content, re.MULTILINE)
title = title_match.group(1) if title_match else 'Unknown'
# Extract detection type from logsource
logsource_match = re.search(r'category:\s*(\w+)', rule_content)
detection_type = logsource_match.group(1) if logsource_match else 'Unknown'
matches.append({
'cve_id': cve_id,
'rule_file': rule_file,
'title': title,
'detection_type': detection_type,
'method': rule_file.replace('rule_', '').replace('.sigma', '')
})
if len(matches) >= limit:
break
if len(matches) >= limit:
break
except Exception as e:
self.error(f"Error searching rules for {cve_id}: {e}")
# Display results
if matches:
headers = ["CVE ID", "Rule File", "Title", "Type", "Method"]
rows = []
for match in matches:
rows.append([
match['cve_id'],
match['rule_file'],
match['title'][:50] + '...' if len(match['title']) > 50 else match['title'],
match['detection_type'],
match['method']
])
self.print_table(headers, rows, f"SIGMA Rule Search Results ({len(matches)} matches)")
else:
self.warning("No matching rules found")

296
cli/commands/stats_commands.py Normal file
View file

@ -0,0 +1,296 @@
"""
Statistics Commands
Commands for generating statistics and reports about CVEs and SIGMA rules.
"""
import json
from datetime import datetime
from collections import defaultdict, Counter
from typing import Dict, List, Optional
from .base_command import BaseCommand
class StatsCommands(BaseCommand):
"""Commands for generating statistics"""
async def overview(self, year: Optional[int], output: Optional[str]):
"""Generate overview statistics"""
self.info("Generating overview statistics...")
# Collect statistics
stats = self._collect_overview_stats(year)
# Display overview
self._display_overview_stats(stats, year)
# Save to file if requested
if output:
try:
with open(output, 'w') as f:
json.dump(stats, f, indent=2, default=str)
self.success(f"Statistics saved to {output}")
except Exception as e:
self.error(f"Failed to save statistics: {e}")
async def poc_stats(self, year: Optional[int]):
"""Generate PoC coverage statistics"""
self.info("Generating PoC coverage statistics...")
cves = self.get_all_cves(year)
if not cves:
self.warning("No CVEs found")
return
# Collect PoC statistics
total_cves = len(cves)
cves_with_pocs = 0
poc_sources = Counter()
quality_distribution = Counter()
severity_poc_breakdown = defaultdict(lambda: {'total': 0, 'with_poc': 0})
for cve_id in cves:
try:
metadata = self.load_cve_metadata(cve_id)
if not metadata:
continue
cve_info = metadata.get('cve_info', {})
poc_data = metadata.get('poc_data', {})
severity = cve_info.get('severity', 'Unknown')
severity_poc_breakdown[severity]['total'] += 1
poc_count = poc_data.get('poc_count', 0)
if poc_count > 0:
cves_with_pocs += 1
severity_poc_breakdown[severity]['with_poc'] += 1
# Count PoC sources
if 'poc_data' in poc_data:
poc_info = poc_data['poc_data']
if 'nomi_sec' in poc_info and poc_info['nomi_sec']:
poc_sources['nomi_sec'] += len(poc_info['nomi_sec'])
if 'github' in poc_info and poc_info['github']:
poc_sources['github'] += len(poc_info['github'])
if 'exploitdb' in poc_info and poc_info['exploitdb']:
poc_sources['exploitdb'] += len(poc_info['exploitdb'])
# Quality assessment based on PoC count
if poc_count >= 5:
quality_distribution['excellent'] += 1
elif poc_count >= 3:
quality_distribution['good'] += 1
elif poc_count >= 1:
quality_distribution['fair'] += 1
except Exception as e:
self.error(f"Error processing {cve_id}: {e}")
# Display PoC statistics
coverage_percent = (cves_with_pocs / total_cves * 100) if total_cves > 0 else 0
title = f"PoC Coverage Statistics"
if year:
title += f" for {year}"
self.info(f"\n{title}")
self.info("=" * len(title))
self.info(f"Total CVEs: {total_cves}")
self.info(f"CVEs with PoCs: {cves_with_pocs}")
self.info(f"Coverage: {coverage_percent:.1f}%")
if poc_sources:
self.info(f"\nPoC Sources:")
for source, count in poc_sources.most_common():
self.info(f" {source}: {count}")
if quality_distribution:
self.info(f"\nQuality Distribution:")
for quality, count in quality_distribution.most_common():
self.info(f" {quality}: {count}")
# Severity breakdown table
if severity_poc_breakdown:
headers = ["Severity", "Total CVEs", "With PoCs", "Coverage %"]
rows = []
for severity, data in sorted(severity_poc_breakdown.items()):
coverage = (data['with_poc'] / data['total'] * 100) if data['total'] > 0 else 0
rows.append([
severity,
str(data['total']),
str(data['with_poc']),
f"{coverage:.1f}%"
])
self.print_table(headers, rows, "PoC Coverage by Severity")
async def rule_stats(self, year: Optional[int], method: Optional[str]):
"""Generate rule generation statistics"""
self.info("Generating rule generation statistics...")
cves = self.get_all_cves(year)
if not cves:
self.warning("No CVEs found")
return
# Collect rule statistics
total_cves = len(cves)
cves_with_rules = 0
method_counts = Counter()
rules_per_cve = []
for cve_id in cves:
try:
rules = self.list_cve_rules(cve_id)
if method:
# Filter rules by method
rules = [r for r in rules if method.lower() in r.lower()]
if rules:
cves_with_rules += 1
rules_per_cve.append(len(rules))
for rule_file in rules:
rule_method = rule_file.replace('rule_', '').replace('.sigma', '')
method_counts[rule_method] += 1
except Exception as e:
self.error(f"Error processing {cve_id}: {e}")
# Calculate statistics
rule_coverage = (cves_with_rules / total_cves * 100) if total_cves > 0 else 0
avg_rules_per_cve = sum(rules_per_cve) / len(rules_per_cve) if rules_per_cve else 0
total_rules = sum(method_counts.values())
# Display rule statistics
title = f"Rule Generation Statistics"
if year:
title += f" for {year}"
if method:
title += f" (method: {method})"
self.info(f"\n{title}")
self.info("=" * len(title))
self.info(f"Total CVEs: {total_cves}")
self.info(f"CVEs with rules: {cves_with_rules}")
self.info(f"Rule coverage: {rule_coverage:.1f}%")
self.info(f"Total rules: {total_rules}")
self.info(f"Average rules per CVE: {avg_rules_per_cve:.1f}")
if method_counts:
headers = ["Generation Method", "Rule Count", "% of Total"]
rows = []
for gen_method, count in method_counts.most_common():
percentage = (count / total_rules * 100) if total_rules > 0 else 0
rows.append([
gen_method,
str(count),
f"{percentage:.1f}%"
])
self.print_table(headers, rows, "Rules by Generation Method")
def _collect_overview_stats(self, year: Optional[int]) -> Dict:
"""Collect comprehensive overview statistics"""
cves = self.get_all_cves(year)
stats = {
'generated_at': datetime.utcnow().isoformat(),
'filter_year': year,
'total_cves': len(cves),
'severity_breakdown': Counter(),
'yearly_breakdown': Counter(),
'poc_stats': {
'cves_with_pocs': 0,
'total_poc_count': 0
},
'rule_stats': {
'cves_with_rules': 0,
'total_rule_count': 0,
'generation_methods': Counter()
}
}
for cve_id in cves:
try:
metadata = self.load_cve_metadata(cve_id)
if not metadata:
continue
cve_info = metadata.get('cve_info', {})
poc_data = metadata.get('poc_data', {})
# Year breakdown
cve_year = cve_id.split('-')[1]
stats['yearly_breakdown'][cve_year] += 1
# Severity breakdown
severity = cve_info.get('severity', 'Unknown')
stats['severity_breakdown'][severity] += 1
# PoC statistics
poc_count = poc_data.get('poc_count', 0)
if poc_count > 0:
stats['poc_stats']['cves_with_pocs'] += 1
stats['poc_stats']['total_poc_count'] += poc_count
# Rule statistics
rules = self.list_cve_rules(cve_id)
if rules:
stats['rule_stats']['cves_with_rules'] += 1
stats['rule_stats']['total_rule_count'] += len(rules)
for rule_file in rules:
method = rule_file.replace('rule_', '').replace('.sigma', '')
stats['rule_stats']['generation_methods'][method] += 1
except Exception as e:
self.error(f"Error collecting stats for {cve_id}: {e}")
return stats
def _display_overview_stats(self, stats: Dict, year: Optional[int]):
"""Display overview statistics"""
title = f"CVE-SIGMA Overview Statistics"
if year:
title += f" for {year}"
self.info(f"\n{title}")
self.info("=" * len(title))
self.info(f"Generated at: {stats['generated_at']}")
self.info(f"Total CVEs: {stats['total_cves']}")
# PoC coverage
poc_stats = stats['poc_stats']
poc_coverage = (poc_stats['cves_with_pocs'] / stats['total_cves'] * 100) if stats['total_cves'] > 0 else 0
self.info(f"PoC coverage: {poc_coverage:.1f}% ({poc_stats['cves_with_pocs']} CVEs)")
# Rule coverage
rule_stats = stats['rule_stats']
rule_coverage = (rule_stats['cves_with_rules'] / stats['total_cves'] * 100) if stats['total_cves'] > 0 else 0
self.info(f"Rule coverage: {rule_coverage:.1f}% ({rule_stats['cves_with_rules']} CVEs)")
self.info(f"Total rules: {rule_stats['total_rule_count']}")
# Severity breakdown
if stats['severity_breakdown']:
headers = ["Severity", "Count", "Percentage"]
rows = []
for severity, count in stats['severity_breakdown'].most_common():
percentage = (count / stats['total_cves'] * 100) if stats['total_cves'] > 0 else 0
rows.append([severity, str(count), f"{percentage:.1f}%"])
self.print_table(headers, rows, "CVEs by Severity")
# Yearly breakdown (if not filtered by year)
if not year and stats['yearly_breakdown']:
headers = ["Year", "CVE Count"]
rows = []
for cve_year, count in sorted(stats['yearly_breakdown'].items()):
rows.append([cve_year, str(count)])
self.print_table(headers, rows, "CVEs by Year")

16
cli/requirements.txt Normal file
View file

@ -0,0 +1,16 @@
# CLI Requirements for SIGMA CLI Tool
# Core dependencies
click>=8.0.0
pyyaml>=6.0
asyncio-throttle>=1.0.0
# Database support (for migration)
sqlalchemy>=1.4.0
psycopg2-binary>=2.9.0
# Optional: Enhanced formatting
colorama>=0.4.0
tabulate>=0.9.0
# Import existing backend requirements
-r ../backend/requirements.txt

313
cli/sigma_cli.py Executable file
View file

@ -0,0 +1,313 @@
#!/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()