Counterfit Walkthrough

# Clone the Counterfit repository
git clone https://github.com/Azure/counterfit.git
cd counterfit
 
# Create a virtual environment
python3 -m venv counterfit-env
source counterfit-env/bin/activate
 
# Install Counterfit and dependencies
pip install -e .
 
# Launch the Counterfit CLI
counterfit

# targets/my_classifier/my_classifier.py
"""
Custom Counterfit target for a sentiment classifier API.
"""
import numpy as np
import requests
from counterfit.core.targets import CFTarget
 
 
class MyClassifier(CFTarget):
    target_name = "my_classifier"
    target_data_type = "text"
    target_endpoint = "https://api.example.com/classify"
    target_input_shape = (1,)
    target_output_classes = ["negative", "neutral", "positive"]
    target_classifier = "closed-box"
 
    X = [
        "This product is excellent, highly recommended.",
        "The service was terrible, never going back.",
        "It was an okay experience, nothing special.",
    ]
 
    def load(self):
        """Initialize any required state."""
        self.api_key = self.target_config.get("api_key", "")
 
    def predict(self, x):
        """
        Send input to the model and return class probabilities.
        Counterfit expects a numpy array of shape
        (n_samples, n_classes).
        """
        results = []
        for text in x:
            response = requests.post(
                self.target_endpoint,
                json={"text": str(text)},
                headers={"Authorization": f"Bearer {self.api_key}"},
            )
            probs = response.json()["probabilities"]
            results.append(probs)
        return np.array(results)

counterfit> interact my_classifier
my_classifier>

my_classifier> list attacks
┌─────────────────────────┬───────────┬────────────────┐
│ Attack Name             │ Data Type │ Framework      │
├─────────────────────────┼───────────┼────────────────┤
│ hop_skip_jump           │ image     │ art            │
│ deepfool                │ image     │ art            │
│ textfooler              │ text      │ textattack     │
│ bae                     │ text      │ textattack     │
│ pwws                    │ text      │ textattack     │
│ textbugger              │ text      │ textattack     │
│ input_filter            │ text      │ textattack     │
│ a2t                     │ text      │ textattack     │
│ ... (many more)         │           │                │
└─────────────────────────┴───────────┴────────────────┘

# Select a text attack
my_classifier> use textfooler
 
# Configure the attack
my_classifier>textfooler> set --sample_index 0
my_classifier>textfooler> show options
 
# Run the attack
my_classifier>textfooler> run
 
# View results
my_classifier>textfooler> show results

# scripts/robustness_assessment.py
"""
Automated robustness assessment using Counterfit.
"""
import json
from counterfit.core import Counterfit
 
 
def run_assessment(target_name: str, attacks: list,
                   sample_indices: list) -> dict:
    """
    Run a systematic robustness assessment.
    """
    cf = Counterfit()
    target = cf.load_target(target_name)
 
    results = {
        "target": target_name,
        "total_samples": len(sample_indices),
        "attacks": {},
    }
 
    for attack_name in attacks:
        attack_results = {
            "success_count": 0,
            "total_attempts": 0,
            "avg_perturbation": 0.0,
            "samples": [],
        }
 
        for idx in sample_indices:
            try:
                attack = cf.load_attack(attack_name)
                attack.set_params(sample_index=idx)
                result = attack.run()
 
                attack_results["total_attempts"] += 1
                if result.success:
                    attack_results["success_count"] += 1
                    attack_results["samples"].append({
                        "index": idx,
                        "original": result.original_input,
                        "adversarial": result.adversarial_input,
                        "original_pred": result.original_prediction,
                        "adversarial_pred": result.adversarial_prediction,
                    })
 
            except Exception as e:
                print(f"Attack {attack_name} failed on sample "
                      f"{idx}: {e}")
 
        if attack_results["total_attempts"] > 0:
            attack_results["success_rate"] = (
                attack_results["success_count"] /
                attack_results["total_attempts"]
            )
        results["attacks"][attack_name] = attack_results
 
    return results
 
 
def generate_report(results: dict) -> str:
    """Generate a human-readable robustness report."""
    lines = [
        f"# Robustness Assessment: {results['target']}",
        f"Total samples tested: {results['total_samples']}",
        "",
        "## Attack Results",
        "",
    ]
 
    for attack, data in results["attacks"].items():
        success_rate = data.get("success_rate", 0)
        risk_level = ("CRITICAL" if success_rate > 0.5
                      else "HIGH" if success_rate > 0.2
                      else "MEDIUM" if success_rate > 0.05
                      else "LOW")
 
        lines.append(f"### {attack}")
        lines.append(f"- Success rate: {success_rate:.1%}")
        lines.append(f"- Risk level: {risk_level}")
        lines.append(f"- Successful attacks: "
                      f"{data['success_count']}/{data['total_attempts']}")
        lines.append("")
 
    return "\n".join(lines)

# .github/workflows/adversarial-testing.yml
name: Adversarial Robustness Test
 
on:
  pull_request:
    paths:
      - 'models/**'
      - 'training/**'
 
jobs:
  robustness-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
 
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.10'
 
      - name: Install Counterfit
        run: |
          pip install counterfit
          pip install -r requirements-test.txt
 
      - name: Run robustness assessment
        run: |
          python scripts/robustness_assessment.py \
            --target my_classifier \
            --attacks textfooler,bae,pwws \
            --samples 50 \
            --output results.json
 
      - name: Check robustness thresholds
        run: |
          python scripts/check_thresholds.py \
            --results results.json \
            --max-success-rate 0.1

# scripts/check_thresholds.py
"""
Check adversarial robustness results against thresholds.
Exit with non-zero status if thresholds are exceeded.
"""
import json
import sys
import argparse
 
 
def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--results", required=True)
    parser.add_argument("--max-success-rate", type=float, default=0.1)
    args = parser.parse_args()
 
    with open(args.results) as f:
        results = json.load(f)
 
    failures = []
    for attack, data in results["attacks"].items():
        rate = data.get("success_rate", 0)
        if rate > args.max_success_rate:
            failures.append(
                f"{attack}: {rate:.1%} success rate "
                f"(threshold: {args.max_success_rate:.1%})"
            )
 
    if failures:
        print("ROBUSTNESS CHECK FAILED:")
        for f in failures:
            print(f"  - {f}")
        sys.exit(1)
    else:
        print("All robustness checks passed.")
        sys.exit(0)
 
 
if __name__ == "__main__":
    main()