LCOV - code coverage report
Current view: top level - proxy/src/auth/backend - mod.rs (source / functions) Coverage Total Hit
Test: 6df3fc19ec669bcfbbf9aba41d1338898d24eaa0.info Lines: 73.9 % 555 410
Test Date: 2025-03-12 18:28:53 Functions: 39.1 % 87 34

            Line data    Source code
       1              : mod classic;
       2              : mod console_redirect;
       3              : mod hacks;
       4              : pub mod jwt;
       5              : pub mod local;
       6              : 
       7              : use std::net::IpAddr;
       8              : use std::sync::Arc;
       9              : 
      10              : pub use console_redirect::ConsoleRedirectBackend;
      11              : pub(crate) use console_redirect::ConsoleRedirectError;
      12              : use ipnet::{Ipv4Net, Ipv6Net};
      13              : use local::LocalBackend;
      14              : use postgres_client::config::AuthKeys;
      15              : use serde::{Deserialize, Serialize};
      16              : use tokio::io::{AsyncRead, AsyncWrite};
      17              : use tracing::{debug, info, warn};
      18              : 
      19              : use crate::auth::credentials::check_peer_addr_is_in_list;
      20              : use crate::auth::{
      21              :     self, AuthError, ComputeUserInfoMaybeEndpoint, IpPattern, validate_password_and_exchange,
      22              : };
      23              : use crate::cache::Cached;
      24              : use crate::config::AuthenticationConfig;
      25              : use crate::context::RequestContext;
      26              : use crate::control_plane::client::ControlPlaneClient;
      27              : use crate::control_plane::errors::GetAuthInfoError;
      28              : use crate::control_plane::{
      29              :     self, AccessBlockerFlags, AuthSecret, CachedAccessBlockerFlags, CachedAllowedIps,
      30              :     CachedAllowedVpcEndpointIds, CachedNodeInfo, CachedRoleSecret, ControlPlaneApi,
      31              : };
      32              : use crate::intern::EndpointIdInt;
      33              : use crate::metrics::Metrics;
      34              : use crate::protocol2::ConnectionInfoExtra;
      35              : use crate::proxy::NeonOptions;
      36              : use crate::proxy::connect_compute::ComputeConnectBackend;
      37              : use crate::rate_limiter::{BucketRateLimiter, EndpointRateLimiter};
      38              : use crate::stream::Stream;
      39              : use crate::types::{EndpointCacheKey, EndpointId, RoleName};
      40              : use crate::{scram, stream};
      41              : 
      42              : /// Alternative to [`std::borrow::Cow`] but doesn't need `T: ToOwned` as we don't need that functionality
      43              : pub enum MaybeOwned<'a, T> {
      44              :     Owned(T),
      45              :     Borrowed(&'a T),
      46              : }
      47              : 
      48              : impl<T> std::ops::Deref for MaybeOwned<'_, T> {
      49              :     type Target = T;
      50              : 
      51           13 :     fn deref(&self) -> &Self::Target {
      52           13 :         match self {
      53           13 :             MaybeOwned::Owned(t) => t,
      54            0 :             MaybeOwned::Borrowed(t) => t,
      55              :         }
      56           13 :     }
      57              : }
      58              : 
      59              : /// This type serves two purposes:
      60              : ///
      61              : /// * When `T` is `()`, it's just a regular auth backend selector
      62              : ///   which we use in [`crate::config::ProxyConfig`].
      63              : ///
      64              : /// * However, when we substitute `T` with [`ComputeUserInfoMaybeEndpoint`],
      65              : ///   this helps us provide the credentials only to those auth
      66              : ///   backends which require them for the authentication process.
      67              : pub enum Backend<'a, T> {
      68              :     /// Cloud API (V2).
      69              :     ControlPlane(MaybeOwned<'a, ControlPlaneClient>, T),
      70              :     /// Local proxy uses configured auth credentials and does not wake compute
      71              :     Local(MaybeOwned<'a, LocalBackend>),
      72              : }
      73              : 
      74              : impl std::fmt::Display for Backend<'_, ()> {
      75            0 :     fn fmt(&self, fmt: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
      76            0 :         match self {
      77            0 :             Self::ControlPlane(api, ()) => match &**api {
      78            0 :                 ControlPlaneClient::ProxyV1(endpoint) => fmt
      79            0 :                     .debug_tuple("ControlPlane::ProxyV1")
      80            0 :                     .field(&endpoint.url())
      81            0 :                     .finish(),
      82              :                 #[cfg(any(test, feature = "testing"))]
      83            0 :                 ControlPlaneClient::PostgresMock(endpoint) => fmt
      84            0 :                     .debug_tuple("ControlPlane::PostgresMock")
      85            0 :                     .field(&endpoint.url())
      86            0 :                     .finish(),
      87              :                 #[cfg(test)]
      88            0 :                 ControlPlaneClient::Test(_) => fmt.debug_tuple("ControlPlane::Test").finish(),
      89              :             },
      90            0 :             Self::Local(_) => fmt.debug_tuple("Local").finish(),
      91              :         }
      92            0 :     }
      93              : }
      94              : 
      95              : impl<T> Backend<'_, T> {
      96              :     /// Very similar to [`std::option::Option::as_ref`].
      97              :     /// This helps us pass structured config to async tasks.
      98            0 :     pub(crate) fn as_ref(&self) -> Backend<'_, &T> {
      99            0 :         match self {
     100            0 :             Self::ControlPlane(c, x) => Backend::ControlPlane(MaybeOwned::Borrowed(c), x),
     101            0 :             Self::Local(l) => Backend::Local(MaybeOwned::Borrowed(l)),
     102              :         }
     103            0 :     }
     104              : 
     105            0 :     pub(crate) fn get_api(&self) -> &ControlPlaneClient {
     106            0 :         match self {
     107            0 :             Self::ControlPlane(api, _) => api,
     108            0 :             Self::Local(_) => panic!("Local backend has no API"),
     109              :         }
     110            0 :     }
     111              : 
     112            0 :     pub(crate) fn is_local_proxy(&self) -> bool {
     113            0 :         matches!(self, Self::Local(_))
     114            0 :     }
     115              : }
     116              : 
     117              : impl<'a, T> Backend<'a, T> {
     118              :     /// Very similar to [`std::option::Option::map`].
     119              :     /// Maps [`Backend<T>`] to [`Backend<R>`] by applying
     120              :     /// a function to a contained value.
     121            0 :     pub(crate) fn map<R>(self, f: impl FnOnce(T) -> R) -> Backend<'a, R> {
     122            0 :         match self {
     123            0 :             Self::ControlPlane(c, x) => Backend::ControlPlane(c, f(x)),
     124            0 :             Self::Local(l) => Backend::Local(l),
     125              :         }
     126            0 :     }
     127              : }
     128              : impl<'a, T, E> Backend<'a, Result<T, E>> {
     129              :     /// Very similar to [`std::option::Option::transpose`].
     130              :     /// This is most useful for error handling.
     131            0 :     pub(crate) fn transpose(self) -> Result<Backend<'a, T>, E> {
     132            0 :         match self {
     133            0 :             Self::ControlPlane(c, x) => x.map(|x| Backend::ControlPlane(c, x)),
     134            0 :             Self::Local(l) => Ok(Backend::Local(l)),
     135              :         }
     136            0 :     }
     137              : }
     138              : 
     139              : pub(crate) struct ComputeCredentials {
     140              :     pub(crate) info: ComputeUserInfo,
     141              :     pub(crate) keys: ComputeCredentialKeys,
     142              : }
     143              : 
     144              : #[derive(Debug, Clone)]
     145              : pub(crate) struct ComputeUserInfoNoEndpoint {
     146              :     pub(crate) user: RoleName,
     147              :     pub(crate) options: NeonOptions,
     148              : }
     149              : 
     150            0 : #[derive(Debug, Clone, Default, Serialize, Deserialize)]
     151              : pub(crate) struct ComputeUserInfo {
     152              :     pub(crate) endpoint: EndpointId,
     153              :     pub(crate) user: RoleName,
     154              :     pub(crate) options: NeonOptions,
     155              : }
     156              : 
     157              : impl ComputeUserInfo {
     158            2 :     pub(crate) fn endpoint_cache_key(&self) -> EndpointCacheKey {
     159            2 :         self.options.get_cache_key(&self.endpoint)
     160            2 :     }
     161              : }
     162              : 
     163              : #[cfg_attr(test, derive(Debug))]
     164              : pub(crate) enum ComputeCredentialKeys {
     165              :     #[cfg(any(test, feature = "testing"))]
     166              :     Password(Vec<u8>),
     167              :     AuthKeys(AuthKeys),
     168              :     JwtPayload(Vec<u8>),
     169              :     None,
     170              : }
     171              : 
     172              : impl TryFrom<ComputeUserInfoMaybeEndpoint> for ComputeUserInfo {
     173              :     // user name
     174              :     type Error = ComputeUserInfoNoEndpoint;
     175              : 
     176            3 :     fn try_from(user_info: ComputeUserInfoMaybeEndpoint) -> Result<Self, Self::Error> {
     177            3 :         match user_info.endpoint_id {
     178            1 :             None => Err(ComputeUserInfoNoEndpoint {
     179            1 :                 user: user_info.user,
     180            1 :                 options: user_info.options,
     181            1 :             }),
     182            2 :             Some(endpoint) => Ok(ComputeUserInfo {
     183            2 :                 endpoint,
     184            2 :                 user: user_info.user,
     185            2 :                 options: user_info.options,
     186            2 :             }),
     187              :         }
     188            3 :     }
     189              : }
     190              : 
     191              : #[derive(PartialEq, PartialOrd, Hash, Eq, Ord, Debug, Copy, Clone)]
     192              : pub struct MaskedIp(IpAddr);
     193              : 
     194              : impl MaskedIp {
     195           15 :     fn new(value: IpAddr, prefix: u8) -> Self {
     196           15 :         match value {
     197           11 :             IpAddr::V4(v4) => Self(IpAddr::V4(
     198           11 :                 Ipv4Net::new(v4, prefix).map_or(v4, |x| x.trunc().addr()),
     199           11 :             )),
     200            4 :             IpAddr::V6(v6) => Self(IpAddr::V6(
     201            4 :                 Ipv6Net::new(v6, prefix).map_or(v6, |x| x.trunc().addr()),
     202            4 :             )),
     203              :         }
     204           15 :     }
     205              : }
     206              : 
     207              : // This can't be just per IP because that would limit some PaaS that share IP addresses
     208              : pub type AuthRateLimiter = BucketRateLimiter<(EndpointIdInt, MaskedIp)>;
     209              : 
     210              : impl AuthenticationConfig {
     211            3 :     pub(crate) fn check_rate_limit(
     212            3 :         &self,
     213            3 :         ctx: &RequestContext,
     214            3 :         secret: AuthSecret,
     215            3 :         endpoint: &EndpointId,
     216            3 :         is_cleartext: bool,
     217            3 :     ) -> auth::Result<AuthSecret> {
     218            3 :         // we have validated the endpoint exists, so let's intern it.
     219            3 :         let endpoint_int = EndpointIdInt::from(endpoint.normalize());
     220              : 
     221              :         // only count the full hash count if password hack or websocket flow.
     222              :         // in other words, if proxy needs to run the hashing
     223            3 :         let password_weight = if is_cleartext {
     224            2 :             match &secret {
     225              :                 #[cfg(any(test, feature = "testing"))]
     226            0 :                 AuthSecret::Md5(_) => 1,
     227            2 :                 AuthSecret::Scram(s) => s.iterations + 1,
     228              :             }
     229              :         } else {
     230              :             // validating scram takes just 1 hmac_sha_256 operation.
     231            1 :             1
     232              :         };
     233              : 
     234            3 :         let limit_not_exceeded = self.rate_limiter.check(
     235            3 :             (
     236            3 :                 endpoint_int,
     237            3 :                 MaskedIp::new(ctx.peer_addr(), self.rate_limit_ip_subnet),
     238            3 :             ),
     239            3 :             password_weight,
     240            3 :         );
     241            3 : 
     242            3 :         if !limit_not_exceeded {
     243            0 :             warn!(
     244              :                 enabled = self.rate_limiter_enabled,
     245            0 :                 "rate limiting authentication"
     246              :             );
     247            0 :             Metrics::get().proxy.requests_auth_rate_limits_total.inc();
     248            0 :             Metrics::get()
     249            0 :                 .proxy
     250            0 :                 .endpoints_auth_rate_limits
     251            0 :                 .get_metric()
     252            0 :                 .measure(endpoint);
     253            0 : 
     254            0 :             if self.rate_limiter_enabled {
     255            0 :                 return Err(auth::AuthError::too_many_connections());
     256            0 :             }
     257            3 :         }
     258              : 
     259            3 :         Ok(secret)
     260            3 :     }
     261              : }
     262              : 
     263              : /// True to its name, this function encapsulates our current auth trade-offs.
     264              : /// Here, we choose the appropriate auth flow based on circumstances.
     265              : ///
     266              : /// All authentication flows will emit an AuthenticationOk message if successful.
     267            3 : async fn auth_quirks(
     268            3 :     ctx: &RequestContext,
     269            3 :     api: &impl control_plane::ControlPlaneApi,
     270            3 :     user_info: ComputeUserInfoMaybeEndpoint,
     271            3 :     client: &mut stream::PqStream<Stream<impl AsyncRead + AsyncWrite + Unpin>>,
     272            3 :     allow_cleartext: bool,
     273            3 :     config: &'static AuthenticationConfig,
     274            3 :     endpoint_rate_limiter: Arc<EndpointRateLimiter>,
     275            3 : ) -> auth::Result<(ComputeCredentials, Option<Vec<IpPattern>>)> {
     276              :     // If there's no project so far, that entails that client doesn't
     277              :     // support SNI or other means of passing the endpoint (project) name.
     278              :     // We now expect to see a very specific payload in the place of password.
     279            3 :     let (info, unauthenticated_password) = match user_info.try_into() {
     280            1 :         Err(info) => {
     281            1 :             let (info, password) =
     282            1 :                 hacks::password_hack_no_authentication(ctx, info, client).await?;
     283            1 :             ctx.set_endpoint_id(info.endpoint.clone());
     284            1 :             (info, Some(password))
     285              :         }
     286            2 :         Ok(info) => (info, None),
     287              :     };
     288              : 
     289            3 :     debug!("fetching authentication info and allowlists");
     290              : 
     291              :     // check allowed list
     292            3 :     let allowed_ips = if config.ip_allowlist_check_enabled {
     293            3 :         let allowed_ips = api.get_allowed_ips(ctx, &info).await?;
     294            3 :         if !check_peer_addr_is_in_list(&ctx.peer_addr(), &allowed_ips) {
     295            0 :             return Err(auth::AuthError::ip_address_not_allowed(ctx.peer_addr()));
     296            3 :         }
     297            3 :         allowed_ips
     298              :     } else {
     299            0 :         Cached::new_uncached(Arc::new(vec![]))
     300              :     };
     301              : 
     302              :     // check if a VPC endpoint ID is coming in and if yes, if it's allowed
     303            3 :     let access_blocks = api.get_block_public_or_vpc_access(ctx, &info).await?;
     304            3 :     if config.is_vpc_acccess_proxy {
     305            0 :         if access_blocks.vpc_access_blocked {
     306            0 :             return Err(AuthError::NetworkNotAllowed);
     307            0 :         }
     308              : 
     309            0 :         let incoming_vpc_endpoint_id = match ctx.extra() {
     310            0 :             None => return Err(AuthError::MissingEndpointName),
     311            0 :             Some(ConnectionInfoExtra::Aws { vpce_id }) => vpce_id.to_string(),
     312            0 :             Some(ConnectionInfoExtra::Azure { link_id }) => link_id.to_string(),
     313              :         };
     314            0 :         let allowed_vpc_endpoint_ids = api.get_allowed_vpc_endpoint_ids(ctx, &info).await?;
     315              :         // TODO: For now an empty VPC endpoint ID list means all are allowed. We should replace that.
     316            0 :         if !allowed_vpc_endpoint_ids.is_empty()
     317            0 :             && !allowed_vpc_endpoint_ids.contains(&incoming_vpc_endpoint_id)
     318              :         {
     319            0 :             return Err(AuthError::vpc_endpoint_id_not_allowed(
     320            0 :                 incoming_vpc_endpoint_id,
     321            0 :             ));
     322            0 :         }
     323            3 :     } else if access_blocks.public_access_blocked {
     324            0 :         return Err(AuthError::NetworkNotAllowed);
     325            3 :     }
     326              : 
     327            3 :     if !endpoint_rate_limiter.check(info.endpoint.clone().into(), 1) {
     328            0 :         return Err(AuthError::too_many_connections());
     329            3 :     }
     330            3 :     let cached_secret = api.get_role_secret(ctx, &info).await?;
     331            3 :     let (cached_entry, secret) = cached_secret.take_value();
     332              : 
     333            3 :     let secret = if let Some(secret) = secret {
     334            3 :         config.check_rate_limit(
     335            3 :             ctx,
     336            3 :             secret,
     337            3 :             &info.endpoint,
     338            3 :             unauthenticated_password.is_some() || allow_cleartext,
     339            0 :         )?
     340              :     } else {
     341              :         // If we don't have an authentication secret, we mock one to
     342              :         // prevent malicious probing (possible due to missing protocol steps).
     343              :         // This mocked secret will never lead to successful authentication.
     344            0 :         info!("authentication info not found, mocking it");
     345            0 :         AuthSecret::Scram(scram::ServerSecret::mock(rand::random()))
     346              :     };
     347              : 
     348            3 :     match authenticate_with_secret(
     349            3 :         ctx,
     350            3 :         secret,
     351            3 :         info,
     352            3 :         client,
     353            3 :         unauthenticated_password,
     354            3 :         allow_cleartext,
     355            3 :         config,
     356            3 :     )
     357            3 :     .await
     358              :     {
     359            3 :         Ok(keys) => Ok((keys, Some(allowed_ips.as_ref().clone()))),
     360            0 :         Err(e) => {
     361            0 :             if e.is_password_failed() {
     362            0 :                 // The password could have been changed, so we invalidate the cache.
     363            0 :                 cached_entry.invalidate();
     364            0 :             }
     365            0 :             Err(e)
     366              :         }
     367              :     }
     368            3 : }
     369              : 
     370            3 : async fn authenticate_with_secret(
     371            3 :     ctx: &RequestContext,
     372            3 :     secret: AuthSecret,
     373            3 :     info: ComputeUserInfo,
     374            3 :     client: &mut stream::PqStream<Stream<impl AsyncRead + AsyncWrite + Unpin>>,
     375            3 :     unauthenticated_password: Option<Vec<u8>>,
     376            3 :     allow_cleartext: bool,
     377            3 :     config: &'static AuthenticationConfig,
     378            3 : ) -> auth::Result<ComputeCredentials> {
     379            3 :     if let Some(password) = unauthenticated_password {
     380            1 :         let ep = EndpointIdInt::from(&info.endpoint);
     381              : 
     382            1 :         let auth_outcome =
     383            1 :             validate_password_and_exchange(&config.thread_pool, ep, &password, secret).await?;
     384            1 :         let keys = match auth_outcome {
     385            1 :             crate::sasl::Outcome::Success(key) => key,
     386            0 :             crate::sasl::Outcome::Failure(reason) => {
     387            0 :                 info!("auth backend failed with an error: {reason}");
     388            0 :                 return Err(auth::AuthError::password_failed(&*info.user));
     389              :             }
     390              :         };
     391              : 
     392              :         // we have authenticated the password
     393            1 :         client.write_message_noflush(&pq_proto::BeMessage::AuthenticationOk)?;
     394              : 
     395            1 :         return Ok(ComputeCredentials { info, keys });
     396            2 :     }
     397            2 : 
     398            2 :     // -- the remaining flows are self-authenticating --
     399            2 : 
     400            2 :     // Perform cleartext auth if we're allowed to do that.
     401            2 :     // Currently, we use it for websocket connections (latency).
     402            2 :     if allow_cleartext {
     403            1 :         ctx.set_auth_method(crate::context::AuthMethod::Cleartext);
     404            1 :         return hacks::authenticate_cleartext(ctx, info, client, secret, config).await;
     405            1 :     }
     406            1 : 
     407            1 :     // Finally, proceed with the main auth flow (SCRAM-based).
     408            1 :     classic::authenticate(ctx, info, client, config, secret).await
     409            3 : }
     410              : 
     411              : impl<'a> Backend<'a, ComputeUserInfoMaybeEndpoint> {
     412              :     /// Get username from the credentials.
     413            0 :     pub(crate) fn get_user(&self) -> &str {
     414            0 :         match self {
     415            0 :             Self::ControlPlane(_, user_info) => &user_info.user,
     416            0 :             Self::Local(_) => "local",
     417              :         }
     418            0 :     }
     419              : 
     420              :     /// Authenticate the client via the requested backend, possibly using credentials.
     421              :     #[tracing::instrument(fields(allow_cleartext = allow_cleartext), skip_all)]
     422              :     pub(crate) async fn authenticate(
     423              :         self,
     424              :         ctx: &RequestContext,
     425              :         client: &mut stream::PqStream<Stream<impl AsyncRead + AsyncWrite + Unpin>>,
     426              :         allow_cleartext: bool,
     427              :         config: &'static AuthenticationConfig,
     428              :         endpoint_rate_limiter: Arc<EndpointRateLimiter>,
     429              :     ) -> auth::Result<(Backend<'a, ComputeCredentials>, Option<Vec<IpPattern>>)> {
     430              :         let res = match self {
     431              :             Self::ControlPlane(api, user_info) => {
     432              :                 debug!(
     433              :                     user = &*user_info.user,
     434              :                     project = user_info.endpoint(),
     435              :                     "performing authentication using the console"
     436              :                 );
     437              : 
     438              :                 let (credentials, ip_allowlist) = auth_quirks(
     439              :                     ctx,
     440              :                     &*api,
     441              :                     user_info,
     442              :                     client,
     443              :                     allow_cleartext,
     444              :                     config,
     445              :                     endpoint_rate_limiter,
     446              :                 )
     447              :                 .await?;
     448              :                 Ok((Backend::ControlPlane(api, credentials), ip_allowlist))
     449              :             }
     450              :             Self::Local(_) => {
     451              :                 return Err(auth::AuthError::bad_auth_method("invalid for local proxy"));
     452              :             }
     453              :         };
     454              : 
     455              :         // TODO: replace with some metric
     456              :         info!("user successfully authenticated");
     457              :         res
     458              :     }
     459              : }
     460              : 
     461              : impl Backend<'_, ComputeUserInfo> {
     462            0 :     pub(crate) async fn get_role_secret(
     463            0 :         &self,
     464            0 :         ctx: &RequestContext,
     465            0 :     ) -> Result<CachedRoleSecret, GetAuthInfoError> {
     466            0 :         match self {
     467            0 :             Self::ControlPlane(api, user_info) => api.get_role_secret(ctx, user_info).await,
     468            0 :             Self::Local(_) => Ok(Cached::new_uncached(None)),
     469              :         }
     470            0 :     }
     471              : 
     472            0 :     pub(crate) async fn get_allowed_ips(
     473            0 :         &self,
     474            0 :         ctx: &RequestContext,
     475            0 :     ) -> Result<CachedAllowedIps, GetAuthInfoError> {
     476            0 :         match self {
     477            0 :             Self::ControlPlane(api, user_info) => api.get_allowed_ips(ctx, user_info).await,
     478            0 :             Self::Local(_) => Ok(Cached::new_uncached(Arc::new(vec![]))),
     479              :         }
     480            0 :     }
     481              : 
     482            0 :     pub(crate) async fn get_allowed_vpc_endpoint_ids(
     483            0 :         &self,
     484            0 :         ctx: &RequestContext,
     485            0 :     ) -> Result<CachedAllowedVpcEndpointIds, GetAuthInfoError> {
     486            0 :         match self {
     487            0 :             Self::ControlPlane(api, user_info) => {
     488            0 :                 api.get_allowed_vpc_endpoint_ids(ctx, user_info).await
     489              :             }
     490            0 :             Self::Local(_) => Ok(Cached::new_uncached(Arc::new(vec![]))),
     491              :         }
     492            0 :     }
     493              : 
     494            0 :     pub(crate) async fn get_block_public_or_vpc_access(
     495            0 :         &self,
     496            0 :         ctx: &RequestContext,
     497            0 :     ) -> Result<CachedAccessBlockerFlags, GetAuthInfoError> {
     498            0 :         match self {
     499            0 :             Self::ControlPlane(api, user_info) => {
     500            0 :                 api.get_block_public_or_vpc_access(ctx, user_info).await
     501              :             }
     502            0 :             Self::Local(_) => Ok(Cached::new_uncached(AccessBlockerFlags::default())),
     503              :         }
     504            0 :     }
     505              : }
     506              : 
     507              : #[async_trait::async_trait]
     508              : impl ComputeConnectBackend for Backend<'_, ComputeCredentials> {
     509           13 :     async fn wake_compute(
     510           13 :         &self,
     511           13 :         ctx: &RequestContext,
     512           13 :     ) -> Result<CachedNodeInfo, control_plane::errors::WakeComputeError> {
     513           13 :         match self {
     514           13 :             Self::ControlPlane(api, creds) => api.wake_compute(ctx, &creds.info).await,
     515            0 :             Self::Local(local) => Ok(Cached::new_uncached(local.node_info.clone())),
     516              :         }
     517           26 :     }
     518              : 
     519            6 :     fn get_keys(&self) -> &ComputeCredentialKeys {
     520            6 :         match self {
     521            6 :             Self::ControlPlane(_, creds) => &creds.keys,
     522            0 :             Self::Local(_) => &ComputeCredentialKeys::None,
     523              :         }
     524            6 :     }
     525              : }
     526              : 
     527              : #[cfg(test)]
     528              : mod tests {
     529              :     #![allow(clippy::unimplemented, clippy::unwrap_used)]
     530              : 
     531              :     use std::net::IpAddr;
     532              :     use std::sync::Arc;
     533              :     use std::time::Duration;
     534              : 
     535              :     use bytes::BytesMut;
     536              :     use control_plane::AuthSecret;
     537              :     use fallible_iterator::FallibleIterator;
     538              :     use once_cell::sync::Lazy;
     539              :     use postgres_protocol::authentication::sasl::{ChannelBinding, ScramSha256};
     540              :     use postgres_protocol::message::backend::Message as PgMessage;
     541              :     use postgres_protocol::message::frontend;
     542              :     use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
     543              : 
     544              :     use super::jwt::JwkCache;
     545              :     use super::{AuthRateLimiter, auth_quirks};
     546              :     use crate::auth::backend::MaskedIp;
     547              :     use crate::auth::{ComputeUserInfoMaybeEndpoint, IpPattern};
     548              :     use crate::config::AuthenticationConfig;
     549              :     use crate::context::RequestContext;
     550              :     use crate::control_plane::{
     551              :         self, AccessBlockerFlags, CachedAccessBlockerFlags, CachedAllowedIps,
     552              :         CachedAllowedVpcEndpointIds, CachedNodeInfo, CachedRoleSecret,
     553              :     };
     554              :     use crate::proxy::NeonOptions;
     555              :     use crate::rate_limiter::{EndpointRateLimiter, RateBucketInfo};
     556              :     use crate::scram::ServerSecret;
     557              :     use crate::scram::threadpool::ThreadPool;
     558              :     use crate::stream::{PqStream, Stream};
     559              : 
     560              :     struct Auth {
     561              :         ips: Vec<IpPattern>,
     562              :         vpc_endpoint_ids: Vec<String>,
     563              :         access_blocker_flags: AccessBlockerFlags,
     564              :         secret: AuthSecret,
     565              :     }
     566              : 
     567              :     impl control_plane::ControlPlaneApi for Auth {
     568            3 :         async fn get_role_secret(
     569            3 :             &self,
     570            3 :             _ctx: &RequestContext,
     571            3 :             _user_info: &super::ComputeUserInfo,
     572            3 :         ) -> Result<CachedRoleSecret, control_plane::errors::GetAuthInfoError> {
     573            3 :             Ok(CachedRoleSecret::new_uncached(Some(self.secret.clone())))
     574            3 :         }
     575              : 
     576            3 :         async fn get_allowed_ips(
     577            3 :             &self,
     578            3 :             _ctx: &RequestContext,
     579            3 :             _user_info: &super::ComputeUserInfo,
     580            3 :         ) -> Result<CachedAllowedIps, control_plane::errors::GetAuthInfoError> {
     581            3 :             Ok(CachedAllowedIps::new_uncached(Arc::new(self.ips.clone())))
     582            3 :         }
     583              : 
     584            0 :         async fn get_allowed_vpc_endpoint_ids(
     585            0 :             &self,
     586            0 :             _ctx: &RequestContext,
     587            0 :             _user_info: &super::ComputeUserInfo,
     588            0 :         ) -> Result<CachedAllowedVpcEndpointIds, control_plane::errors::GetAuthInfoError> {
     589            0 :             Ok(CachedAllowedVpcEndpointIds::new_uncached(Arc::new(
     590            0 :                 self.vpc_endpoint_ids.clone(),
     591            0 :             )))
     592            0 :         }
     593              : 
     594            3 :         async fn get_block_public_or_vpc_access(
     595            3 :             &self,
     596            3 :             _ctx: &RequestContext,
     597            3 :             _user_info: &super::ComputeUserInfo,
     598            3 :         ) -> Result<CachedAccessBlockerFlags, control_plane::errors::GetAuthInfoError> {
     599            3 :             Ok(CachedAccessBlockerFlags::new_uncached(
     600            3 :                 self.access_blocker_flags.clone(),
     601            3 :             ))
     602            3 :         }
     603              : 
     604            0 :         async fn get_endpoint_jwks(
     605            0 :             &self,
     606            0 :             _ctx: &RequestContext,
     607            0 :             _endpoint: crate::types::EndpointId,
     608            0 :         ) -> Result<Vec<super::jwt::AuthRule>, control_plane::errors::GetEndpointJwksError>
     609            0 :         {
     610            0 :             unimplemented!()
     611              :         }
     612              : 
     613            0 :         async fn wake_compute(
     614            0 :             &self,
     615            0 :             _ctx: &RequestContext,
     616            0 :             _user_info: &super::ComputeUserInfo,
     617            0 :         ) -> Result<CachedNodeInfo, control_plane::errors::WakeComputeError> {
     618            0 :             unimplemented!()
     619              :         }
     620              :     }
     621              : 
     622            3 :     static CONFIG: Lazy<AuthenticationConfig> = Lazy::new(|| AuthenticationConfig {
     623            3 :         jwks_cache: JwkCache::default(),
     624            3 :         thread_pool: ThreadPool::new(1),
     625            3 :         scram_protocol_timeout: std::time::Duration::from_secs(5),
     626            3 :         rate_limiter_enabled: true,
     627            3 :         rate_limiter: AuthRateLimiter::new(&RateBucketInfo::DEFAULT_AUTH_SET),
     628            3 :         rate_limit_ip_subnet: 64,
     629            3 :         ip_allowlist_check_enabled: true,
     630            3 :         is_vpc_acccess_proxy: false,
     631            3 :         is_auth_broker: false,
     632            3 :         accept_jwts: false,
     633            3 :         console_redirect_confirmation_timeout: std::time::Duration::from_secs(5),
     634            3 :     });
     635              : 
     636            5 :     async fn read_message(r: &mut (impl AsyncRead + Unpin), b: &mut BytesMut) -> PgMessage {
     637              :         loop {
     638            7 :             r.read_buf(&mut *b).await.unwrap();
     639            7 :             if let Some(m) = PgMessage::parse(&mut *b).unwrap() {
     640            5 :                 break m;
     641            2 :             }
     642              :         }
     643            5 :     }
     644              : 
     645              :     #[test]
     646            1 :     fn masked_ip() {
     647            1 :         let ip_a = IpAddr::V4([127, 0, 0, 1].into());
     648            1 :         let ip_b = IpAddr::V4([127, 0, 0, 2].into());
     649            1 :         let ip_c = IpAddr::V4([192, 168, 1, 101].into());
     650            1 :         let ip_d = IpAddr::V4([192, 168, 1, 102].into());
     651            1 :         let ip_e = IpAddr::V6("abcd:abcd:abcd:abcd:abcd:abcd:abcd:abcd".parse().unwrap());
     652            1 :         let ip_f = IpAddr::V6("abcd:abcd:abcd:abcd:1234:abcd:abcd:abcd".parse().unwrap());
     653            1 : 
     654            1 :         assert_ne!(MaskedIp::new(ip_a, 64), MaskedIp::new(ip_b, 64));
     655            1 :         assert_ne!(MaskedIp::new(ip_a, 32), MaskedIp::new(ip_b, 32));
     656            1 :         assert_eq!(MaskedIp::new(ip_a, 30), MaskedIp::new(ip_b, 30));
     657            1 :         assert_eq!(MaskedIp::new(ip_c, 30), MaskedIp::new(ip_d, 30));
     658              : 
     659            1 :         assert_ne!(MaskedIp::new(ip_e, 128), MaskedIp::new(ip_f, 128));
     660            1 :         assert_eq!(MaskedIp::new(ip_e, 64), MaskedIp::new(ip_f, 64));
     661            1 :     }
     662              : 
     663              :     #[test]
     664            1 :     fn test_default_auth_rate_limit_set() {
     665            1 :         // these values used to exceed u32::MAX
     666            1 :         assert_eq!(
     667            1 :             RateBucketInfo::DEFAULT_AUTH_SET,
     668            1 :             [
     669            1 :                 RateBucketInfo {
     670            1 :                     interval: Duration::from_secs(1),
     671            1 :                     max_rpi: 1000 * 4096,
     672            1 :                 },
     673            1 :                 RateBucketInfo {
     674            1 :                     interval: Duration::from_secs(60),
     675            1 :                     max_rpi: 600 * 4096 * 60,
     676            1 :                 },
     677            1 :                 RateBucketInfo {
     678            1 :                     interval: Duration::from_secs(600),
     679            1 :                     max_rpi: 300 * 4096 * 600,
     680            1 :                 }
     681            1 :             ]
     682            1 :         );
     683              : 
     684            4 :         for x in RateBucketInfo::DEFAULT_AUTH_SET {
     685            3 :             let y = x.to_string().parse().unwrap();
     686            3 :             assert_eq!(x, y);
     687              :         }
     688            1 :     }
     689              : 
     690              :     #[tokio::test]
     691            1 :     async fn auth_quirks_scram() {
     692            1 :         let (mut client, server) = tokio::io::duplex(1024);
     693            1 :         let mut stream = PqStream::new(Stream::from_raw(server));
     694            1 : 
     695            1 :         let ctx = RequestContext::test();
     696            1 :         let api = Auth {
     697            1 :             ips: vec![],
     698            1 :             vpc_endpoint_ids: vec![],
     699            1 :             access_blocker_flags: AccessBlockerFlags::default(),
     700            1 :             secret: AuthSecret::Scram(ServerSecret::build("my-secret-password").await.unwrap()),
     701            1 :         };
     702            1 : 
     703            1 :         let user_info = ComputeUserInfoMaybeEndpoint {
     704            1 :             user: "conrad".into(),
     705            1 :             endpoint_id: Some("endpoint".into()),
     706            1 :             options: NeonOptions::default(),
     707            1 :         };
     708            1 : 
     709            1 :         let handle = tokio::spawn(async move {
     710            1 :             let mut scram = ScramSha256::new(b"my-secret-password", ChannelBinding::unsupported());
     711            1 : 
     712            1 :             let mut read = BytesMut::new();
     713            1 : 
     714            1 :             // server should offer scram
     715            1 :             match read_message(&mut client, &mut read).await {
     716            1 :                 PgMessage::AuthenticationSasl(a) => {
     717            1 :                     let options: Vec<&str> = a.mechanisms().collect().unwrap();
     718            1 :                     assert_eq!(options, ["SCRAM-SHA-256"]);
     719            1 :                 }
     720            1 :                 _ => panic!("wrong message"),
     721            1 :             }
     722            1 : 
     723            1 :             // client sends client-first-message
     724            1 :             let mut write = BytesMut::new();
     725            1 :             frontend::sasl_initial_response("SCRAM-SHA-256", scram.message(), &mut write).unwrap();
     726            1 :             client.write_all(&write).await.unwrap();
     727            1 : 
     728            1 :             // server response with server-first-message
     729            1 :             match read_message(&mut client, &mut read).await {
     730            1 :                 PgMessage::AuthenticationSaslContinue(a) => {
     731            1 :                     scram.update(a.data()).await.unwrap();
     732            1 :                 }
     733            1 :                 _ => panic!("wrong message"),
     734            1 :             }
     735            1 : 
     736            1 :             // client response with client-final-message
     737            1 :             write.clear();
     738            1 :             frontend::sasl_response(scram.message(), &mut write).unwrap();
     739            1 :             client.write_all(&write).await.unwrap();
     740            1 : 
     741            1 :             // server response with server-final-message
     742            1 :             match read_message(&mut client, &mut read).await {
     743            1 :                 PgMessage::AuthenticationSaslFinal(a) => {
     744            1 :                     scram.finish(a.data()).unwrap();
     745            1 :                 }
     746            1 :                 _ => panic!("wrong message"),
     747            1 :             }
     748            1 :         });
     749            1 :         let endpoint_rate_limiter = Arc::new(EndpointRateLimiter::new_with_shards(
     750            1 :             EndpointRateLimiter::DEFAULT,
     751            1 :             64,
     752            1 :         ));
     753            1 : 
     754            1 :         let _creds = auth_quirks(
     755            1 :             &ctx,
     756            1 :             &api,
     757            1 :             user_info,
     758            1 :             &mut stream,
     759            1 :             false,
     760            1 :             &CONFIG,
     761            1 :             endpoint_rate_limiter,
     762            1 :         )
     763            1 :         .await
     764            1 :         .unwrap();
     765            1 : 
     766            1 :         // flush the final server message
     767            1 :         stream.flush().await.unwrap();
     768            1 : 
     769            1 :         handle.await.unwrap();
     770            1 :     }
     771              : 
     772              :     #[tokio::test]
     773            1 :     async fn auth_quirks_cleartext() {
     774            1 :         let (mut client, server) = tokio::io::duplex(1024);
     775            1 :         let mut stream = PqStream::new(Stream::from_raw(server));
     776            1 : 
     777            1 :         let ctx = RequestContext::test();
     778            1 :         let api = Auth {
     779            1 :             ips: vec![],
     780            1 :             vpc_endpoint_ids: vec![],
     781            1 :             access_blocker_flags: AccessBlockerFlags::default(),
     782            1 :             secret: AuthSecret::Scram(ServerSecret::build("my-secret-password").await.unwrap()),
     783            1 :         };
     784            1 : 
     785            1 :         let user_info = ComputeUserInfoMaybeEndpoint {
     786            1 :             user: "conrad".into(),
     787            1 :             endpoint_id: Some("endpoint".into()),
     788            1 :             options: NeonOptions::default(),
     789            1 :         };
     790            1 : 
     791            1 :         let handle = tokio::spawn(async move {
     792            1 :             let mut read = BytesMut::new();
     793            1 :             let mut write = BytesMut::new();
     794            1 : 
     795            1 :             // server should offer cleartext
     796            1 :             match read_message(&mut client, &mut read).await {
     797            1 :                 PgMessage::AuthenticationCleartextPassword => {}
     798            1 :                 _ => panic!("wrong message"),
     799            1 :             }
     800            1 : 
     801            1 :             // client responds with password
     802            1 :             write.clear();
     803            1 :             frontend::password_message(b"my-secret-password", &mut write).unwrap();
     804            1 :             client.write_all(&write).await.unwrap();
     805            1 :         });
     806            1 :         let endpoint_rate_limiter = Arc::new(EndpointRateLimiter::new_with_shards(
     807            1 :             EndpointRateLimiter::DEFAULT,
     808            1 :             64,
     809            1 :         ));
     810            1 : 
     811            1 :         let _creds = auth_quirks(
     812            1 :             &ctx,
     813            1 :             &api,
     814            1 :             user_info,
     815            1 :             &mut stream,
     816            1 :             true,
     817            1 :             &CONFIG,
     818            1 :             endpoint_rate_limiter,
     819            1 :         )
     820            1 :         .await
     821            1 :         .unwrap();
     822            1 : 
     823            1 :         handle.await.unwrap();
     824            1 :     }
     825              : 
     826              :     #[tokio::test]
     827            1 :     async fn auth_quirks_password_hack() {
     828            1 :         let (mut client, server) = tokio::io::duplex(1024);
     829            1 :         let mut stream = PqStream::new(Stream::from_raw(server));
     830            1 : 
     831            1 :         let ctx = RequestContext::test();
     832            1 :         let api = Auth {
     833            1 :             ips: vec![],
     834            1 :             vpc_endpoint_ids: vec![],
     835            1 :             access_blocker_flags: AccessBlockerFlags::default(),
     836            1 :             secret: AuthSecret::Scram(ServerSecret::build("my-secret-password").await.unwrap()),
     837            1 :         };
     838            1 : 
     839            1 :         let user_info = ComputeUserInfoMaybeEndpoint {
     840            1 :             user: "conrad".into(),
     841            1 :             endpoint_id: None,
     842            1 :             options: NeonOptions::default(),
     843            1 :         };
     844            1 : 
     845            1 :         let handle = tokio::spawn(async move {
     846            1 :             let mut read = BytesMut::new();
     847            1 : 
     848            1 :             // server should offer cleartext
     849            1 :             match read_message(&mut client, &mut read).await {
     850            1 :                 PgMessage::AuthenticationCleartextPassword => {}
     851            1 :                 _ => panic!("wrong message"),
     852            1 :             }
     853            1 : 
     854            1 :             // client responds with password
     855            1 :             let mut write = BytesMut::new();
     856            1 :             frontend::password_message(b"endpoint=my-endpoint;my-secret-password", &mut write)
     857            1 :                 .unwrap();
     858            1 :             client.write_all(&write).await.unwrap();
     859            1 :         });
     860            1 : 
     861            1 :         let endpoint_rate_limiter = Arc::new(EndpointRateLimiter::new_with_shards(
     862            1 :             EndpointRateLimiter::DEFAULT,
     863            1 :             64,
     864            1 :         ));
     865            1 : 
     866            1 :         let creds = auth_quirks(
     867            1 :             &ctx,
     868            1 :             &api,
     869            1 :             user_info,
     870            1 :             &mut stream,
     871            1 :             true,
     872            1 :             &CONFIG,
     873            1 :             endpoint_rate_limiter,
     874            1 :         )
     875            1 :         .await
     876            1 :         .unwrap();
     877            1 : 
     878            1 :         assert_eq!(creds.0.info.endpoint, "my-endpoint");
     879            1 : 
     880            1 :         handle.await.unwrap();
     881            1 :     }
     882              : }
        

Generated by: LCOV version 2.1-beta