Как гарантировать чтобы при работе с RabbitMQ в него не клал значение другой микросервис?

Контроль доступа к RabbitMQ между микросервисами

Это критически важный вопрос безопасности в микросервисной архитектуре. Нужно гарантировать, что только авторизованный микросервис может публиковать сообщения в очередь.

1. Authentication и Authorization

Уровень RabbitMQ: Каждый микросервис должен иметь свои учетные данные.

// service-a.js
const amqp = require('amqplib');

const connection = await amqp.connect({
  protocol: 'amqp',
  hostname: 'rabbitmq.prod.local',
  port: 5672,
  username: 'service-a',
  password: 'secure-password-a', // разные для каждого сервиса
  vhost: '/app'
});

Управление пользователями RabbitMQ:

# Создание пользователя
sudo rabbitmqctl add_user service-a secure-password-a

# Установка permissions
sudo rabbitmqctl set_permissions -p /app service-a \
  '^(orders|events).*' \
  '^(orders|events).*' \
  '^(orders|events).*'

# Для service-b
sudo rabbitmqctl add_user service-b secure-password-b
sudo rabbitmqctl set_permissions -p /app service-b \
  '^(payments|notifications).*' \
  '^(payments|notifications).*' \
  '^(payments|notifications).*'

Значение permissions:

• Первый regex — configure (создание очередей)
• Второй regex — write (публикация)
• Третий regex — read (потребление)

2. VirtualHost изоляция

# Создание отдельных VirtualHost для групп сервисов
sudo rabbitmqctl add_vhost /orders
sudo rabbitmqctl add_vhost /payments

# service-a работает только с /orders
const connection = await amqp.connect({
  vhost: '/orders',
  username: 'service-a',
  password: 'password-a'
});

# service-b работает только с /payments
const connection = await amqp.connect({
  vhost: '/payments',
  username: 'service-b',
  password: 'password-b'
});

3. Очередь на основе паттернов обмена

// Сервис, который только публикует (producer)
class OrderPublisher {
  constructor(channel) {
    this.channel = channel;
  }

  async publishOrderCreated(orderId) {
    const exchange = 'orders.events';
    
    // Declare exchange (только для написи)
    await this.channel.assertExchange(exchange, 'topic', { durable: true });
    
    // Публикация
    const message = { orderId, timestamp: Date.now() };
    this.channel.publish(
      exchange,
      'order.created',
      Buffer.from(JSON.stringify(message))
    );
  }
}

// Сервис, который только подписывается (consumer)
class PaymentConsumer {
  constructor(channel) {
    this.channel = channel;
  }

  async setupSubscription() {
    const exchange = 'orders.events';
    const queue = 'payment.orders.queue';
    
    // Declare exchange (только для чтения)
    await this.channel.assertExchange(exchange, 'topic', { durable: true });
    
    // Создание очереди
    await this.channel.assertQueue(queue, { durable: true });
    
    // Binding
    await this.channel.bindQueue(queue, exchange, 'order.created');
    
    // Потребление
    await this.channel.consume(queue, (msg) => {
      const data = JSON.parse(msg.content.toString());
      this.handleOrderCreated(data);
      this.channel.ack(msg);
    });
  }
}

4. Policy-based контроль

# RabbitMQ Policy для enforce на durable очередях
sudo rabbitmqctl set_policy -p /app high-availability \
  '^(orders|events)' \
  '{"ha-mode":"all"}'

# Политика для автоудаления неиспользуемых очередей
sudo rabbitmqctl set_policy -p /app auto-delete \
  '^temp\..*' \
  '{"expires":3600000}'

5. Сертификаты и TLS

// Безопасное подключение с TLS
const fs = require('fs');

const connection = await amqp.connect({
  protocol: 'amqps',
  hostname: 'rabbitmq.prod.local',
  port: 5671,
  username: 'service-a',
  password: 'password',
  ca: [fs.readFileSync('/path/to/ca.crt')],
  cert: fs.readFileSync('/path/to/client.crt'),
  key: fs.readFileSync('/path/to/client.key'),
  heartbeat: 60
});

6. Метаданные сообщений (Publisher confirmation)

// Убедиться, что сообщение дошло до очереди
const channel = connection.createChannel();
await channel.confirmSelect();

const published = await new Promise((resolve, reject) => {
  const ok = channel.publish(
    'orders.events',
    'order.created',
    Buffer.from(JSON.stringify(message)),
    { persistent: true, correlationId: uuid() },
    (err, ok) => {
      if (err) reject(err);
      else resolve(ok);
    }
  );
  
  if (!ok) reject(new Error('Channel publish failed'));
});

7. Aудит и логирование

// Логирование всех публикаций
class AuditedPublisher {
  constructor(channel, logger) {
    this.channel = channel;
    this.logger = logger;
  }

  async publish(exchange, routingKey, message) {
    const metadata = {
      timestamp: new Date().toISOString(),
      source: process.env.SERVICE_NAME,
      version: '1.0',
      correlationId: uuid()
    };
    
    this.logger.info('Publishing message', {
      exchange,
      routingKey,
      messageSize: JSON.stringify(message).length,
      ...metadata
    });
    
    this.channel.publish(
      exchange,
      routingKey,
      Buffer.from(JSON.stringify(message)),
      { ...metadata, persistent: true }
    );
  }
}

8. Network policies (Kubernetes)

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: rabbitmq-access
spec:
  podSelector:
    matchLabels:
      app: rabbitmq
  ingress:
  - from:
    - podSelector:
        matchLabels:
          access: rabbitmq
    ports:
    - protocol: TCP
      port: 5672

9. Практический пример

// service-a (микросервис заказов)
const setupOrderService = async () => {
  const conn = await amqp.connect({
    url: process.env.RABBITMQ_URL,
    username: 'order-service',
    password: process.env.ORDER_SERVICE_PASSWORD
  });
  
  const channel = await conn.createChannel();
  await channel.assertExchange('orders', 'topic', { durable: true });
  
  // Может ТОЛЬКО писать
  channel.publish('orders', 'order.created', Buffer.from('{}'));
};

// service-b (микросервис платежей)
const setupPaymentService = async () => {
  const conn = await amqp.connect({
    url: process.env.RABBITMQ_URL,
    username: 'payment-service',
    password: process.env.PAYMENT_SERVICE_PASSWORD
  });
  
  const channel = await conn.createChannel();
  const queue = await channel.assertQueue('payment.orders', { durable: true });
  
  // Может ТОЛЬКО читать
  await channel.bindQueue('payment.orders', 'orders', 'order.*');
  await channel.consume('payment.orders', (msg) => {
    // обработка
  });
};

Ключевые принципы

1. Least Privilege — каждый сервис получает минимально необходимые права
2. Separation of Concerns — producer и consumer имеют разные credentials
3. Audit Trail — логирование всех операций
4. Encryption in Transit — TLS для соединений
5. Rotation — регулярная смена паролей
6. Monitoring — алерты на подозрительную активность

Этот подход гарантирует, что только авторизованные сервисы могут взаимодействовать с RabbitMQ, и каждое действие отслеживается.