Proxy pass for existing users.

Ask console to check per-cluster auth info.

Cargo.lock

@@ -1406,6 +1406,7 @@ dependencies = [
  "hex",
  "md5",
  "rand",
+ "reqwest",
  "rustls",
  "serde",
  "serde_json",

proxy/Cargo.toml

@@ -18,5 +18,6 @@ tokio = "1.11"
 tokio-postgres = { git = "https://github.com/zenithdb/rust-postgres.git", rev="9eb0dbfbeb6a6c1b79099b9f7ae4a8c021877858" }
 clap = "2.33.0"
 rustls = "0.19.1"
+reqwest = { version = "0.11", features = ["blocking", "json"] }
 
 zenith_utils = { path = "../zenith_utils" }

proxy/src/cplane_api.rs

@@ -1,15 +1,14 @@
 use anyhow::{bail, Result};
+use hex;
+use reqwest;
 use serde::{Deserialize, Serialize};
-use std::{
-    collections::HashMap,
-    net::{IpAddr, SocketAddr},
-};
+use std::net::{IpAddr, SocketAddr};
 
 pub struct CPlaneApi {
-    // address: SocketAddr,
+    auth_endpoint: &'static str,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Debug)]
 pub struct DatabaseInfo {
     pub host: IpAddr, // TODO: allow host name here too
     pub port: u16,
@@ -31,62 +30,33 @@ impl DatabaseInfo {
     }
 }
 
-// mock cplane api
 impl CPlaneApi {
-    pub fn new(_address: &SocketAddr) -> CPlaneApi {
-        CPlaneApi {
-            // address: address.clone(),
-        }
+    pub fn new(auth_endpoint: &'static str) -> CPlaneApi {
+        CPlaneApi { auth_endpoint }
     }
 
-    pub fn check_auth(&self, user: &str, md5_response: &[u8], salt: &[u8; 4]) -> Result<()> {
-        // passwords for both is "mypass"
-        let auth_map: HashMap<_, &str> = vec![
-            ("stas@zenith", "716ee6e1c4a9364d66285452c47402b1"),
-            ("stas2@zenith", "3996f75df64c16a8bfaf01301b61d582"),
-        ]
-        .into_iter()
-        .collect();
+    pub fn authenticate_proxy_request(
+        &self,
+        user: &str,
+        md5_response: &[u8],
+        salt: &[u8; 4],
+    ) -> Result<DatabaseInfo> {
+        let mut url = reqwest::Url::parse(self.auth_endpoint)?;
+        url.query_pairs_mut()
+            .append_pair("login", user)
+            .append_pair("md5response", std::str::from_utf8(md5_response)?)
+            .append_pair("salt", &hex::encode(salt));
 
-        let stored_hash = auth_map
-            .get(&user)
-            .ok_or_else(|| anyhow::Error::msg("user not found"))?;
-        let salted_stored_hash = format!(
-            "md5{:x}",
-            md5::compute([stored_hash.as_bytes(), salt].concat())
-        );
+        println!("cplane request: {}", url.as_str());
 
-        let received_hash = std::str::from_utf8(md5_response)?;
+        let resp = reqwest::blocking::get(url)?;
 
-        println!(
-            "auth: {} rh={} sh={} ssh={} {:?}",
-            user, received_hash, stored_hash, salted_stored_hash, salt
-        );
-
-        if received_hash == salted_stored_hash {
-            Ok(())
+        if resp.status().is_success() {
+            let conn_info: DatabaseInfo = serde_json::from_str(resp.text()?.as_str())?;
+            println!("got conn info: #{:?}", conn_info);
+            Ok(conn_info)
         } else {
             bail!("Auth failed")
         }
     }
-
-    pub fn get_database_uri(&self, _user: &str, _database: &str) -> Result<DatabaseInfo> {
-        Ok(DatabaseInfo {
-            host: "127.0.0.1".parse()?,
-            port: 5432,
-            dbname: "stas".to_string(),
-            user: "stas".to_string(),
-            password: "mypass".to_string(),
-        })
-    }
-
-    // pub fn create_database(&self, _user: &String, _database: &String) -> Result<DatabaseInfo> {
-    //     Ok(DatabaseInfo {
-    //         host: "127.0.0.1".parse()?,
-    //         port: 5432,
-    //         dbname: "stas".to_string(),
-    //         user: "stas".to_string(),
-    //         password: "mypass".to_string(),
-    //     })
-    // }
 }

proxy/src/main.rs

@@ -34,7 +34,7 @@ pub struct ProxyConf {
     pub redirect_uri: String,
 
     /// control plane address where we would check auth.
-    pub cplane_address: SocketAddr,
+    pub auth_endpoint: String,
 
     pub ssl_config: Option<Arc<ServerConfig>>,
 }
@@ -101,6 +101,14 @@ fn main() -> anyhow::Result<()> {
                 .help("redirect unauthenticated users to given uri")
                 .default_value("http://localhost:3000/psql_session/"),
         )
+        .arg(
+            Arg::with_name("auth-endpoint")
+                .short("a")
+                .long("auth-endpoint")
+                .takes_value(true)
+                .help("redirect unauthenticated users to given uri")
+                .default_value("http://localhost:3000/authenticate_proxy_request/"),
+        )
         .arg(
             Arg::with_name("ssl-key")
                 .short("k")
@@ -121,7 +129,7 @@ fn main() -> anyhow::Result<()> {
         proxy_address: arg_matches.value_of("proxy").unwrap().parse()?,
         mgmt_address: arg_matches.value_of("mgmt").unwrap().parse()?,
         redirect_uri: arg_matches.value_of("uri").unwrap().parse()?,
-        cplane_address: "127.0.0.1:3000".parse()?,
+        auth_endpoint: arg_matches.value_of("auth-endpoint").unwrap().parse()?,
         ssl_config: configure_ssl(&arg_matches)?,
     };
     let state = ProxyState {

proxy/src/proxy.rs

@@ -57,7 +57,7 @@ pub fn proxy_conn_main(
 ) -> anyhow::Result<()> {
     let mut conn = ProxyConnection {
         state,
-        cplane: CPlaneApi::new(&state.conf.cplane_address),
+        cplane: CPlaneApi::new(&state.conf.auth_endpoint),
         user: "".into(),
         database: "".into(),
         pgb: PostgresBackend::new(
@@ -80,6 +80,8 @@ pub fn proxy_conn_main(
         conn.handle_new_user()?
     };
 
+    // XXX: move that inside handle_new_user/handle_existing_user to be able to
+    // report wrong connection error.
     proxy_pass(conn.pgb, db_info)
 }
 
@@ -172,21 +174,30 @@ impl ProxyConnection {
                 .split_last()
                 .ok_or_else(|| anyhow::Error::msg("unexpected password message"))?;
 
-            if let Err(e) = self.check_auth_md5(md5_response) {
-                self.pgb
-                    .write_message(&BeMessage::ErrorResponse(format!("{}", e)))?;
-                bail!("auth failed: {}", e);
-            } else {
-                self.pgb
-                    .write_message_noflush(&BeMessage::AuthenticationOk)?;
-                self.pgb
-                    .write_message_noflush(&BeMessage::ParameterStatus)?;
-                self.pgb.write_message(&BeMessage::ReadyForQuery)?;
-            }
-        }
+            match self.cplane.authenticate_proxy_request(
+                self.user.as_str(),
+                md5_response,
+                &self.md5_salt,
+            ) {
+                Err(e) => {
+                    self.pgb
+                        .write_message(&BeMessage::ErrorResponse(format!("{}", e)))?;
 
-        // ok, we are authorized
-        self.cplane.get_database_uri(&self.user, &self.database)
+                    bail!("auth failed: {}", e);
+                }
+                Ok(conn_info) => {
+                    self.pgb
+                        .write_message_noflush(&BeMessage::AuthenticationOk)?;
+                    self.pgb
+                        .write_message_noflush(&BeMessage::ParameterStatus)?;
+                    self.pgb.write_message(&BeMessage::ReadyForQuery)?;
+
+                    return Ok(conn_info);
+                }
+            }
+        } else {
+            bail!("protocol violation");
+        }
     }
 
     fn handle_new_user(&mut self) -> anyhow::Result<DatabaseInfo> {
@@ -232,12 +243,6 @@ databases without opening the browser.
 
         Ok(dbinfo)
     }
-
-    fn check_auth_md5(&self, md5_response: &[u8]) -> anyhow::Result<()> {
-        assert!(self.is_existing_user());
-        self.cplane
-            .check_auth(self.user.as_str(), md5_response, &self.md5_salt)
-    }
 }
 
 /// Create a TCP connection to a postgres database, authenticate with it, and receive the ReadyForQuery message
@@ -273,7 +278,9 @@ fn proxy(
 
 /// Proxy a client connection to a postgres database
 fn proxy_pass(pgb: PostgresBackend, db_info: DatabaseInfo) -> anyhow::Result<()> {
-    let runtime = tokio::runtime::Builder::new_current_thread().build()?;
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()?;
     let db_stream = runtime.block_on(connect_to_db(db_info))?;
     let db_stream = db_stream.into_std()?;
     db_stream.set_nonblocking(false)?;