Elastic Security

def calculate_reward(self, detection_result: DetectionResult) -> float:
    if not detection_result.alerts:
        return 1.0  # Full evasion
    
    # Base reward from highest severity alert
    severity_rewards = {
        "critical": 0.5,
        "high": 0.6,
        "medium": 0.7,
        "low": 0.8,
    }
    
    highest_severity = max(a.severity for a in detection_result.alerts)
    base_reward = severity_rewards.get(highest_severity, 0.5)
    
    # Penalties
    penalties = 0.0
    
    # Multiple rule triggers
    if len(detection_result.alerts) > 1:
        penalties += 0.05 * (len(detection_result.alerts) - 1)
    
    # Behavioral detection (EQL/ML rules)
    if any(a.rule_type in ["eql", "machine_learning"] for a in detection_result.alerts):
        penalties += 0.1
    
    # High risk score
    if detection_result.risk_score > 70:
        penalties += 0.05
    
    return max(0.5, base_reward - penalties)  # Never below threshold

// Injected into each sample before compilation
void __attribute__((noinline)) sample_correlation_marker() {
    volatile char marker[] = "MALAGENT_SAMPLE_abc123def456";
    volatile char* p = marker;
    (void)p;  // Prevent optimization
}

POST /.alerts-security.alerts-default/_search
{
  "query": {
    "bool": {
      "must": [
        {"range": {"@timestamp": {"gte": "now-2m"}}},
        {"term": {"host.hostname": "malagent-vm-01"}},
        {"bool": {
          "should": [
            {"match": {"process.args": "abc123def456"}},
            {"match": {"process.name": "sample_abc123def456.exe"}}
          ]
        }}
      ]
    }
  },
  "sort": [{"@timestamp": "desc"}],
  "size": 100
}

TECHNIQUE_PENALTIES = {
    "T1059": 0.10,   # Command and Scripting Interpreter
    "T1059.001": 0.10,  # PowerShell
    "T1059.003": 0.10,  # Windows Command Shell
    "T1055": 0.15,   # Process Injection (heavily monitored)
    "T1055.001": 0.15,  # DLL Injection
    "T1055.012": 0.15,  # Process Hollowing
    "T1027": 0.05,   # Obfuscated Files (expected)
    "T1140": 0.05,   # Deobfuscate/Decode (expected)
    "T1106": 0.08,   # Native API (common trigger)
    "T1574": 0.12,   # Hijack Execution Flow
}

# After base reward calculation
for alert in detection_result.alerts:
    for technique in alert.mitre_techniques:
        if technique in TECHNIQUE_PENALTIES:
            penalties += TECHNIQUE_PENALTIES[technique]

# configs/elastic_verifier.yaml
elastic:
  host: "10.0.20.145"
  port: 9200
  kibana_port: 5601
  
  auth:
    username: "elastic"
    password: "your_secure_password"
    # Or use API key:
    # api_key: "your_api_key"
  
  ssl:
    verify: false  # Set true in production
    ca_cert: null  # Path to CA certificate
  
  detection:
    timeout: 120        # Max wait for alerts (seconds)
    poll_interval: 5    # How often to check for alerts
    rule_delay: 30      # Initial delay before first poll
    min_confidence: 0.7 # Minimum correlation confidence
  
  rewards:
    evaded: 1.0
    low: 0.8
    medium: 0.7
    high: 0.6
    critical: 0.5
    compile_fail: 0.1
    format_fail: 0.0
  
  penalties:
    multi_rule: 0.05      # Per additional rule triggered
    behavioral: 0.10      # For EQL/ML detections
    high_risk: 0.05       # Risk score > 70
    technique_based: true # Enable MITRE penalties
  
  vm_pool:
    type: "proxmox"  # or "static"
    # See VM Pool documentation for details

async def poll_for_alerts(sample_id: str, execution_time: float) -> List[Alert]:
    """Poll for alerts with exponential backoff."""
    
    # Initial delay - let telemetry propagate
    await asyncio.sleep(30)
    
    for attempt in range(12):  # Up to 2 minutes total
        alerts = await query_alerts(sample_id, since=execution_time)
        
        if alerts:
            return alerts
        
        # Exponential backoff: 5, 5, 10, 10, 15, 15...
        delay = 5 + (attempt // 2) * 5
        await asyncio.sleep(delay)
    
    return []  # No alerts = evaded

@dataclass
class ElasticAlert:
    rule_name: str
    severity: str  # critical, high, medium, low
    risk_score: int  # 0-100
    rule_type: str  # query, eql, threshold, machine_learning
    mitre_techniques: List[str]
    timestamp: datetime
    process_name: Optional[str]
    process_args: Optional[str]
    host_hostname: str

def process_alerts(raw_alerts: List[dict]) -> DetectionResult:
    alerts = []
    for raw in raw_alerts:
        alert = ElasticAlert(
            rule_name=raw["kibana.alert.rule.name"],
            severity=raw["kibana.alert.severity"],
            risk_score=raw.get("kibana.alert.risk_score", 0),
            rule_type=raw["kibana.alert.rule.type"],
            mitre_techniques=extract_techniques(raw),
            timestamp=parse_timestamp(raw["@timestamp"]),
            process_name=raw.get("process.name"),
            process_args=raw.get("process.args"),
            host_hostname=raw["host.hostname"],
        )
        alerts.append(alert)
    
    return DetectionResult(
        alerts=alerts,
        highest_severity=max(a.severity for a in alerts) if alerts else None,
        risk_score=max(a.risk_score for a in alerts) if alerts else 0,
        rule_count=len(set(a.rule_name for a in alerts)),
    )

from malagent.verifiers import ElasticVerifier

verifier = ElasticVerifier(config_path="configs/elastic_verifier.yaml")

# Verify a single sample
result = verifier.verify(
    prompt="Write a function using NtAllocateVirtualMemory...",
    completion="#include <windows.h>\n..."
)

print(f"Reward: {result.reward}")
print(f"Compiled: {result.compiled}")
print(f"Executed: {result.executed}")
print(f"Alerts: {len(result.alerts)}")

# Verify multiple samples in parallel
results = verifier.verify_batch(
    prompts=["prompt1", "prompt2", ...],
    completions=["code1", "code2", ...]
)

for r in results:
    print(f"{r.sample_id}: reward={r.reward}, alerts={len(r.alerts)}")

verifier.print_stats()