FEATURE: Add support for updatable authentication secrets

[cressie176]

Some authentication protocols expire authentication tokens unless they are refreshed. RabbitMQ supports this via the update-secret operation. amqplib was recently updated to support this, so it would be nice to add something similar to Rascal, along the following lines

broker.updateSecret(vhost, secret, reason, cb)

behind the scenes the broker would have to find the correct vhost and call vhost.updateSecret(secret, reason, cb). In turn the vhost would have to call connection.updateSecret(secret, reason, cb)

We need to make the secret semi-permanent, so that if a connection is dropped, Rascal will use the latest secret to reconnect.

Finally we need to consider what to do if the connection had already been dropped. Ideally we would abort (or immediately repeat) any in progress reconnection, using the new token.

Here's how...

• Change tasks/createConnection to prefer a secret from the vhost configuration to the connection password
• Add broker.updateSecret(vhost, secret, reason, cb). This looks up the vhost and calls vhost.updateSecret as below
• Add vhost.updateSecret(secret, reason, cb).
  1. Update the vhost config with the new secret.
  2. If there is a connection, calls connection.updateSecret(secret, reason, cb)
  3. If there is not a connection, registers a listener for the "connect" event, and calls connection.updateSecret(secret, reason, cb) once connected.

[hugowschneider]

Hi @cressie176,

I have a question to this feature. I see in documentation updateSecret but I cannot see the methods in the amqplib.d.ts. If I dig into the javacript code, I can see the implementation and tests using the Connection to do that. What am I missing?

I am using amqplib version 0.10.5

Many thanks for the help! :)

npm info amqplib

[email protected] | MIT | deps: 3 | versions: 33
An AMQP 0-9-1 (e.g., RabbitMQ) library and client.
http://amqp-node.github.io/amqplib/

keywords: AMQP, AMQP 0-9-1, RabbitMQ

dist
.tarball: https://registry.npmjs.org/amqplib/-/amqplib-0.10.5.tgz
.shasum: 2b217a3c56f2935ee2e36206bfd7e25ee866491f
.integrity: sha512-Dx5zmy0Ur+Q7LPPdhz+jx5IzmJBoHd15tOeAfQ8SuvEtyPJ20hBemhOBA4b1WeORCRa0ENM/kHCzmem1w/zHvQ==
.unpackedSize: 448.5 kB

dependencies:
@acuminous/bitsyntax: ^0.1.2 buffer-more-ints: ~1.0.0     url-parse: ~1.5.10           

maintainers:
- squaremo <[email protected]>
- cressie176 <[email protected]>

dist-tags:
latest: 0.10.5  

published 2 weeks ago by cressie176 <[email protected]>

/// amqplib.d.ts

/// <reference types="node" />

import * as events from "events";
import { ConsumeMessage, GetMessage, Message, Options, Replies, ServerProperties } from "./properties";
export * from "./properties";

export interface Connection extends events.EventEmitter {
    close(): Promise<void>;
    createChannel(): Promise<Channel>;
    createConfirmChannel(): Promise<ConfirmChannel>;
    connection: {
        serverProperties: ServerProperties;
    };
}

export interface Channel extends events.EventEmitter {
    connection: Connection;

    close(): Promise<void>;

    assertQueue(queue: string, options?: Options.AssertQueue): Promise<Replies.AssertQueue>;
    checkQueue(queue: string): Promise<Replies.AssertQueue>;

    deleteQueue(queue: string, options?: Options.DeleteQueue): Promise<Replies.DeleteQueue>;
    purgeQueue(queue: string): Promise<Replies.PurgeQueue>;

    bindQueue(queue: string, source: string, pattern: string, args?: any): Promise<Replies.Empty>;
    unbindQueue(queue: string, source: string, pattern: string, args?: any): Promise<Replies.Empty>;

    assertExchange(
        exchange: string,
        type: "direct" | "topic" | "headers" | "fanout" | "match" | string,
        options?: Options.AssertExchange,
    ): Promise<Replies.AssertExchange>;
    checkExchange(exchange: string): Promise<Replies.Empty>;

    deleteExchange(exchange: string, options?: Options.DeleteExchange): Promise<Replies.Empty>;

    bindExchange(destination: string, source: string, pattern: string, args?: any): Promise<Replies.Empty>;
    unbindExchange(destination: string, source: string, pattern: string, args?: any): Promise<Replies.Empty>;

    publish(exchange: string, routingKey: string, content: Buffer, options?: Options.Publish): boolean;
    sendToQueue(queue: string, content: Buffer, options?: Options.Publish): boolean;

    consume(
        queue: string,
        onMessage: (msg: ConsumeMessage | null) => void,
        options?: Options.Consume,
    ): Promise<Replies.Consume>;

    cancel(consumerTag: string): Promise<Replies.Empty>;
    get(queue: string, options?: Options.Get): Promise<GetMessage | false>;

    ack(message: Message, allUpTo?: boolean): void;
    ackAll(): void;

    nack(message: Message, allUpTo?: boolean, requeue?: boolean): void;
    nackAll(requeue?: boolean): void;
    reject(message: Message, requeue?: boolean): void;

    prefetch(count: number, global?: boolean): Promise<Replies.Empty>;
    recover(): Promise<Replies.Empty>;
}

export interface ConfirmChannel extends Channel {
    publish(
        exchange: string,
        routingKey: string,
        content: Buffer,
        options?: Options.Publish,
        callback?: (err: any, ok: Replies.Empty) => void,
    ): boolean;
    sendToQueue(
        queue: string,
        content: Buffer,
        options?: Options.Publish,
        callback?: (err: any, ok: Replies.Empty) => void,
    ): boolean;

    waitForConfirms(): Promise<void>;
}

export const credentials: {
    amqplain(username: string, password: string): {
        mechanism: string;
        response(): Buffer;
        username: string;
        password: string;
    };
    external(): {
        mechanism: string;
        response(): Buffer;
    };
    plain(username: string, password: string): {
        mechanism: string;
        response(): Buffer;
        username: string;
        password: string;
    };
};

export function connect(url: string | Options.Connect, socketOptions?: any): Promise<Connection>;

[cressie176]

Hello @hugowschneider,
It's going to take a while for me to implement this feature in Racal - it's a surprisingly hard change because rascal is configuration based.

Re amqplib
The TypeScript definitions are maintained separately so it is possible they haven't been updated. I'm sure they'd appreciate a PR.

[hugowschneider]

Hello @cressie176 ,

First thanks for the amazingly fast reply. I just realized now I ended up in the Rascal repo. Sorry about that. I am interested in the amqplib.

Sorry for the confusion and I just found the PR (6 days old) DefinitelyTyped/DefinitelyTyped#71226