Add Authentication (#7)

.travis.yml

@@ -2,10 +2,14 @@ language: rust
 
 sudo: required
 
+services:
+    - docker
+
 before_install:
-    - wget https://dl.influxdata.com/influxdb/releases/influxdb_1.7.6_amd64.deb
-    - sudo dpkg -i influxdb_1.7.6_amd64.deb
-    - sudo influxd > $HOME/influx.log 2>&1 &
+    - docker pull influxdb
+    - docker run -d -p 8086:8086 --name influxdb influxdb
+    - docker run -d -p 9086:8086 --name authed_influxdb -e INFLUXDB_HTTP_AUTH_ENABLED=true -e INFLUXDB_ADMIN_USER=admin -e INFLUXDB_ADMIN_PASSWORD=password -e INFLUXDB_USER=nopriv_user -e INFLUXDB_USER_PASSWORD=password influxdb
+    - docker ps
 
 branches:
     only:

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "influxdb"
-version = "0.0.3"
+version = "0.0.4"
 authors = ["Gero Gerke <[email protected]>"]
 edition = "2018"
 description = "InfluxDB Driver for Rust"
@@ -20,7 +20,6 @@ futures = "0.1.27"
 tokio = "0.1.20"
 itertools = "0.8"
 failure = "0.1.5"
-url = "1.7.2"
 serde = { version = "1.0.92", optional = true }
 serde_json = { version = "1.0", optional = true }
 

src/client/mod.rs

@@ -17,6 +17,7 @@
 
 use futures::{Future, Stream};
 use reqwest::r#async::{Client, Decoder};
+use reqwest::{StatusCode, Url};
 
 use std::mem;
 
@@ -25,15 +26,45 @@ use crate::query::read_query::InfluxDbReadQuery;
 use crate::query::write_query::InfluxDbWriteQuery;
 use crate::query::InfluxDbQuery;
 
-use url::form_urlencoded;
-
 use std::any::Any;
 
+#[derive(Clone, Debug)]
+/// Internal Authentication representation
+pub(crate) struct InfluxDbAuthentication {
+    pub username: String,
+    pub password: String,
+}
+
+#[derive(Clone, Debug)]
 /// Internal Representation of a Client
 pub struct InfluxDbClient {
     url: String,
     database: String,
-    // auth: Option<InfluxDbAuthentication>
+    auth: Option<InfluxDbAuthentication>,
+}
+
+impl Into<Vec<(String, String)>> for InfluxDbClient {
+    fn into(self) -> Vec<(String, String)> {
+        let mut vec: Vec<(String, String)> = Vec::new();
+        vec.push(("db".to_string(), self.database));
+        if let Some(auth) = self.auth {
+            vec.push(("u".to_string(), auth.username));
+            vec.push(("p".to_string(), auth.password));
+        }
+        vec
+    }
+}
+
+impl<'a> Into<Vec<(String, String)>> for &'a InfluxDbClient {
+    fn into(self) -> Vec<(String, String)> {
+        let mut vec: Vec<(String, String)> = Vec::new();
+        vec.push(("db".to_string(), self.database.to_owned()));
+        if let Some(auth) = &self.auth {
+            vec.push(("u".to_string(), auth.username.to_owned()));
+            vec.push(("p".to_string(), auth.password.to_owned()));
+        }
+        vec
+    }
 }
 
 impl InfluxDbClient {
@@ -59,9 +90,36 @@ impl InfluxDbClient {
         InfluxDbClient {
             url: url.to_string(),
             database: database.to_string(),
+            auth: None,
         }
     }
 
+    /// Add authentication/authorization information to [`InfluxDbClient`](crate::client::InfluxDbClient)
+    ///
+    /// # Arguments
+    ///
+    /// * username: The Username for InfluxDB.
+    /// * password: THe Password for the user.
+    ///
+    /// # Examples
+    ///
+    /// ```rust
+    /// use influxdb::client::InfluxDbClient;
+    ///
+    /// let _client = InfluxDbClient::new("http://localhost:9086", "test").with_auth("admin", "password");
+    /// ```
+    pub fn with_auth<'a, S1, S2>(mut self, username: S1, password: S2) -> Self
+    where
+        S1: ToString,
+        S2: ToString,
+    {
+        self.auth = Some(InfluxDbAuthentication {
+            username: username.to_string(),
+            password: password.to_string(),
+        });
+        self
+    }
+
     /// Returns the name of the database the client is using
     pub fn database_name(&self) -> &str {
         &self.database
@@ -100,7 +158,7 @@ impl InfluxDbClient {
             })
     }
 
-    /// Sends a [`InfluxDbReadQuery`](crate::query::read_query::InfluxDbReadQuery) or [`InfluxDbWriteQuery`](crate::query::write_query::InfluxDbWriteQuery) to the InfluxDB Server.InfluxDbError
+    /// Sends a [`InfluxDbReadQuery`](crate::query::read_query::InfluxDbReadQuery) or [`InfluxDbWriteQuery`](crate::query::write_query::InfluxDbWriteQuery) to the InfluxDB Server.
     ///
     /// A version capable of parsing the returned string is available under the [serde_integration](crate::integrations::serde_integration)
     ///
@@ -120,6 +178,12 @@ impl InfluxDbClient {
     ///         .add_field("temperature", 82)
     /// );
     /// ```
+    /// # Errors
+    ///
+    /// If the function can not finish the query,
+    /// a [`InfluxDbError`] variant will be returned.
+    ///
+    /// [`InfluxDbError`]: enum.InfluxDbError.html
     pub fn query<Q>(&self, q: &Q) -> Box<dyn Future<Item = String, Error = InfluxDbError>>
     where
         Q: Any + InfluxDbQuery,
@@ -137,48 +201,71 @@ impl InfluxDbClient {
         };
 
         let any_value = q as &dyn Any;
+        let basic_parameters: Vec<(String, String)> = self.into();
 
         let client = if let Some(_) = any_value.downcast_ref::<InfluxDbReadQuery>() {
             let read_query = query.get();
-            let encoded: String = form_urlencoded::Serializer::new(String::new())
-                .append_pair("db", self.database_name())
-                .append_pair("q", &read_query)
-                .finish();
-            let http_query_string = format!(
-                "{url}/query?{encoded}",
-                url = self.database_url(),
-                encoded = encoded
-            );
+
+            let mut url = match Url::parse_with_params(
+                format!("{url}/query", url = self.database_url()).as_str(),
+                basic_parameters,
+            ) {
+                Ok(url) => url,
+                Err(err) => {
+                    let error = InfluxDbError::UrlConstructionError {
+                        error: format!("{}", err),
+                    };
+                    return Box::new(future::err::<String, InfluxDbError>(error));
+                }
+            };
+            url.query_pairs_mut().append_pair("q", &read_query.clone());
+
             if read_query.contains("SELECT") || read_query.contains("SHOW") {
-                Client::new().get(http_query_string.as_str())
+                Client::new().get(url)
             } else {
-                Client::new().post(http_query_string.as_str())
+                Client::new().post(url)
             }
         } else if let Some(write_query) = any_value.downcast_ref::<InfluxDbWriteQuery>() {
-            Client::new()
-                .post(
-                    format!(
-                        "{url}/write?db={db}{precision_str}",
-                        url = self.database_url(),
-                        db = self.database_name(),
-                        precision_str = write_query.get_precision_modifier()
-                    )
-                    .as_str(),
-                )
-                .body(query.get())
+            let mut url = match Url::parse_with_params(
+                format!("{url}/write", url = self.database_url()).as_str(),
+                basic_parameters,
+            ) {
+                Ok(url) => url,
+                Err(err) => {
+                    let error = InfluxDbError::InvalidQueryError {
+                        error: format!("{}", err),
+                    };
+                    return Box::new(future::err::<String, InfluxDbError>(error));
+                }
+            };
+            url.query_pairs_mut()
+                .append_pair("precision", &write_query.get_precision());
+            Client::new().post(url).body(query.get())
         } else {
             unreachable!()
         };
-
         Box::new(
             client
                 .send()
+                .map_err(|err| InfluxDbError::ConnectionError { error: err })
+                .and_then(
+                    |res| -> future::FutureResult<reqwest::r#async::Response, InfluxDbError> {
+                        match res.status() {
+                            StatusCode::UNAUTHORIZED => {
+                                futures::future::err(InfluxDbError::AuthorizationError)
+                            }
+                            StatusCode::FORBIDDEN => {
+                                futures::future::err(InfluxDbError::AuthenticationError)
+                            }
+                            _ => futures::future::ok(res),
+                        }
+                    },
+                )
                 .and_then(|mut res| {
                     let body = mem::replace(res.body_mut(), Decoder::empty());
-                    body.concat2()
-                })
-                .map_err(|err| InfluxDbError::ProtocolError {
-                    error: format!("{}", err),
+                    body.concat2().map_err(|err| InfluxDbError::ProtocolError {
+                        error: format!("{}", err),
+                    })
                 })
                 .and_then(|body| {
                     if let Ok(utf8) = std::str::from_utf8(&body) {
@@ -201,3 +288,70 @@ impl InfluxDbClient {
         )
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::client::InfluxDbClient;
+
+    #[test]
+    fn test_fn_database() {
+        let client = InfluxDbClient::new("http://localhost:8068", "database");
+        assert_eq!("database", client.database_name());
+    }
+
+    #[test]
+    fn test_with_auth() {
+        let client = InfluxDbClient::new("http://localhost:8068", "database");
+        assert_eq!(client.url, "http://localhost:8068");
+        assert_eq!(client.database, "database");
+        assert!(client.auth.is_none());
+        let with_auth = client.with_auth("username", "password");
+        assert!(with_auth.auth.is_some());
+        let auth = with_auth.auth.unwrap();
+        assert_eq!(&auth.username, "username");
+        assert_eq!(&auth.password, "password");
+    }
+
+    #[test]
+    fn test_into_impl() {
+        let client = InfluxDbClient::new("http://localhost:8068", "database");
+        assert!(client.auth.is_none());
+        let basic_parameters: Vec<(String, String)> = client.into();
+        assert_eq!(
+            vec![("db".to_string(), "database".to_string())],
+            basic_parameters
+        );
+
+        let with_auth = InfluxDbClient::new("http://localhost:8068", "database")
+            .with_auth("username", "password");
+        let basic_parameters_with_auth: Vec<(String, String)> = with_auth.into();
+        assert_eq!(
+            vec![
+                ("db".to_string(), "database".to_string()),
+                ("u".to_string(), "username".to_string()),
+                ("p".to_string(), "password".to_string())
+            ],
+            basic_parameters_with_auth
+        );
+
+        let client = InfluxDbClient::new("http://localhost:8068", "database");
+        assert!(client.auth.is_none());
+        let basic_parameters: Vec<(String, String)> = (&client).into();
+        assert_eq!(
+            vec![("db".to_string(), "database".to_string())],
+            basic_parameters
+        );
+
+        let with_auth = InfluxDbClient::new("http://localhost:8068", "database")
+            .with_auth("username", "password");
+        let basic_parameters_with_auth: Vec<(String, String)> = (&with_auth).into();
+        assert_eq!(
+            vec![
+                ("db".to_string(), "database".to_string()),
+                ("u".to_string(), "username".to_string()),
+                ("p".to_string(), "password".to_string())
+            ],
+            basic_parameters_with_auth
+        );
+    }
+}

src/error.rs

@@ -1,11 +1,16 @@
 //! Errors that might happen in the crate
+use reqwest;
 
 #[derive(Debug, Fail)]
 pub enum InfluxDbError {
     #[fail(display = "query is invalid: {}", error)]
     /// Error happens when a query is invalid
     InvalidQueryError { error: String },
 
+    #[fail(display = "Failed to build URL: {}", error)]
+    /// Error happens when a query is invalid
+    UrlConstructionError { error: String },
+
     #[fail(display = "http protocol error: {}", error)]
     /// Error happens when a query is invalid
     ProtocolError { error: String },
@@ -17,4 +22,19 @@ pub enum InfluxDbError {
     #[fail(display = "InfluxDB encountered the following error: {}", error)]
     /// Error which has happened inside InfluxDB
     DatabaseError { error: String },
+
+    #[fail(display = "authentication error. No or incorrect credentials")]
+    /// Error happens when no or incorrect credentials are used. `HTTP 401 Unauthorized`
+    AuthenticationError,
+
+    #[fail(display = "authorization error. User not authorized")]
+    /// Error happens when the supplied user is not authorized. `HTTP 403 Forbidden`
+    AuthorizationError,
+
+    #[fail(display = "connection error: {}", error)]
+    /// Error happens when reqwest fails
+    ConnectionError {
+        #[fail(cause)]
+        error: reqwest::Error,
+    },
 }