scripts/adapters/b3_harmbench_adapter.py

"""
HarmBench Dataset Adapter for B3 (Direct Query) Schema

HarmBench: A Standardized Evaluation Framework for Automated Red Teaming
License: MIT
Source: https://github.com/centerforaisafety/HarmBench
Paper: https://arxiv.org/abs/2402.04249

Maps HarmBench behaviors to the B3 Direct Query Safety schema.
"""

import csv
import logging
from pathlib import Path
from typing import Dict, Any, Optional, Generator

from .base_adapter import BaseAdapter, DatasetMetadata

logger = logging.getLogger(__name__)


class HarmBenchAdapter(BaseAdapter):
    """
    Adapter for HarmBench dataset.

    HarmBench provides 400 carefully curated harmful behaviors in
    `harmbench_behaviors_text_all.csv` for evaluating LLM robustness.
    """

    CATEGORY_MAPPING = {
        'standard': 'General Harm',
        'copyright': 'Copyright Violation',
        'contextual': 'Contextual Harm',
        'multimodal': 'Multimodal Attack'
    }

    FUNCTIONAL_CATEGORY_MAPPING = {
        'chemical_biological': 'Chemical/Biological Harm',
        'cybercrime': 'Cybercrime',
        'harassment_bullying': 'Harassment/Bullying',
        'illegal_activities': 'Illegal Activities',
        'misinformation_disinformation': 'Misinformation',
        'violence': 'Violence',
        'adult_content': 'Adult Content',
        'self_harm': 'Self Harm',
        'other': 'Other Harm'
    }

    def __init__(
        self,
        output_dir: Path,
        csv_path: Path = None,
        verbose: bool = False
    ):
        """
        Initialize HarmBench adapter.

        Args:
            output_dir: Directory for output files
            csv_path: Path to harmbench_behaviors_text_all.csv
            verbose: Enable verbose logging
        """
        super().__init__(output_dir, verbose)
        self.csv_path = csv_path

    @property
    def metadata(self) -> DatasetMetadata:
        return DatasetMetadata(
            name="HarmBench",
            version="1.0",
            license="MIT",
            license_url="https://github.com/centerforaisafety/HarmBench/blob/main/LICENSE",
            source_url="https://github.com/centerforaisafety/HarmBench",
            paper_url="https://arxiv.org/abs/2402.04249",
            can_redistribute=True,
            target_surface="B3",
            description="400 curated harmful behaviors for red teaming evaluation"
        )

    def load_source_data(self) -> Generator[Dict[str, Any], None, None]:
        """Load HarmBench CSV data."""
        if not self.csv_path:
            raise ValueError(
                "CSV path not provided. Download harmbench_behaviors_text_all.csv from "
                "https://github.com/centerforaisafety/HarmBench/tree/main/data/behavior_datasets"
            )

        csv_path = Path(self.csv_path)
        if not csv_path.exists():
            raise FileNotFoundError(f"HarmBench CSV not found: {csv_path}")

        logger.info(f"Loading HarmBench data from {csv_path}")

        with open(csv_path, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            for row in reader:
                yield row

    def transform_record(self, record: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        """Transform HarmBench record to B3 schema."""
        behavior = record.get('Behavior', record.get('behavior', ''))
        if not behavior:
            logger.warning("Empty behavior field in record")
            return None

        # Extract category information
        functional_cat = record.get('FunctionalCategory', record.get('functional_category', 'other'))
        semantic_cat = record.get('SemanticCategory', record.get('semantic_category', ''))
        behavior_id = record.get('BehaviorID', record.get('behavior_id', ''))

        # Map to our attack category taxonomy
        attack_category = self.FUNCTIONAL_CATEGORY_MAPPING.get(
            functional_cat.lower().replace(' ', '_'),
            'General Harm'
        )

        # Generate deterministic ID
        record_id = self.generate_id("B3-HarmBench", behavior)

        return {
            "id": record_id,
            "attack_category": attack_category,
            "attack_subtype": f"HarmBench_{functional_cat}",
            "adversarial_goal": behavior,
            "user_query": behavior,  # In HarmBench, behavior IS the query
            "target_llm_output": "",  # Not provided in source
            "judge_assessment": {
                "assessment": "Pending",
                "justification": "Imported from HarmBench - requires evaluation",
                "confidence": "N/A"
            },
            "benign_content": None,
            "mode": None,
            "role_play": None,
            "manipulation_strategy": None,
            "communication_style": None,
            "query_structure": "direct question",
            "intent_Opacity": "explicit",
            "topic_domain": semantic_cat if semantic_cat else None,
            "temporal_context": None,
            "language": "English",
            "Obfuscation": None,
            "adversarial_suffix": None,
            "payload_splitting": None,
            "encoding": None,
            "target_rag_component": "Generator_LLM",
            "target_component_model": "External_Evaluation",
            "target_model_system_prompt": "",
            "all_llm_details": {},
            "agent_architecture_description": "HarmBench standardized evaluation",
            "attack_gen_pipeline_description": "HarmBench curated behaviors",
            "generation_timestamp": self.get_timestamp(),
            "script_version": self.BENCHMARK_VERSION,
            "source_dataset": "HarmBench",
            "source_id": behavior_id if behavior_id else None,
            "source_license": "MIT"
        }


def main():
    """CLI entry point for HarmBench adapter."""
    import argparse

    parser = argparse.ArgumentParser(description="HarmBench to B3 Schema Adapter")
    parser.add_argument(
        "--csv-path",
        type=Path,
        required=True,
        help="Path to harmbench_behaviors_text_all.csv"
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path("data/external_augmented/core"),
        help="Output directory"
    )
    parser.add_argument(
        "--max-records",
        type=int,
        default=None,
        help="Maximum records to process"
    )
    parser.add_argument(
        "-v", "--verbose",
        action="store_true",
        help="Enable verbose output"
    )

    args = parser.parse_args()

    adapter = HarmBenchAdapter(
        output_dir=args.output_dir,
        csv_path=args.csv_path,
        verbose=args.verbose
    )

    result = adapter.run(max_records=args.max_records)

    print(f"\nAdapter Results:")
    print(f"  Success: {result.success}")
    print(f"  Records processed: {result.records_processed}")
    print(f"  Records failed: {result.records_failed}")
    if result.output_file:
        print(f"  Output file: {result.output_file}")
    if result.errors:
        print(f"  Errors: {result.errors}")
    if result.warnings:
        print(f"  Warnings: {result.warnings}")


if __name__ == "__main__":
    main()