credential_provider

credential_provider supplies the user and password on each AUTH or HELLO, so a rotated password reaches a running service.

A provider implements CredentialProvider, fred's trait re-exported by the crate. Its fetch returns the user and the password, and the client calls it before every login: at startup, on each new connection and after each reconnect. Its refresh_interval makes the client log in again on a schedule, so a short-lived password is replaced before it expires. The provider wins over credentials set on the same broker.

The trait is asynchronous through async-trait, so the service depends on that crate too. It comes with the credential-provider feature. A TestApp test runs no handshake, so the provider is not called there.

Example

The dependency lines:

ruststream-fred = { version = "0.7", features = ["credential-provider"] }
async-trait = "0.1"

The provider reads the password of the user worker from a secret file the platform rotates, and logs in again every ten minutes.

Macros
Manual

examples/redis_broker_credential_provider.rs
/// The password of the ACL user `worker`, read from a secret file the platform rotates.
#[derive(Debug)]
struct RotatedSecret {
    path: PathBuf,
}

#[async_trait]
impl CredentialProvider for RotatedSecret {
    async fn fetch(
        &self,
        _server: Option<&Server>,
    ) -> Result<(Option<String>, Option<String>), FredError> {
        let path = self.path.clone();
        // A file read blocks, so it runs off the connection's task.
        let read = task::spawn_blocking(move || fs::read_to_string(path))
            .await
            .map_err(|err| FredError::new(ErrorKind::Auth, err.to_string()))?;
        let password = read.map_err(|err| FredError::new(ErrorKind::Auth, err.to_string()))?;
        Ok((Some("worker".to_owned()), Some(password.trim().to_owned())))
    }

    /// Logs in again every ten minutes, with whatever the file holds by then.
    fn refresh_interval(&self) -> Option<Duration> {
        Some(Duration::from_mins(10))
    }
}

fn broker(url: &str, secret: PathBuf) -> RedisBroker {
    RedisBroker::standalone(url).credential_provider(Arc::new(RotatedSecret { path: secret }))
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(broker(
        "redis://localhost:6379",
        PathBuf::from("/run/secrets/redis-password"),
    ))
}

examples/manual/redis_broker_credential_provider.rs
/// The password of the ACL user `worker`, read from a secret file the platform rotates.
#[derive(Debug)]
struct RotatedSecret {
    path: PathBuf,
}

#[async_trait]
impl CredentialProvider for RotatedSecret {
    async fn fetch(
        &self,
        _server: Option<&Server>,
    ) -> Result<(Option<String>, Option<String>), FredError> {
        let path = self.path.clone();
        // A file read blocks, so it runs off the connection's task.
        let read = task::spawn_blocking(move || fs::read_to_string(path))
            .await
            .map_err(|err| FredError::new(ErrorKind::Auth, err.to_string()))?;
        let password = read.map_err(|err| FredError::new(ErrorKind::Auth, err.to_string()))?;
        Ok((Some("worker".to_owned()), Some(password.trim().to_owned())))
    }

    /// Logs in again every ten minutes, with whatever the file holds by then.
    fn refresh_interval(&self) -> Option<Duration> {
        Some(Duration::from_mins(10))
    }
}

fn broker(url: &str, secret: PathBuf) -> RedisBroker {
    RedisBroker::standalone(url).credential_provider(Arc::new(RotatedSecret { path: secret }))
}

fn app() -> impl App<State = ()> {
    service(broker(
        "redis://localhost:6379",
        PathBuf::from("/run/secrets/redis-password"),
    ))
}

Example​

Example