Add better RDS UX

CMakeLists.txt

@@ -22,6 +22,7 @@ endif()
 add_compile_definitions(DUCKDB_AWS_GIT_SHA="${DUCKDB_AWS_GIT_SHA}")
 
 set(EXTENSION_SOURCES src/aws_extension.cpp src/aws_secret.cpp src/cloudformation_functions.cpp
+                      src/rds/rds_utils.cpp src/rds/rds_storage.cpp
                       src/quack_on_ec2_resource.cpp
                       src/redshift/redshift_utils.cpp src/redshift/redshift_storage.cpp
                       src/utils/utils.cpp)

extension_config.cmake

@@ -10,10 +10,9 @@ duckdb_extension_load(aws
 # Currently Disabled since this will not build on CI. To get CI working you (most likely)
 # need to copy the vcpkg_ports/libpq in the duckdb/duckdb-postgres repo. 
 # For now commented out so unblock development
-# duckdb_extension_load(postgres_scanner
+#duckdb_extension_load(postgres_scanner
 #    DONT_LINK
 #    GIT_URL https://github.com/duckdb/duckdb-postgres
-#    GIT_TAG f77b0cb511748fd70fb8a4eb265e2990599d286c
+#    GIT_TAG a4e03aad76a002e913e676cce1fc2600f64a614f
 #    SUBMODULES database-connector
-#    APPLY_PATCHES
 #)

src/aws_extension.cpp

@@ -1,6 +1,7 @@
 #include "aws_secret.hpp"
 #include "aws_extension.hpp"
 #include "cloudformation_functions.hpp"
+#include "rds/rds_utils.hpp"
 #include "quack_on_ec2_resource.hpp"
 #include "redshift/redshift_utils.hpp"
 
@@ -26,6 +27,9 @@ static void LoadInternal(ExtensionLoader &loader) {
 	// Makes `ATTACH '<cluster-id>' (TYPE redshift, ...)` resolve to the redshift storage extension.
 	Redshift::RegisterStorageExtension(loader);
 
+	// Same for `ATTACH '<db-cluster-id>' (TYPE rds, ...)`.
+	Rds::RegisterStorageExtension(loader);
+
 	CloudFormationFunctions::Register(loader);
 	QuackOnEc2Resource::Register(loader);
 }

src/include/rds/rds_utils.hpp

@@ -0,0 +1,53 @@
+#pragma once
+
+#include "duckdb.hpp"
+#include "duckdb/main/secret/secret.hpp"
+
+#include <aws/core/auth/AWSCredentialsProvider.h>
+#include <memory>
+
+namespace duckdb {
+
+class ExtensionLoader;
+
+//! The subset of an RDS DB instance description that we need to open a connection to it. Filled in
+//! from the `DescribeDBInstances` API, i.e. what `aws rds describe-db-instances
+//! --db-instance-identifier <id>` reports.
+struct RdsInstanceInfo {
+	string endpoint_address;
+	int32_t endpoint_port = 0;
+	//! The engine the instance runs, e.g. 'postgres', 'aurora-postgresql', 'mysql' or 'oracle-se2'.
+	//! Only the Postgres-flavored ones can be attached; see RdsEngineIsPostgres.
+	string engine;
+	string engine_version;
+	//! The instance's initial database, which may be unset - RDS does not require one.
+	string db_name;
+	string master_username;
+	string status;
+	bool iam_auth_enabled = false;
+};
+
+//! Whether an RDS engine speaks the Postgres wire protocol, i.e. whether the postgres extension
+//! can connect to it. Covers both 'postgres' (RDS for PostgreSQL) and 'aurora-postgresql'.
+bool RdsEngineIsPostgres(const string &engine);
+
+struct Rds {
+	//! Register the 'rds' storage extension, which is what makes `ATTACH '<instance-id>' (TYPE rds)`
+	//! resolve here.
+	static void RegisterStorageExtension(ExtensionLoader &loader);
+
+	//! Look up a DB instance by identifier via the RDS `DescribeDBInstances` API. Throws when the
+	//! instance does not exist or exposes no endpoint (e.g. while it is still being created).
+	static RdsInstanceInfo DescribeDBInstance(const std::shared_ptr<Aws::Auth::AWSCredentialsProvider> &provider,
+	                                          const string &instance_id, const string &region);
+
+	//! Mint a short-lived (15 minute) IAM authentication token for `db_user` on the given endpoint,
+	//! to be used as the database password. Unlike Redshift's GetClusterCredentialsWithIAM this is
+	//! not an API call: the token is a request to the endpoint presigned with the AWS identity the
+	//! provider holds, so the database user must already exist and be granted `rds_iam`.
+	//! See https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/UsingWithRDS.IAMDBAuth.html
+	static string GenerateAuthToken(const std::shared_ptr<Aws::Auth::AWSCredentialsProvider> &provider,
+	                                const string &host, int32_t port, const string &region, const string &db_user);
+};
+
+} // namespace duckdb

src/include/utils/utils.hpp

@@ -4,6 +4,9 @@
 #include "duckdb/main/secret/secret_manager.hpp"
 #include "duckdb/storage/storage_extension.hpp"
 
+#include <aws/core/auth/AWSCredentialsProvider.h>
+#include <memory>
+
 namespace duckdb {
 
 //! Resolve the AWS region to use, taking the first of these that yields a value:
@@ -30,11 +33,35 @@ bool IsAwsSecretType(const string &type);
 //! that is present. Throws rather than returning null.
 unique_ptr<SecretEntry> FindAwsSecret(ClientContext &context, const string &secret_name);
 
+//! Read a string field out of a key-value secret, "" when the field is not set.
+string GetSecretString(const KeyValueSecret &secret, const string &key);
+
+//! A credentials provider for the AWS identity the secret resolved to. `service` only names the
+//! caller in SDK allocation tags. Throws when the secret carries no credentials.
+std::shared_ptr<Aws::Auth::AWSCredentialsProvider> CredentialsProviderFromSecret(const KeyValueSecret &secret,
+                                                                                 const string &service);
+
+//! Quote a value for a libpq connection string. Backslashes and single quotes must be escaped
+//! with a backslash; quoting the whole value keeps empty values and embedded spaces valid.
+string EscapeConnectionValue(const string &value);
+
 //! The name of the postgres extension, for autoloading it.
 extern const char *const POSTGRES_EXTENSION_NAME;
 
-//! The postgres extension's storage extension, or null when it is not loaded. Redshift speaks the
-//! Postgres wire protocol, so attaching one hands off to postgres once the endpoint is resolved.
+//! The postgres extension's storage extension, or null when it is not loaded. Redshift and RDS's
+//! postgres engines speak the Postgres wire protocol, so attaching one hands off to postgres once
+//! the endpoint is resolved.
 optional_ptr<StorageExtension> FindPostgresStorageExtension(const DBConfig &config);
 
+//! The postgres storage extension, autoloading the postgres extension if it is not loaded yet.
+//! `attach_type` only names the caller in the error thrown when it cannot be loaded.
+optional_ptr<StorageExtension> RequirePostgresStorageExtension(ClientContext &context, const string &attach_type);
+
+//! The message to report for a failed postgres attach. Postgres quotes the entire connection
+//! string back, which buries the server's own diagnostic and puts `password` - for Redshift and
+//! RDS a live, AWS-generated credential - in the error and in whatever log it lands in. Returns
+//! the server's message alone when the connection got far enough to produce one, and otherwise
+//! the full message with the password redacted.
+string PostgresAttachErrorMessage(std::exception &ex, const string &password);
+
 } // namespace duckdb

src/rds/rds_storage.cpp

@@ -0,0 +1,222 @@
+#include "rds/rds_utils.hpp"
+
+#include "aws_client.hpp"
+#include "utils/utils.hpp"
+
+#include "duckdb/catalog/catalog.hpp"
+#include "duckdb/common/exception.hpp"
+#include "duckdb/common/string_util.hpp"
+#include "duckdb/main/attached_database.hpp"
+#include "duckdb/main/database.hpp"
+#include "duckdb/main/extension/extension_loader.hpp"
+#include "duckdb/main/secret/secret_manager.hpp"
+#include "duckdb/main/settings.hpp"
+#include "duckdb/parser/parsed_data/attach_info.hpp"
+#include "duckdb/storage/storage_extension.hpp"
+#include "duckdb/transaction/transaction_manager.hpp"
+
+namespace duckdb {
+
+namespace {
+
+//! ATTACH options that we consume ourselves. Everything else is forwarded to the postgres
+//! extension, which rejects options it does not know.
+struct RdsAttachOptions {
+	string secret_name;
+	string region;
+	string db_name;
+	string host;
+	string port;
+	string user;
+};
+
+RdsAttachOptions ParseAttachOptions(AttachOptions &options) {
+	RdsAttachOptions parsed;
+	// Consumed options are erased, since postgres throws on any option it does not recognize.
+	for (auto it = options.options.begin(); it != options.options.end();) {
+		auto key = StringUtil::Lower(it->first);
+		auto value = it->second.ToString();
+		if (key == "secret") {
+			// Left in place: postgres looks the secret up by name, and it must find one. See the
+			// note in RdsAttach.
+			parsed.secret_name = value;
+			it++;
+			continue;
+		}
+		if (key == "region") {
+			parsed.region = value;
+		} else if (key == "database" || key == "dbname") {
+			parsed.db_name = value;
+		} else if (key == "host") {
+			parsed.host = value;
+		} else if (key == "port") {
+			parsed.port = value;
+		} else if (key == "user") {
+			parsed.user = value;
+		} else {
+			it++;
+			continue;
+		}
+		it = options.options.erase(it);
+	}
+	return parsed;
+}
+
+//! `ATTACH '<db-instance-id>' AS db (TYPE rds, SECRET <aws-or-s3-secret>)`.
+//!
+//! The DB instance identifier is all the user gives us, so we ask RDS for the rest:
+//! DescribeDBInstances - the API behind `aws rds describe-db-instances --db-instance-identifier` -
+//! supplies the endpoint host/port, the engine and the default database name, and an IAM auth
+//! token signed with the AWS identity behind the secret serves as the database password. Those are
+//! assembled into a libpq connection string and handed to the postgres extension's own attach.
+//!
+//! Only the Postgres-flavored engines can be attached this way; every other engine (MySQL,
+//! MariaDB, Oracle, ...) is rejected up front.
+unique_ptr<Catalog> RdsAttach(optional_ptr<StorageExtensionInfo> storage_info, ClientContext &context,
+                              AttachedDatabase &db, const string &name, AttachInfo &info, AttachOptions &options) {
+	if (!Settings::Get<EnableExternalAccessSetting>(context)) {
+		throw PermissionException("Attaching RDS databases is disabled through configuration");
+	}
+
+	auto instance_id = info.path;
+	if (instance_id.empty()) {
+		throw BinderException("No RDS DB instance identifier given. Pass it as the ATTACH path, e.g. "
+		                      "ATTACH '<db-instance-id>' AS db (TYPE rds)");
+	}
+
+	auto attach_options = ParseAttachOptions(options);
+	auto secret_entry = FindAwsSecret(context, attach_options.secret_name);
+	const auto &secret = dynamic_cast<const KeyValueSecret &>(*secret_entry->secret);
+
+	// An s3 secret's region is the bucket region, which need not be the instance's, so an explicit
+	// ATTACH region wins over it. Past those two, fall back to the sources CREATE SECRET uses.
+	auto explicit_region = attach_options.region.empty() ? GetSecretString(secret, "region") : attach_options.region;
+	auto region = ResolveAwsRegion(context, explicit_region, "");
+	if (region.empty()) {
+		throw InvalidConfigurationException("No AWS region found for the RDS instance. Pass it to ATTACH, e.g. "
+		                                    "ATTACH '<db-instance-id>' AS db (TYPE rds, REGION '<region>'), set it on "
+		                                    "the secret, or configure the AWS_REGION environment variable");
+	}
+
+	auto provider = CredentialsProviderFromSecret(secret, "RDS");
+	auto instance = Rds::DescribeDBInstance(provider, instance_id, region);
+
+	// RDS runs engines DuckDB cannot speak to; only the Postgres ones have an extension to hand off
+	// to. MySQL/MariaDB instances would need the mysql extension, which does not take a connection
+	// string in the shape we build below.
+	if (!RdsEngineIsPostgres(instance.engine)) {
+		throw NotImplementedException("RDS instance '%s' runs the '%s' engine, which cannot be attached. Only the "
+		                              "Postgres engines ('postgres', 'aurora-postgresql') are supported",
+		                              instance_id, instance.engine);
+	}
+
+	// Resolved after the engine check, so an unsupported engine does not first demand postgres.
+	auto postgres_extension = RequirePostgresStorageExtension(context, "an RDS instance");
+
+	// Anything ATTACH pins explicitly wins over what the instance reports.
+	auto host = attach_options.host.empty() ? instance.endpoint_address : attach_options.host;
+	auto db_name = attach_options.db_name.empty() ? instance.db_name : attach_options.db_name;
+	if (db_name.empty()) {
+		// An instance created without an initial database reports none, and libpq then defaults
+		// dbname to the *user* name, which is never a database that exists here. Every Postgres
+		// engine provisions 'postgres', and the engine check above has established this is one.
+		db_name = "postgres";
+	}
+
+	auto port = instance.endpoint_port;
+	if (!attach_options.port.empty()) {
+		try {
+			port = std::stoi(attach_options.port);
+		} catch (const std::exception &) {
+			throw InvalidInputException("'PORT' must be an integer, got '%s'", attach_options.port);
+		}
+	}
+
+	// Unlike Redshift, RDS does not mint the user along with the password: the IAM token is issued
+	// *for* a database user that must already exist. The instance's master user is the only one we
+	// can guess at, so anything else has to be named explicitly.
+	auto user = attach_options.user.empty() ? instance.master_username : attach_options.user;
+	if (user.empty()) {
+		throw InvalidConfigurationException("No database user to connect to RDS instance '%s' as, and the instance "
+		                                    "reports no master username. Pass one to ATTACH, e.g. ATTACH '%s' AS db "
+		                                    "(TYPE rds, USER '<db-user>')",
+		                                    instance_id, instance_id);
+	}
+
+	// The token is only accepted when the instance has IAM database authentication turned on, and
+	// the failure it produces otherwise is an opaque password-authentication error.
+	if (!instance.iam_auth_enabled) {
+		throw InvalidConfigurationException(
+		    "RDS instance '%s' does not have IAM database authentication enabled, so its AWS identity cannot be used "
+		    "to log in. Enable it on the instance (`aws rds modify-db-instance --db-instance-identifier %s "
+		    "--enable-iam-database-authentication`) and grant the 'rds_iam' role to database user '%s'",
+		    instance_id, instance_id, user);
+	}
+
+	// The token is signed for a specific host/port/user triple, so it has to be minted from the
+	// endpoint we are about to connect to rather than the one the instance reported.
+	auto token = Rds::GenerateAuthToken(provider, host, port, region, user);
+
+	// RDS rejects an IAM login over an unencrypted connection.
+	string connection_string = "host=" + EscapeConnectionValue(host) +
+	                           " port=" + EscapeConnectionValue(to_string(port)) +
+	                           " user=" + EscapeConnectionValue(user) + " password=" + EscapeConnectionValue(token) +
+	                           " sslmode='require'" + " dbname=" + EscapeConnectionValue(db_name);
+
+	// Hand the postgres extension a plain connection string as the attach path.
+	info.path = connection_string;
+
+	// Postgres must be given a secret name it can resolve: with none it falls back to the implicit
+	// '__default_postgres' secret, which it probes in the 'local_file' storage - and that throws
+	// outright when persistent secrets are disabled. Naming the aws/s3 secret we just used is safe,
+	// because postgres only harvests libpq option names (host, port, user, ...) from a secret and
+	// an aws/s3 secret holds none of them.
+	options.options["secret"] = Value(secret.GetName().GetIdentifierName());
+
+	try {
+		return postgres_extension->attach(postgres_extension->storage_info.get(), context, db, name, info, options);
+	} catch (std::exception &ex) {
+		auto message = PostgresAttachErrorMessage(ex, token);
+		auto error_message = StringUtil::Format("Unable to connect to RDS instance '%s': %s", instance_id, message);
+		if (StringUtil::Contains(message, "password authentication failed")) {
+			// RDS hands the token to Postgres as an ordinary password, so a database user that has
+			// not been switched over to IAM authentication rejects it as one - and says nothing
+			// about IAM, because as far as it knows a password simply did not match.
+			error_message += StringUtil::Format("\nCheck that user '%s' exists on the instance and has rds_iam "
+			                                    "privileges (`GRANT rds_iam TO \"%s\";`, run as the master user)."
+			                                    "\nThe AWS identity behind the secret must also be allowed "
+			                                    "'rds-db:connect' privileges as that user.",
+			                                    user, user);
+		}
+		throw IOException(error_message);
+	}
+}
+
+unique_ptr<TransactionManager> RdsCreateTransactionManager(optional_ptr<StorageExtensionInfo> storage_info,
+                                                           AttachedDatabase &db, Catalog &catalog) {
+	// RdsAttach returned a PostgresCatalog, so the transaction manager has to come from the same
+	// place. Attach has already established that the postgres extension is loaded.
+	auto &db_config = DBConfig::GetConfig(db.GetDatabase());
+	auto postgres_extension = FindPostgresStorageExtension(db_config);
+	if (!postgres_extension || !postgres_extension->create_transaction_manager) {
+		throw InternalException("RDS attach: the postgres storage extension disappeared after attaching");
+	}
+	return postgres_extension->create_transaction_manager(postgres_extension->storage_info.get(), db, catalog);
+}
+
+class RdsStorageExtension : public StorageExtension {
+public:
+	RdsStorageExtension() {
+		attach = RdsAttach;
+		create_transaction_manager = RdsCreateTransactionManager;
+	}
+};
+
+} // namespace
+
+void Rds::RegisterStorageExtension(ExtensionLoader &loader) {
+	auto &config = DBConfig::GetConfig(loader.GetDatabaseInstance());
+	StorageExtension::Register(config, "rds", make_shared_ptr<RdsStorageExtension>());
+}
+
+} // namespace duckdb