scripts/adapters/b3_mindgard_adapter.py

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

Mindgard: Evaded Prompt Injection and Jailbreak Samples
License: CC-BY-NC-4.0 (DO NOT REDISTRIBUTE)
HuggingFace: https://huggingface.co/datasets/Mindgard/evaded-prompt-injection-and-jailbreak-samples

IMPORTANT: This dataset is CC-BY-NC licensed. Do not redistribute for commercial use.

Maps Mindgard evaded prompts to the B3 Direct Query Safety schema.
"""

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

try:
    from datasets import load_dataset
    HAS_DATASETS = True
except ImportError:
    HAS_DATASETS = False

from .base_adapter import BaseAdapter, DatasetMetadata

logger = logging.getLogger(__name__)


class MindgardAdapter(BaseAdapter):
    """
    Adapter for Mindgard dataset.

    Mindgard provides evaded prompt injection and jailbreak samples
    that have successfully bypassed guardrails.

    WARNING: This is an Extended dataset (CC-BY-NC).
    Do not redistribute for commercial purposes.
    """

    ATTACK_NAME_MAPPING = {
        'leetspeak': 'Leetspeak Encoding',
        'rot13': 'ROT13 Encoding',
        'base64': 'Base64 Encoding',
        'unicode': 'Unicode Substitution',
        'homoglyph': 'Homoglyph Attack',
        'payload_split': 'Payload Splitting',
        'instruction_override': 'Instruction Override',
        'context_switch': 'Context Switching',
        'role_play': 'Role Play Attack'
    }

    def __init__(
        self,
        output_dir: Path,
        split: str = "train",
        verbose: bool = False
    ):
        """
        Initialize Mindgard adapter.

        Args:
            output_dir: Directory for output files
            split: Dataset split to use
            verbose: Enable verbose logging
        """
        super().__init__(output_dir, verbose)
        self.split = split

        if not HAS_DATASETS:
            raise ImportError(
                "The 'datasets' library is required. Install with: pip install datasets"
            )

        logger.warning(
            "Mindgard is CC-BY-NC licensed. "
            "Do not redistribute for commercial purposes."
        )

    @property
    def metadata(self) -> DatasetMetadata:
        return DatasetMetadata(
            name="Mindgard",
            version="1.0",
            license="CC-BY-NC-4.0",
            license_url="https://huggingface.co/datasets/Mindgard/evaded-prompt-injection-and-jailbreak-samples",
            source_url="https://huggingface.co/datasets/Mindgard/evaded-prompt-injection-and-jailbreak-samples",
            paper_url=None,
            can_redistribute=False,  # CC-BY-NC restricts commercial use
            target_surface="B3",
            description="Evaded prompt injection and jailbreak samples (research only)"
        )

    def load_source_data(self) -> Generator[Dict[str, Any], None, None]:
        """Load Mindgard data from HuggingFace."""
        logger.info(f"Loading Mindgard dataset (split: {self.split}) from HuggingFace...")
        logger.warning("Remember: CC-BY-NC license - non-commercial use only!")

        try:
            dataset = load_dataset(
                "Mindgard/evaded-prompt-injection-and-jailbreak-samples",
                split=self.split
            )

            for item in dataset:
                yield dict(item)

        except Exception as e:
            if "403" in str(e) or "gated" in str(e).lower():
                logger.error(
                    "Access denied to Mindgard dataset. This is a gated repository. "
                    "Please visit https://huggingface.co/datasets/Mindgard/evaded-prompt-injection-and-jailbreak-samples "
                    "to request access, then login with `huggingface-cli login`."
                )
            logger.error(f"Failed to load Mindgard: {e}")
            raise

    def transform_record(self, record: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        """Transform Mindgard record to B3 schema."""
        # Field names may vary
        jailbreak_prompt = record.get('jailbreak_prompt', record.get('prompt', ''))
        attack_name = record.get('attack_name', record.get('technique', 'unknown'))
        original_prompt = record.get('original_prompt', record.get('original', ''))

        if not jailbreak_prompt:
            logger.warning("Empty jailbreak prompt in record")
            return None

        jailbreak_prompt = self.sanitize_text(jailbreak_prompt, max_length=15000)

        # Determine obfuscation type from attack name
        attack_lower = attack_name.lower() if attack_name else ''
        obfuscation = None
        encoding = None

        if 'leet' in attack_lower:
            obfuscation = "leetspeak"
        elif 'rot13' in attack_lower:
            encoding = "rot13 encoding"
        elif 'base64' in attack_lower:
            encoding = "base64 encode payload"
        elif 'unicode' in attack_lower or 'homoglyph' in attack_lower:
            obfuscation = "use homoglyphs"

        attack_category = self.ATTACK_NAME_MAPPING.get(
            attack_lower.replace(' ', '_'),
            'Obfuscation/Evasion'
        )

        record_id = self.generate_id("B3-Mindgard", jailbreak_prompt)

        return {
            "id": record_id,
            "attack_category": "Guardrail Evasion",
            "attack_subtype": f"Mindgard_{attack_name.replace(' ', '_') if attack_name else 'Unknown'}",
            "adversarial_goal": "Guardrail Bypass / Evasion",
            "user_query": jailbreak_prompt,
            "target_llm_output": "",
            "judge_assessment": {
                "assessment": "Success",  # These are already evaded samples
                "justification": f"Mindgard evaded sample - Attack: {attack_name}",
                "confidence": "High"
            },
            "benign_content": original_prompt if original_prompt else None,
            "mode": None,
            "role_play": None,
            "manipulation_strategy": None,
            "communication_style": None,
            "query_structure": "obfuscated prompt",
            "intent_Opacity": "masked",
            "topic_domain": None,
            "temporal_context": None,
            "language": "English",
            "Obfuscation": obfuscation,
            "adversarial_suffix": None,
            "payload_splitting": "high" if 'split' in attack_lower else None,
            "encoding": encoding,
            "target_rag_component": "Generator_LLM",
            "target_component_model": "External_Evaluation",
            "target_model_system_prompt": "",
            "all_llm_details": {},
            "agent_architecture_description": "Mindgard evasion testing",
            "attack_gen_pipeline_description": f"Mindgard evaded samples - Technique: {attack_name}",
            "generation_timestamp": self.get_timestamp(),
            "script_version": self.BENCHMARK_VERSION,
            "source_dataset": "Mindgard",
            "source_license": "CC-BY-NC-4.0 (Research Only)",
            "research_only": True,
            "attack_technique": attack_name,
            "original_prompt": original_prompt
        }


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

    parser = argparse.ArgumentParser(description="Mindgard to B3 Schema Adapter (Research Only)")
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path("data/external_augmented/extended"),
        help="Output directory"
    )
    parser.add_argument(
        "--split",
        type=str,
        default="train",
        help="Dataset split to use"
    )
    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()

    print("\n" + "="*60)
    print("WARNING: Mindgard is CC-BY-NC licensed.")
    print("Do not redistribute for commercial purposes.")
    print("="*60 + "\n")

    adapter = MindgardAdapter(
        output_dir=args.output_dir,
        split=args.split,
        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}")
    print("\nREMINDER: CC-BY-NC - non-commercial use only.")


if __name__ == "__main__":
    main()