Red Team a Financial AI System End-to-End with DiscoveR

from mirror_sdk import MirrorSDK
from mirror_sdk.core.mirror_api_models import RedTeamApplicationRequest

sdk = MirrorSDK()

# Register the trading desk AI assistant as a scan target
app_request = RedTeamApplicationRequest(
    name="Trading Desk AI Assistant",
    description="Internal RAG-based research assistant for equity analysts",
    type="api",
    base_url="https://trading-ai.internal.yourbank.com",
    model_type="openai",
    endpoints=[{
        "path": "/v1/query",
        "method": "POST",
        "status": "active"
    }],
    securityControls={
        "headers": {
            "Authorization": "Bearer your-internal-api-key",
            "Content-Type": "application/json"
        }
    },
    # Domain hint tells DiscoveR to tailor attack prompts for financial AI
    domainHint="finance",
    purpose="Internal equity research assistant with access to trade notes and analyst commentary",
    consent=True
)

application = sdk.redteam.create_application(app_request)
print(f"Application registered: {application.id}")
print(f"Status: {application.status}")

# Wait for validation before scanning
sdk.redteam.wait_for_application_validation(application.id)
print("Application validated. Ready to scan.")

from mirror_sdk import MirrorSDK
from mirror_sdk.core.mirror_api_models import RedTeamScanRequest, ScanStatus
import time
import json

sdk = MirrorSDK()

# Comprehensive scan for quarterly security review
scan_request = RedTeamScanRequest(
    application_id="your-app-id",
    name="Q2 2026 Financial AI Security Assessment",
    security_categories=[
        "jailbreakAndInjection",
        "extractionAttacks",
        "trainingDataPrivacy",
        "ragSecurity",
        "biasAndSafety"
    ],
    # 100 prompts: good coverage for quarterly review and DORA evidence
    max_depth=100,
    max_prompts_per_attack=20,
    max_duration=90  # 90 minute cap
)

scan = sdk.redteam.create_discover_scan(scan_request)
print(f"Scan started: {scan.id}")
print("This will take approximately 60-90 minutes for max_depth=100")

# Poll for completion
while True:
    status = sdk.redteam.get_scan_status(scan.id)
    current = status.get("status")
    print(f"Status: {current}")

    if current in ["completed", "failed", "cancelled"]:
        break
    time.sleep(30)

# Retrieve results
results = sdk.redteam.get_scan_results(scan.id)
vulnerabilities = results.get("vulnerabilities", [])

print(f"\nScan complete.")
print(f"Vulnerabilities found: {len(vulnerabilities)}")
print(f"Risk score: {results.get('riskScore', 'N/A')}")

# Save results for compliance evidence
with open(f"scan_{scan.id}_results.json", "w") as f:
    json.dump(results, f, indent=2)
print(f"Results saved to scan_{scan.id}_results.json")

from mirror_sdk import MirrorSDK
from mirror_sdk.core.mirror_api_models import RedTeamScanRequest, CompareScansRequest
import time

sdk = MirrorSDK()
original_scan_id = "your-original-scan-id"

# Run the same scan again after remediation
# DiscoveR links this scan to the original via correlation_id
original_results = sdk.redteam.get_scan_results(original_scan_id)

rerun_scan = sdk.redteam.create_scan_from_results(
    application_id="your-app-id",
    scan_results=original_results,
    name="Q2 2026 Post-Remediation Verification",
    # Rerun only the tests that previously failed
    filter_failed_only=True
)

print(f"Rerun scan started: {rerun_scan['data']['scan_id']}")

# Wait for completion
rerun_id = rerun_scan['data']['scan_id']
while True:
    status = sdk.redteam.get_scan_status(rerun_id)
    if status.get("status") in ["completed", "failed"]:
        break
    time.sleep(30)

# Compare original and rerun
comparison = sdk.redteam.compare_scans(
    CompareScansRequest(
        scan_ids=[original_scan_id, rerun_id],
        include_details=True
    )
)

print("Comparison summary:")
print(f"Original risk score:   {comparison.get('original_risk_score')}")
print(f"Post-remediation score: {comparison.get('current_risk_score')}")
print(f"Vulnerabilities resolved: {comparison.get('resolved_count')}")
print(f"New vulnerabilities: {comparison.get('new_count')}")

from mirror_sdk import MirrorSDK
from mirror_sdk.core.mirror_api_models import RedTeamScanRequest
import time
import sys

sdk = MirrorSDK()

def ci_security_gate(app_id: str) -> bool:
    """
    Run a quick security gate scan.
    Returns True if the system passes (no critical findings).
    Returns False if the deployment should be blocked.
    """

    scan_request = RedTeamScanRequest(
        application_id=app_id,
        name=f"CI Gate Scan",
        security_categories=["quickScan"],
        max_depth=15  # fast: 2-5 minutes
    )

    scan = sdk.redteam.create_discover_scan(scan_request)

    while True:
        status = sdk.redteam.get_scan_status(scan.id)
        if status.get("status") in ["completed", "failed"]:
            break
        time.sleep(15)

    results = sdk.redteam.get_scan_results(scan.id)
    vulnerabilities = results.get("vulnerabilities", [])

    # Block deployment on any critical finding
    critical = [v for v in vulnerabilities if v.get("severity") == "critical"]

    if critical:
        print(f"DEPLOYMENT BLOCKED: {len(critical)} critical vulnerabilities found")
        for v in critical:
            print(f"  - {v.get('attack_mode')}: {v.get('description')}")
        return False

    print(f"Security gate passed. {len(vulnerabilities)} non-critical findings logged.")
    return True


if __name__ == "__main__":
    passed = ci_security_gate("your-app-id")
    sys.exit(0 if passed else 1)