连接选项与建立连接

    pub fn new_without_pgpass() -> Self {
        let port = var("PGPORT")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(5432);

        let host = var("PGHOSTADDR")
            .ok()
            .or_else(|| var("PGHOST").ok())
            .unwrap_or_else(|| default_host(port));

        let username = if let Ok(username) = var("PGUSER") {
            username
        } else if let Ok(username) = whoami::username() {
            username
        } else {
            // keep the same fallback as previous version
            "unknown".to_string()
        };

        let database = var("PGDATABASE").ok();

        PgConnectOptions {
            port,
            host,
            socket: None,
            username,
            password: var("PGPASSWORD").ok(),
            // ...
            ssl_mode: var("PGSSLMODE")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or_default(),
            statement_cache_capacity: 100,
            application_name: var("PGAPPNAME").ok(),
            extra_float_digits: Some("2".into()),
            log_settings: Default::default(),
            options: var("PGOPTIONS").ok(),
        }
    }

        let mut options = Self::new_without_pgpass();

                _ => tracing::warn!(%key, %value, "ignoring unrecognized connect parameter"),

    pub(crate) fn apply_pgpass(mut self) -> Self {
        if self.password.is_none() {
            self.password = pgpass::load_password(
                &self.host,
                self.port,
                &self.username,
                self.database.as_deref(),
            );
        }

        self
    }

use sqlx::postgres::PgConnectOptions;
use sqlx::{ConnectOptions, Connection, PgConnection};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
  tracing_subscriber::fmt()
    .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
    .without_time()
    .with_ansi(false)
    .init();

  // 1. what new() picks up from the environment
  let o = PgConnectOptions::new();
  println!(
    "new(): host={} port={} user={} db={:?} ssl={:?}",
    o.get_host(),
    o.get_port(),
    o.get_username(),
    o.get_database(),
    o.get_ssl_mode()
  );

  // 2. URL: host starting with '/' (percent-encoded) means a Unix socket directory
  let o: PgConnectOptions = "postgres://%2Fvar%2Frun%2Fpostgresql/lab".parse()?;
  println!("socket url: host={} socket={:?}", o.get_host(), o.get_socket());

  // 3. query parameters, including one sqlx does not know
  let o: PgConnectOptions =
    "postgres://lab@192.168.64.106/lab?sslmode=disable&application_name=pg1&options[search_path]=pg1&connect_timeout=5"
      .parse()?;
  println!("options = {:?}", o.get_options());
  println!("to_url_lossy = {}", o.to_url_lossy());

  // 4. no password in the URL: the password comes from PGPASSWORD / the passfile, or nowhere
  match PgConnection::connect_with(&o).await {
    Ok(c) => {
      println!("connected without a password in the URL");
      c.close().await?;
    }
    Err(e) => println!("connect error: {e}"),
  }
  Ok(())
}

== A：没有任何 PG* 变量
new(): host=localhost port=5432 user=alice db=None ssl=Prefer
socket url: host=localhost socket=Some("/var/run/postgresql")
 WARN sqlx_postgres::options::parse: ignoring unrecognized connect parameter key=connect_timeout value=5
options = Some("-c search_path=pg1")
to_url_lossy = postgres://lab@192.168.64.106:5432/lab?sslmode=disable&statement-cache-capacity=100
connect error: error returned from database: password authentication failed for user "lab" at line 331
== B：只加 PGPASSFILE
to_url_lossy = postgres://lab:lab@192.168.64.106:5432/lab?sslmode=disable&statement-cache-capacity=100
connected without a password in the URL

        let mut params = vec![
            // Sets the display format for date and time values,
            // as well as the rules for interpreting ambiguous date input values.
            ("DateStyle", "ISO, MDY"),
            // Sets the client-side encoding (character set).
            // <https://www.postgresql.org/docs/devel/multibyte.html#MULTIBYTE-CHARSET-SUPPORTED>
            ("client_encoding", "UTF8"),
            // Sets the time zone for displaying and interpreting time stamps.
            ("TimeZone", "UTC"),
        ];

        if let Some(ref extra_float_digits) = options.extra_float_digits {
            params.push(("extra_float_digits", extra_float_digits));
        }

        if let Some(ref application_name) = options.application_name {
            params.push(("application_name", application_name));
        }

        if let Some(ref options) = options.options {
            params.push(("options", options));
        }

        stream.write(Startup {
            username: Some(&options.username),
            database: options.database.as_deref(),
            params: &params,
        })?;

        stream.flush().await?;

        loop {
            let message = stream.recv().await?;
            match message.format {
                BackendMessageFormat::Authentication => match message.decode()? {
                    Authentication::Ok => {
                        // the authentication exchange is successfully completed
                        // do nothing; no more information is required to continue
                    }

                    Authentication::CleartextPassword => {
                        // ...
                        stream
                            .send(Password::Cleartext(
                                options.password.as_deref().unwrap_or_default(),
                            ))
                            .await?;
                    }

                    Authentication::Md5Password(body) => {
                        // ...
                        stream
                            .send(Password::Md5 {
                                username: &options.username,
                                password: options.password.as_deref().unwrap_or_default(),
                                salt: body.salt,
                            })
                            .await?;
                    }

                    Authentication::Sasl(body) => {
                        sasl::authenticate(&mut stream, options, body).await?;
                    }

                    method => {
                        return Err(err_protocol!(
                            "unsupported authentication method: {:?}",
                            method
                        ));
                    }
                },

                BackendMessageFormat::BackendKeyData => {
                    // ...
                    let data: BackendKeyData = message.decode()?;

                    process_id = data.process_id;
                    secret_key = data.secret_key;
                }

                BackendMessageFormat::ReadyForQuery => {
                    // start-up is completed. The frontend can now issue commands
                    transaction_status = message.decode::<ReadyForQuery>()?.transaction_status;

                    break;
                }
                // ...
            }
        }

pub const UPSTREAM: &str = "192.168.64.106:5432";

pub fn direct() -> anyhow::Result<PgConnectOptions> {
  Ok(std::env::var("DATABASE_URL")?.parse::<PgConnectOptions>()?.options([("search_path", "pg1")]))
}

// Start a proxy and return options pointing at it (TLS off unless the caller changes it).
pub async fn via_proxy() -> anyhow::Result<(proxy::Proxy, PgConnectOptions)> {
  let p = proxy::start(UPSTREAM).await?;
  let opts = direct()?.host("127.0.0.1").port(p.port).ssl_mode(PgSslMode::Disable);
  Ok((p, opts))
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
  // how is the password stored on the server?
  let mut c = PgConnection::connect_with(&direct()?).await?;
  let enc: String = sqlx::query_scalar("show password_encryption").fetch_one(&mut c).await?;
  let stored: String = sqlx::query_scalar("select left(rolpassword, 14) from pg_authid where rolname = current_user")
    .fetch_one(&mut c)
    .await?;
  println!("password_encryption = {enc}, stored = {stored}...");
  c.close().await?;

  let (_p, opts) = via_proxy().await?;
  for mode in [PgSslMode::Prefer, PgSslMode::Require] {
    println!("--- ssl_mode = {mode:?} ---");
    let r = PgConnection::connect_with(&opts.clone().ssl_mode(mode).application_name("pg1-startup")).await;
    settle().await;
    match r {
      Ok(conn) => {
        println!("connected, server_version_num = {:?}", conn.server_version_num());
        conn.close().await?;
        settle().await;
      }
      Err(e) => println!("error: {e}"),
    }
  }
  Ok(())
}

password_encryption = scram-sha-256, stored = SCRAM-SHA-256$...
--- ssl_mode = Prefer ---
  [conn 1] C->S SSLRequest
  [conn 1] S->C SSL 回复 'N'
  [conn 1] C->S StartupMessage(v3.0 user=lab database=lab DateStyle=ISO, MDY client_encoding=UTF8 TimeZone=UTC extra_float_digits=2 application_name=pg1-startup options=-c search_path=pg1)
  [conn 1] S->C AuthenticationSASL(SCRAM-SHA-256)
  [conn 1] C->S SASLInitialResponse(SCRAM-SHA-256, "n,,n=lab,r=xX`TIl…(96)")
  [conn 1] S->C AuthenticationSASLContinue("r=xX`TIl…(120),s=pmdfF7…(26),i=4096")
  [conn 1] C->S SASLResponse("c=biws,r=xX`TIl…(120),p=3L6fHM…(46)")
  [conn 1] S->C AuthenticationSASLFinal("v=bh2DwO…(46)"), AuthenticationOk, ParameterStatus(in_hot_standby=off), ParameterStatus(integer_datetimes=on), ParameterStatus(TimeZone=UTC), ParameterStatus(IntervalStyle=postgres), ParameterStatus(is_superuser=on), ParameterStatus(application_name=pg1-startup), ParameterStatus(default_transaction_read_only=off), ParameterStatus(scram_iterations=4096), ParameterStatus(DateStyle=ISO, MDY), ParameterStatus(standard_conforming_strings=on), ParameterStatus(session_authorization=lab), ParameterStatus(client_encoding=UTF8), ParameterStatus(server_version=16.14), ParameterStatus(server_encoding=UTF8), BackendKeyData, ReadyForQuery(I)
connected, server_version_num = Some(160014)
  [conn 1] C->S Terminate
--- ssl_mode = Require ---
  [conn 2] C->S SSLRequest
  [conn 2] S->C SSL 回复 'N'
error: error occurred while attempting to establish a TLS connection: server does not support TLS

    // authentication is only considered valid if this verification passes
    mac.verify_slice(&data.verifier).map_err(Error::protocol)?;

    for i in 1..iter_count {
        mac.update(u.as_slice());
        u = mac.finalize_reset().into_bytes();
        hi = hi.iter().zip(u.iter()).map(|(&a, &b)| a ^ b).collect();

        // For large iteration counts, this process can take a long time and block the event loop.
        // It was measured as taking ~50ms for 4096 iterations (the default) on a developer machine.
        // If we want to yield every 10-100us (as generally advised for tokio), then we can yield
        // every 5 iterations which should be every ~50us.
        if i % 5 == 0 {
            rt::yield_now().await;
        }
    }

    match options.ssl_mode {
        // FIXME: Implement ALLOW
        PgSslMode::Allow | PgSslMode::Disable => return Ok(Box::new(socket)),

        PgSslMode::Prefer => {
            if !tls::available() {
                return Ok(Box::new(socket));
            }

            // try upgrade, but its okay if we fail
            if !request_upgrade(&mut socket, options).await? {
                return Ok(Box::new(socket));
            }
        }

        PgSslMode::Require | PgSslMode::VerifyFull | PgSslMode::VerifyCa => {
            tls::error_if_unavailable()?;

            if !request_upgrade(&mut socket, options).await? {
                // upgrade failed, die
                return Err(Error::Tls("server does not support TLS".into()));
            }
        }
    }

    let accept_invalid_certs = !matches!(
        options.ssl_mode,
        PgSslMode::VerifyCa | PgSslMode::VerifyFull
    );
    let accept_invalid_hostnames = !matches!(options.ssl_mode, PgSslMode::VerifyFull);

== C：B 的环境再加 PGPASSWORD=postgres
to_url_lossy = postgres://lab:postgres@192.168.64.106:5432/lab?sslmode=disable&statement-cache-capacity=100
connect error: error returned from database: password authentication failed for user "lab" at line 331