CLO-117: reject AWS PrivateLink SERVICE NAME that is not a service name

doc/user/data/examples/create_connection.yml

@@ -656,7 +656,8 @@
       description: |
         *Value:* `text`. Required.
 
-        The name of the AWS PrivateLink service.
+        The name of the AWS VPC endpoint service, which always starts with
+        `com.amazonaws.`.
     - name: "`AVAILABILITY ZONES`"
       description: |
         *Value:* `text[]`. Required.

src/adapter/src/coord/sequencer/inner.rs

@@ -60,7 +60,7 @@ use mz_sql::names::{
     SchemaSpecifier, SystemObjectId,
 };
 use mz_sql::plan::{
-    AlterMaterializedViewApplyReplacementPlan, ConnectionDetails, NetworkPolicyRule,
+    AlterMaterializedViewApplyReplacementPlan, ConnectionDetails, NetworkPolicyRule, PlanError,
     StatementContext,
 };
 use mz_sql::pure::{PurifiedSourceExport, generate_subsource_statements};
@@ -88,6 +88,7 @@ use mz_sql_parser::ast::{
 use mz_ssh_util::keys::SshKeyPairSet;
 use mz_storage_client::controller::ExportDescription;
 use mz_storage_types::AlterCompatible;
+use mz_storage_types::connections::AwsPrivatelinkConnection;
 use mz_storage_types::connections::inline::IntoInlineConnection;
 use mz_storage_types::controller::StorageError;
 use mz_storage_types::sources::SourceConnection;
@@ -173,6 +174,18 @@ where
     }
 }
 
+/// Rejects connection options whose values cannot work, independent of any
+/// external system.
+fn check_connection_details(details: &ConnectionDetails) -> Result<(), PlanError> {
+    match details {
+        ConnectionDetails::AwsPrivatelink(privatelink) => {
+            AwsPrivatelinkConnection::check_service_name(&privatelink.service_name)
+                .map_err(PlanError::InvalidPrivatelinkServiceName)
+        }
+        _ => Ok(()),
+    }
+}
+
 struct DropOps {
     ops: Vec<catalog::Op>,
     dropped_active_db: bool,
@@ -710,6 +723,10 @@ impl Coordinator {
         plan: plan::CreateConnectionPlan,
         resolved_ids: ResolvedIds,
     ) {
+        if let Err(err) = check_connection_details(&plan.connection.details) {
+            return ctx.retire(Err(AdapterError::PlanError(err)));
+        }
+
         let (connection_id, connection_gid) = match self.allocate_user_id().await {
             Ok(item_id) => item_id,
             Err(err) => return ctx.retire(Err(err)),
@@ -3756,6 +3773,12 @@ impl Coordinator {
             }
         };
 
+        // `conn` is the whole re-planned connection, so this also rejects a
+        // stored value the statement did not touch.
+        if let Err(err) = check_connection_details(&conn.details) {
+            return ctx.retire(Err(AdapterError::InvalidAlter("CONNECTION", err)));
+        }
+
         // Inspect guarded secrets whether or not validation was requested,
         // before the altered connection is installed in the catalog.
         if let Err(err) = self

src/sql/src/plan/error.rs

@@ -31,6 +31,7 @@ use mz_sql_parser::ast::display::AstDisplay;
 use mz_sql_parser::ast::{IdentError, UnresolvedItemName};
 use mz_sql_parser::parser::{ParserError, ParserStatementError};
 use mz_sql_server_util::SqlServerError;
+use mz_storage_types::connections::InvalidAwsPrivatelinkServiceName;
 use mz_storage_types::sources::ExternalReferenceResolutionError;
 
 use crate::catalog::{
@@ -237,6 +238,7 @@ pub enum PlanError {
         name: String,
         supported_azs: BTreeSet<String>,
     },
+    InvalidPrivatelinkServiceName(InvalidAwsPrivatelinkServiceName),
     DuplicatePrivatelinkAvailabilityZone {
         duplicate_azs: BTreeSet<String>,
     },
@@ -457,6 +459,7 @@ impl PlanError {
                 let supported_azs_str = supported_azs.iter().join("\n  ");
                 Some(format!("Did you supply an availability zone name instead of an ID? Known availability zone IDs:\n  {}", supported_azs_str))
             }
+            Self::InvalidPrivatelinkServiceName(err) => Some(err.hint()),
             Self::DuplicatePrivatelinkAvailabilityZone { duplicate_azs, ..} => {
                 let duplicate_azs  = duplicate_azs.iter().join("\n  ");
                 Some(format!("Duplicated availability zones:\n  {}", duplicate_azs))
@@ -762,6 +765,7 @@ impl fmt::Display for PlanError {
                 })
             },
             Self::InvalidPrivatelinkAvailabilityZone { name, ..} => write!(f, "invalid AWS PrivateLink availability zone {}", name.quoted()),
+            Self::InvalidPrivatelinkServiceName(err) => err.fmt(f),
             Self::DuplicatePrivatelinkAvailabilityZone {..} =>   write!(f, "connection cannot contain duplicate availability zones"),
             Self::InvalidSchemaName => write!(f, "no valid schema selected"),
             Self::ItemAlreadyExists { name, item_type } => write!(f, "{item_type} {} already exists", name.quoted()),

src/storage-types/src/connections.rs

@@ -11,6 +11,7 @@
 
 use std::borrow::Cow;
 use std::collections::{BTreeMap, BTreeSet};
+use std::fmt;
 use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::SystemTime;
@@ -50,6 +51,7 @@ use mz_ore::error::ErrorExt;
 use mz_ore::future::{InTask, OreFutureExt};
 use mz_ore::netio::resolve_address;
 use mz_ore::num::NonNeg;
+use mz_ore::str::StrExt;
 use mz_repr::{CatalogItemId, GlobalId};
 use mz_secrets::SecretsReader;
 use mz_sql_parser::ast::ConnectionRulePattern;
@@ -543,6 +545,8 @@ pub enum ConnectionValidationError {
     Aws(#[from] AwsConnectionValidationError),
     #[error(transparent)]
     Gcp(#[from] gcp::GcpConnectionValidationError),
+    #[error(transparent)]
+    AwsPrivatelinkServiceName(#[from] InvalidAwsPrivatelinkServiceName),
     #[error("{}", .0.display_with_causes())]
     Other(#[from] anyhow::Error),
 }
@@ -556,6 +560,7 @@ impl ConnectionValidationError {
             ConnectionValidationError::SqlServer(e) => e.detail(),
             ConnectionValidationError::Aws(e) => e.detail(),
             ConnectionValidationError::Gcp(e) => e.detail(),
+            ConnectionValidationError::AwsPrivatelinkServiceName(_) => None,
             ConnectionValidationError::Other(_) => None,
         }
     }
@@ -568,6 +573,7 @@ impl ConnectionValidationError {
             ConnectionValidationError::SqlServer(e) => e.hint(),
             ConnectionValidationError::Aws(e) => e.hint(),
             ConnectionValidationError::Gcp(e) => e.hint(),
+            ConnectionValidationError::AwsPrivatelinkServiceName(e) => Some(e.hint()),
             ConnectionValidationError::Other(_) => None,
         }
     }
@@ -1014,6 +1020,51 @@ impl AlterCompatible for AwsPrivatelinkConnection {
     }
 }
 
+/// A `SERVICE NAME` that cannot name an AWS VPC endpoint service.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct InvalidAwsPrivatelinkServiceName {
+    pub name: String,
+}
+
+impl fmt::Display for InvalidAwsPrivatelinkServiceName {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "invalid AWS PrivateLink service name {}",
+            self.name.quoted()
+        )
+    }
+}
+
+impl std::error::Error for InvalidAwsPrivatelinkServiceName {}
+
+impl InvalidAwsPrivatelinkServiceName {
+    /// Explains how to find the right value.
+    pub fn hint(&self) -> String {
+        "SERVICE NAME must name an AWS VPC endpoint service, for example \
+         `com.amazonaws.vpce.us-east-1.vpce-svc-0e123abc123198abc`. Endpoint service names are \
+         listed in the AWS console under VPC > Endpoint services."
+            .into()
+    }
+}
+
+impl AwsPrivatelinkConnection {
+    /// Checks that `service_name` could name an AWS VPC endpoint service.
+    ///
+    /// Every endpoint service name starts with `com.amazonaws.`, whether the
+    /// service is customer-owned (`com.amazonaws.vpce.<region>.vpce-svc-<id>`)
+    /// or AWS-managed (`com.amazonaws.<region>.<service>`). Only the prefix is
+    /// checked, so a name AWS would accept is never rejected here.
+    pub fn check_service_name(service_name: &str) -> Result<(), InvalidAwsPrivatelinkServiceName> {
+        if service_name.starts_with("com.amazonaws.") {
+            return Ok(());
+        }
+        Err(InvalidAwsPrivatelinkServiceName {
+            name: service_name.to_string(),
+        })
+    }
+}
+
 #[derive(Clone, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)]
 pub struct KafkaTlsConfig {
     pub identity: Option<TlsIdentity>,
@@ -3213,12 +3264,16 @@ impl AwsPrivatelinkConnection {
         &self,
         id: CatalogItemId,
         storage_configuration: &StorageConfiguration,
-    ) -> Result<(), anyhow::Error> {
+    ) -> Result<(), ConnectionValidationError> {
+        // An endpoint for an unusable service name reports a misleading
+        // condition (missing availability zones), so check the name first.
+        Self::check_service_name(&self.service_name)?;
+
         let Some(ref cloud_resource_reader) = storage_configuration
             .connection_context
             .cloud_resource_reader
         else {
-            return Err(anyhow!("AWS PrivateLink connections are unsupported"));
+            return Err(anyhow!("AWS PrivateLink connections are unsupported").into());
         };
 
         // No need to optionally run this in a task, as we are just validating from envd.
@@ -3231,12 +3286,47 @@ impl AwsPrivatelinkConnection {
 
         match availability {
             Some(condition) if condition.status == "True" => Ok(()),
-            Some(condition) => Err(anyhow!("{}", condition.message)),
-            None => Err(anyhow!("Endpoint availability is unknown")),
+            Some(condition) => Err(anyhow!("{}", condition.message).into()),
+            None => Err(anyhow!("Endpoint availability is unknown").into()),
         }
     }
 
     fn validate_by_default(&self) -> bool {
         false
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[mz_ore::test]
+    fn test_check_service_name() {
+        // Customer-owned and AWS-managed endpoint services are both accepted,
+        // as is anything else that could be an endpoint service name.
+        for name in [
+            "com.amazonaws.vpce.us-east-1.vpce-svc-0e123abc123198abc",
+            "com.amazonaws.vpce.test.vpce-svc-e2e-test",
+            "com.amazonaws.us-east-1.s3",
+            "com.amazonaws.anything",
+        ] {
+            assert_eq!(
+                AwsPrivatelinkConnection::check_service_name(name),
+                Ok(()),
+                "expected {name} to be accepted"
+            );
+        }
+
+        for name in [
+            "",
+            "com.amazonaws",
+            "vpce-svc-0e123abc123198abc",
+            "my-db-lb-0123456789abcdef.elb.eu-central-1.amazonaws.com",
+            "db.internal.example.org",
+        ] {
+            let err = AwsPrivatelinkConnection::check_service_name(name)
+                .expect_err("service name should be rejected");
+            assert_eq!(err.name, name);
+        }
+    }
+}

test/testdrive/connection-create-drop.td

@@ -532,6 +532,9 @@ contains: BROKERS must contain at least one static broker address
 
 > DROP CONNECTION kafka_matching_ok
 
+! ALTER CONNECTION pl_for_matching SET (SERVICE NAME 'my-db-lb-0123456789abcdef.elb.eu-central-1.amazonaws.com')
+contains: invalid AWS PrivateLink service name "my-db-lb-0123456789abcdef.elb.eu-central-1.amazonaws.com"
+
 $ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
 ALTER SYSTEM RESET enable_kafka_broker_matching_rules;
 ALTER SYSTEM RESET max_aws_privatelink_connections;
@@ -549,3 +552,17 @@ contains: unknown catalog item 'foo'
     AVAILABILITY ZONES ('use1-az1', 'use1-az2')
   )
 contains: AWS PrivateLink
+
+# SERVICE NAME must name a VPC endpoint service, not the DNS name of the target.
+! CREATE CONNECTION privatelink_elb_hostname TO AWS PRIVATELINK (
+    SERVICE NAME 'my-db-lb-0123456789abcdef.elb.eu-central-1.amazonaws.com',
+    AVAILABILITY ZONES ()
+  )
+contains: invalid AWS PrivateLink service name "my-db-lb-0123456789abcdef.elb.eu-central-1.amazonaws.com"
+hint: SERVICE NAME must name an AWS VPC endpoint service
+
+! CREATE CONNECTION privatelink_not_a_service_name TO AWS PRIVATELINK (
+    SERVICE NAME 'vpce-svc-0e123abc123198abc',
+    AVAILABILITY ZONES ()
+  )
+contains: invalid AWS PrivateLink service name "vpce-svc-0e123abc123198abc"